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