Source code

001/*
002 * Copyright (C) 2011 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
005 * in compliance with the License. You may obtain a copy of the License at
006 *
007 * http://www.apache.org/licenses/LICENSE-2.0
008 *
009 * Unless required by applicable law or agreed to in writing, software distributed under the License
010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
011 * or implied. See the License for the specific language governing permissions and limitations under
012 * the License.
013 */
014
015package com.google.common.collect;
016
017import com.google.common.annotations.GwtCompatible;
018import com.google.common.annotations.GwtIncompatible;
019import com.google.common.annotations.J2ktIncompatible;
020import com.google.common.base.Preconditions;
021import com.google.errorprone.annotations.CanIgnoreReturnValue;
022import java.util.ArrayDeque;
023import java.util.Collection;
024import java.util.Deque;
025import java.util.PriorityQueue;
026import java.util.Queue;
027import java.util.concurrent.ArrayBlockingQueue;
028import java.util.concurrent.BlockingQueue;
029import java.util.concurrent.ConcurrentLinkedQueue;
030import java.util.concurrent.LinkedBlockingDeque;
031import java.util.concurrent.LinkedBlockingQueue;
032import java.util.concurrent.PriorityBlockingQueue;
033import java.util.concurrent.SynchronousQueue;
034import java.util.concurrent.TimeUnit;
035import org.checkerframework.checker.nullness.qual.Nullable;
036
037/**
038 * Static utility methods pertaining to {@link Queue} and {@link Deque} instances. Also see this
039 * class's counterparts {@link Lists}, {@link Sets}, and {@link Maps}.
040 *
041 * @author Kurt Alfred Kluever
042 * @since 11.0
043 */
044@GwtCompatible(emulated = true)
045@ElementTypesAreNonnullByDefault
046public final class Queues {
047  private Queues() {}
048
049  // ArrayBlockingQueue
050
051  /**
052   * Creates an empty {@code ArrayBlockingQueue} with the given (fixed) capacity and nonfair access
053   * policy.
054   */
055  @J2ktIncompatible
056  @GwtIncompatible // ArrayBlockingQueue
057  public static <E> ArrayBlockingQueue<E> newArrayBlockingQueue(int capacity) {
058    return new ArrayBlockingQueue<E>(capacity);
059  }
060
061  // ArrayDeque
062
063  /**
064   * Creates an empty {@code ArrayDeque}.
065   *
066   * @since 12.0
067   */
068  public static <E> ArrayDeque<E> newArrayDeque() {
069    return new ArrayDeque<E>();
070  }
071
072  /**
073   * Creates an {@code ArrayDeque} containing the elements of the specified iterable, in the order
074   * they are returned by the iterable's iterator.
075   *
076   * @since 12.0
077   */
078  public static <E> ArrayDeque<E> newArrayDeque(Iterable<? extends E> elements) {
079    if (elements instanceof Collection) {
080      return new ArrayDeque<E>((Collection<? extends E>) elements);
081    }
082    ArrayDeque<E> deque = new ArrayDeque<E>();
083    Iterables.addAll(deque, elements);
084    return deque;
085  }
086
087  // ConcurrentLinkedQueue
088
089  /** Creates an empty {@code ConcurrentLinkedQueue}. */
090  @J2ktIncompatible
091  @GwtIncompatible // ConcurrentLinkedQueue
092  public static <E> ConcurrentLinkedQueue<E> newConcurrentLinkedQueue() {
093    return new ConcurrentLinkedQueue<E>();
094  }
095
096  /**
097   * Creates a {@code ConcurrentLinkedQueue} containing the elements of the specified iterable, in
098   * the order they are returned by the iterable's iterator.
099   */
100  @J2ktIncompatible
101  @GwtIncompatible // ConcurrentLinkedQueue
102  public static <E> ConcurrentLinkedQueue<E> newConcurrentLinkedQueue(
103      Iterable<? extends E> elements) {
104    if (elements instanceof Collection) {
105      return new ConcurrentLinkedQueue<E>((Collection<? extends E>) elements);
106    }
107    ConcurrentLinkedQueue<E> queue = new ConcurrentLinkedQueue<E>();
108    Iterables.addAll(queue, elements);
109    return queue;
110  }
111
112  // LinkedBlockingDeque
113
114  /**
115   * Creates an empty {@code LinkedBlockingDeque} with a capacity of {@link Integer#MAX_VALUE}.
116   *
117   * @since 12.0
118   */
119  @J2ktIncompatible
120  @GwtIncompatible // LinkedBlockingDeque
121  public static <E> LinkedBlockingDeque<E> newLinkedBlockingDeque() {
122    return new LinkedBlockingDeque<E>();
123  }
124
125  /**
126   * Creates an empty {@code LinkedBlockingDeque} with the given (fixed) capacity.
127   *
128   * @throws IllegalArgumentException if {@code capacity} is less than 1
129   * @since 12.0
130   */
131  @J2ktIncompatible
132  @GwtIncompatible // LinkedBlockingDeque
133  public static <E> LinkedBlockingDeque<E> newLinkedBlockingDeque(int capacity) {
134    return new LinkedBlockingDeque<E>(capacity);
135  }
136
137  /**
138   * Creates a {@code LinkedBlockingDeque} with a capacity of {@link Integer#MAX_VALUE}, containing
139   * the elements of the specified iterable, in the order they are returned by the iterable's
140   * iterator.
141   *
142   * @since 12.0
143   */
144  @J2ktIncompatible
145  @GwtIncompatible // LinkedBlockingDeque
146  public static <E> LinkedBlockingDeque<E> newLinkedBlockingDeque(Iterable<? extends E> elements) {
147    if (elements instanceof Collection) {
148      return new LinkedBlockingDeque<E>((Collection<? extends E>) elements);
149    }
150    LinkedBlockingDeque<E> deque = new LinkedBlockingDeque<E>();
151    Iterables.addAll(deque, elements);
152    return deque;
153  }
154
155  // LinkedBlockingQueue
156
157  /** Creates an empty {@code LinkedBlockingQueue} with a capacity of {@link Integer#MAX_VALUE}. */
158  @J2ktIncompatible
159  @GwtIncompatible // LinkedBlockingQueue
160  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue() {
161    return new LinkedBlockingQueue<E>();
162  }
163
164  /**
165   * Creates an empty {@code LinkedBlockingQueue} with the given (fixed) capacity.
166   *
167   * @throws IllegalArgumentException if {@code capacity} is less than 1
168   */
169  @J2ktIncompatible
170  @GwtIncompatible // LinkedBlockingQueue
171  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue(int capacity) {
172    return new LinkedBlockingQueue<E>(capacity);
173  }
174
175  /**
176   * Creates a {@code LinkedBlockingQueue} with a capacity of {@link Integer#MAX_VALUE}, containing
177   * the elements of the specified iterable, in the order they are returned by the iterable's
178   * iterator.
179   *
180   * @param elements the elements that the queue should contain, in order
181   * @return a new {@code LinkedBlockingQueue} containing those elements
182   */
183  @J2ktIncompatible
184  @GwtIncompatible // LinkedBlockingQueue
185  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue(Iterable<? extends E> elements) {
186    if (elements instanceof Collection) {
187      return new LinkedBlockingQueue<E>((Collection<? extends E>) elements);
188    }
189    LinkedBlockingQueue<E> queue = new LinkedBlockingQueue<E>();
190    Iterables.addAll(queue, elements);
191    return queue;
192  }
193
194  // LinkedList: see {@link com.google.common.collect.Lists}
195
196  // PriorityBlockingQueue
197
198  /**
199   * Creates an empty {@code PriorityBlockingQueue} with the ordering given by its elements' natural
200   * ordering.
201   *
202   * @since 11.0 (but the bound of {@code E} was changed from {@code Object} to {@code Comparable}
203   *     in 15.0)
204   */
205  @SuppressWarnings("rawtypes") // https://github.com/google/guava/issues/989
206  @J2ktIncompatible
207  @GwtIncompatible // PriorityBlockingQueue
208  public static <E extends Comparable> PriorityBlockingQueue<E> newPriorityBlockingQueue() {
209    return new PriorityBlockingQueue<E>();
210  }
211
212  /**
213   * Creates a {@code PriorityBlockingQueue} containing the given elements.
214   *
215   * <p><b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
216   * this priority queue will be ordered according to the same ordering.
217   *
218   * @since 11.0 (but the bound of {@code E} was changed from {@code Object} to {@code Comparable}
219   *     in 15.0)
220   */
221  @SuppressWarnings("rawtypes") // https://github.com/google/guava/issues/989
222  @J2ktIncompatible
223  @GwtIncompatible // PriorityBlockingQueue
224  public static <E extends Comparable> PriorityBlockingQueue<E> newPriorityBlockingQueue(
225      Iterable<? extends E> elements) {
226    if (elements instanceof Collection) {
227      return new PriorityBlockingQueue<E>((Collection<? extends E>) elements);
228    }
229    PriorityBlockingQueue<E> queue = new PriorityBlockingQueue<E>();
230    Iterables.addAll(queue, elements);
231    return queue;
232  }
233
234  // PriorityQueue
235
236  /**
237   * Creates an empty {@code PriorityQueue} with the ordering given by its elements' natural
238   * ordering.
239   *
240   * @since 11.0 (but the bound of {@code E} was changed from {@code Object} to {@code Comparable}
241   *     in 15.0)
242   */
243  @SuppressWarnings("rawtypes") // https://github.com/google/guava/issues/989
244  public static <E extends Comparable> PriorityQueue<E> newPriorityQueue() {
245    return new PriorityQueue<E>();
246  }
247
248  /**
249   * Creates a {@code PriorityQueue} containing the given elements.
250   *
251   * <p><b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
252   * this priority queue will be ordered according to the same ordering.
253   *
254   * @since 11.0 (but the bound of {@code E} was changed from {@code Object} to {@code Comparable}
255   *     in 15.0)
256   */
257  @SuppressWarnings("rawtypes") // https://github.com/google/guava/issues/989
258  public static <E extends Comparable> PriorityQueue<E> newPriorityQueue(
259      Iterable<? extends E> elements) {
260    if (elements instanceof Collection) {
261      return new PriorityQueue<E>((Collection<? extends E>) elements);
262    }
263    PriorityQueue<E> queue = new PriorityQueue<E>();
264    Iterables.addAll(queue, elements);
265    return queue;
266  }
267
268  // SynchronousQueue
269
270  /** Creates an empty {@code SynchronousQueue} with nonfair access policy. */
271  @J2ktIncompatible
272  @GwtIncompatible // SynchronousQueue
273  public static <E> SynchronousQueue<E> newSynchronousQueue() {
274    return new SynchronousQueue<E>();
275  }
276
277  /**
278   * Drains the queue as {@link BlockingQueue#drainTo(Collection, int)}, but if the requested {@code
279   * numElements} elements are not available, it will wait for them up to the specified timeout.
280   *
281   * @param q the blocking queue to be drained
282   * @param buffer where to add the transferred elements
283   * @param numElements the number of elements to be waited for
284   * @param timeout how long to wait before giving up
285   * @return the number of elements transferred
286   * @throws InterruptedException if interrupted while waiting
287   * @since 28.0
288   */
289  @CanIgnoreReturnValue
290  @J2ktIncompatible
291  @GwtIncompatible // BlockingQueue
292  public static <E> int drain(
293      BlockingQueue<E> q, Collection<? super E> buffer, int numElements, java.time.Duration timeout)
294      throws InterruptedException {
295    // TODO(b/126049426): Consider using saturateToNanos(timeout) instead.
296    return drain(q, buffer, numElements, timeout.toNanos(), TimeUnit.NANOSECONDS);
297  }
298
299  /**
300   * Drains the queue as {@link BlockingQueue#drainTo(Collection, int)}, but if the requested {@code
301   * numElements} elements are not available, it will wait for them up to the specified timeout.
302   *
303   * @param q the blocking queue to be drained
304   * @param buffer where to add the transferred elements
305   * @param numElements the number of elements to be waited for
306   * @param timeout how long to wait before giving up, in units of {@code unit}
307   * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter
308   * @return the number of elements transferred
309   * @throws InterruptedException if interrupted while waiting
310   */
311  @CanIgnoreReturnValue
312  @J2ktIncompatible
313  @GwtIncompatible // BlockingQueue
314  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
315  public static <E> int drain(
316      BlockingQueue<E> q,
317      Collection<? super E> buffer,
318      int numElements,
319      long timeout,
320      TimeUnit unit)
321      throws InterruptedException {
322    Preconditions.checkNotNull(buffer);
323    /*
324     * This code performs one System.nanoTime() more than necessary, and in return, the time to
325     * execute Queue#drainTo is not added *on top* of waiting for the timeout (which could make
326     * the timeout arbitrarily inaccurate, given a queue that is slow to drain).
327     */
328    long deadline = System.nanoTime() + unit.toNanos(timeout);
329    int added = 0;
330    while (added < numElements) {
331      // we could rely solely on #poll, but #drainTo might be more efficient when there are multiple
332      // elements already available (e.g. LinkedBlockingQueue#drainTo locks only once)
333      added += q.drainTo(buffer, numElements - added);
334      if (added < numElements) { // not enough elements immediately available; will have to poll
335        E e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
336        if (e == null) {
337          break; // we already waited enough, and there are no more elements in sight
338        }
339        buffer.add(e);
340        added++;
341      }
342    }
343    return added;
344  }
345
346  /**
347   * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, Duration)}, but with a
348   * different behavior in case it is interrupted while waiting. In that case, the operation will
349   * continue as usual, and in the end the thread's interruption status will be set (no {@code
350   * InterruptedException} is thrown).
351   *
352   * @param q the blocking queue to be drained
353   * @param buffer where to add the transferred elements
354   * @param numElements the number of elements to be waited for
355   * @param timeout how long to wait before giving up
356   * @return the number of elements transferred
357   * @since 28.0
358   */
359  @CanIgnoreReturnValue
360  @J2ktIncompatible
361  @GwtIncompatible // BlockingQueue
362  public static <E> int drainUninterruptibly(
363      BlockingQueue<E> q,
364      Collection<? super E> buffer,
365      int numElements,
366      java.time.Duration timeout) {
367    // TODO(b/126049426): Consider using saturateToNanos(timeout) instead.
368    return drainUninterruptibly(q, buffer, numElements, timeout.toNanos(), TimeUnit.NANOSECONDS);
369  }
370
371  /**
372   * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, long, TimeUnit)}, but
373   * with a different behavior in case it is interrupted while waiting. In that case, the operation
374   * will continue as usual, and in the end the thread's interruption status will be set (no {@code
375   * InterruptedException} is thrown).
376   *
377   * @param q the blocking queue to be drained
378   * @param buffer where to add the transferred elements
379   * @param numElements the number of elements to be waited for
380   * @param timeout how long to wait before giving up, in units of {@code unit}
381   * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter
382   * @return the number of elements transferred
383   */
384  @CanIgnoreReturnValue
385  @J2ktIncompatible
386  @GwtIncompatible // BlockingQueue
387  @SuppressWarnings("GoodTime") // should accept a java.time.Duration
388  public static <E> int drainUninterruptibly(
389      BlockingQueue<E> q,
390      Collection<? super E> buffer,
391      int numElements,
392      long timeout,
393      TimeUnit unit) {
394    Preconditions.checkNotNull(buffer);
395    long deadline = System.nanoTime() + unit.toNanos(timeout);
396    int added = 0;
397    boolean interrupted = false;
398    try {
399      while (added < numElements) {
400        // we could rely solely on #poll, but #drainTo might be more efficient when there are
401        // multiple elements already available (e.g. LinkedBlockingQueue#drainTo locks only once)
402        added += q.drainTo(buffer, numElements - added);
403        if (added < numElements) { // not enough elements immediately available; will have to poll
404          E e; // written exactly once, by a successful (uninterrupted) invocation of #poll
405          while (true) {
406            try {
407              e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
408              break;
409            } catch (InterruptedException ex) {
410              interrupted = true; // note interruption and retry
411            }
412          }
413          if (e == null) {
414            break; // we already waited enough, and there are no more elements in sight
415          }
416          buffer.add(e);
417          added++;
418        }
419      }
420    } finally {
421      if (interrupted) {
422        Thread.currentThread().interrupt();
423      }
424    }
425    return added;
426  }
427
428  /**
429   * Returns a synchronized (thread-safe) queue backed by the specified queue. In order to guarantee
430   * serial access, it is critical that <b>all</b> access to the backing queue is accomplished
431   * through the returned queue.
432   *
433   * <p>It is imperative that the user manually synchronize on the returned queue when accessing the
434   * queue's iterator:
435   *
436   * <pre>{@code
437   * Queue<E> queue = Queues.synchronizedQueue(MinMaxPriorityQueue.<E>create());
438   * ...
439   * queue.add(element);  // Needn't be in synchronized block
440   * ...
441   * synchronized (queue) {  // Must synchronize on queue!
442   *   Iterator<E> i = queue.iterator(); // Must be in synchronized block
443   *   while (i.hasNext()) {
444   *     foo(i.next());
445   *   }
446   * }
447   * }</pre>
448   *
449   * <p>Failure to follow this advice may result in non-deterministic behavior.
450   *
451   * <p>The returned queue will be serializable if the specified queue is serializable.
452   *
453   * @param queue the queue to be wrapped in a synchronized view
454   * @return a synchronized view of the specified queue
455   * @since 14.0
456   */
457  public static <E extends @Nullable Object> Queue<E> synchronizedQueue(Queue<E> queue) {
458    return Synchronized.queue(queue, null);
459  }
460
461  /**
462   * Returns a synchronized (thread-safe) deque backed by the specified deque. In order to guarantee
463   * serial access, it is critical that <b>all</b> access to the backing deque is accomplished
464   * through the returned deque.
465   *
466   * <p>It is imperative that the user manually synchronize on the returned deque when accessing any
467   * of the deque's iterators:
468   *
469   * <pre>{@code
470   * Deque<E> deque = Queues.synchronizedDeque(Queues.<E>newArrayDeque());
471   * ...
472   * deque.add(element);  // Needn't be in synchronized block
473   * ...
474   * synchronized (deque) {  // Must synchronize on deque!
475   *   Iterator<E> i = deque.iterator(); // Must be in synchronized block
476   *   while (i.hasNext()) {
477   *     foo(i.next());
478   *   }
479   * }
480   * }</pre>
481   *
482   * <p>Failure to follow this advice may result in non-deterministic behavior.
483   *
484   * <p>The returned deque will be serializable if the specified deque is serializable.
485   *
486   * @param deque the deque to be wrapped in a synchronized view
487   * @return a synchronized view of the specified deque
488   * @since 15.0
489   */
490  public static <E extends @Nullable Object> Deque<E> synchronizedDeque(Deque<E> deque) {
491    return Synchronized.deque(deque, null);
492  }
493}