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