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