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