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