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.
038 * Also see this 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
051   * and nonfair access 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,
071   * in the order 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>(Collections2.cast(elements));
078    }
079    ArrayDeque<E> deque = new ArrayDeque<E>();
080    Iterables.addAll(deque, elements);
081    return deque;
082  }
083
084  // ConcurrentLinkedQueue
085
086  /**
087   * Creates an empty {@code ConcurrentLinkedQueue}.
088   */
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,
096   * in 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>(Collections2.cast(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},
134   * containing the elements of the specified iterable,
135   * in the order they are returned by the iterable's 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>(Collections2.cast(elements));
143    }
144    LinkedBlockingDeque<E> deque = new LinkedBlockingDeque<E>();
145    Iterables.addAll(deque, elements);
146    return deque;
147  }
148
149  // LinkedBlockingQueue
150
151  /**
152   * Creates an empty {@code LinkedBlockingQueue} with a capacity of {@link Integer#MAX_VALUE}.
153   */
154  @GwtIncompatible // LinkedBlockingQueue
155  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue() {
156    return new LinkedBlockingQueue<E>();
157  }
158
159  /**
160   * Creates an empty {@code LinkedBlockingQueue} with the given (fixed) capacity.
161   *
162   * @throws IllegalArgumentException if {@code capacity} is less than 1
163   */
164  @GwtIncompatible // LinkedBlockingQueue
165  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue(int capacity) {
166    return new LinkedBlockingQueue<E>(capacity);
167  }
168
169  /**
170   * Creates a {@code LinkedBlockingQueue} with a capacity of {@link Integer#MAX_VALUE},
171   * containing the elements of the specified iterable,
172   * in the order they are returned by the iterable's iterator.
173   *
174   * @param elements the elements that the queue should contain, in order
175   * @return a new {@code LinkedBlockingQueue} containing those elements
176   */
177  @GwtIncompatible // LinkedBlockingQueue
178  public static <E> LinkedBlockingQueue<E> newLinkedBlockingQueue(Iterable<? extends E> elements) {
179    if (elements instanceof Collection) {
180      return new LinkedBlockingQueue<E>(Collections2.cast(elements));
181    }
182    LinkedBlockingQueue<E> queue = new LinkedBlockingQueue<E>();
183    Iterables.addAll(queue, elements);
184    return queue;
185  }
186
187  // LinkedList: see {@link com.google.common.collect.Lists}
188
189  // PriorityBlockingQueue
190
191  /**
192   * Creates an empty {@code PriorityBlockingQueue} with the ordering given by its
193   * elements' natural ordering.
194   *
195   * @since 11.0 (requires that {@code E} be {@code Comparable} since 15.0).
196   */
197  @GwtIncompatible // PriorityBlockingQueue
198  public static <E extends Comparable> PriorityBlockingQueue<E> newPriorityBlockingQueue() {
199    return new PriorityBlockingQueue<E>();
200  }
201
202  /**
203   * Creates a {@code PriorityBlockingQueue} containing the given elements.
204   *
205   * <b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
206   * this priority queue will be ordered according to the same ordering.
207   *
208   * @since 11.0 (requires that {@code E} be {@code Comparable} since 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>(Collections2.cast(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
225   * elements' natural ordering.
226   *
227   * @since 11.0 (requires that {@code E} be {@code Comparable} since 15.0).
228   */
229  public static <E extends Comparable> PriorityQueue<E> newPriorityQueue() {
230    return new PriorityQueue<E>();
231  }
232
233  /**
234   * Creates a {@code PriorityQueue} containing the given elements.
235   *
236   * <b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
237   * this priority queue will be ordered according to the same ordering.
238   *
239   * @since 11.0 (requires that {@code E} be {@code Comparable} since 15.0).
240   */
241  public static <E extends Comparable> PriorityQueue<E> newPriorityQueue(
242      Iterable<? extends E> elements) {
243    if (elements instanceof Collection) {
244      return new PriorityQueue<E>(Collections2.cast(elements));
245    }
246    PriorityQueue<E> queue = new PriorityQueue<E>();
247    Iterables.addAll(queue, elements);
248    return queue;
249  }
250
251  // SynchronousQueue
252
253  /**
254   * Creates an empty {@code SynchronousQueue} with nonfair access policy.
255   */
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
263   * {@code numElements} elements are not available, it will wait for them up to the specified
264   * timeout.
265   *
266   * @param q the blocking queue to be drained
267   * @param buffer where to add the transferred elements
268   * @param numElements the number of elements to be waited for
269   * @param timeout how long to wait before giving up, in units of {@code unit}
270   * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter
271   * @return the number of elements transferred
272   * @throws InterruptedException if interrupted while waiting
273   */
274  @Beta
275  @CanIgnoreReturnValue
276  @GwtIncompatible // BlockingQueue
277  public static <E> int drain(
278      BlockingQueue<E> q,
279      Collection<? super E> buffer,
280      int numElements,
281      long timeout,
282      TimeUnit unit)
283      throws InterruptedException {
284    Preconditions.checkNotNull(buffer);
285    /*
286     * This code performs one System.nanoTime() more than necessary, and in return, the time to
287     * execute Queue#drainTo is not added *on top* of waiting for the timeout (which could make
288     * the timeout arbitrarily inaccurate, given a queue that is slow to drain).
289     */
290    long deadline = System.nanoTime() + unit.toNanos(timeout);
291    int added = 0;
292    while (added < numElements) {
293      // we could rely solely on #poll, but #drainTo might be more efficient when there are multiple
294      // elements already available (e.g. LinkedBlockingQueue#drainTo locks only once)
295      added += q.drainTo(buffer, numElements - added);
296      if (added < numElements) { // not enough elements immediately available; will have to poll
297        E e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
298        if (e == null) {
299          break; // we already waited enough, and there are no more elements in sight
300        }
301        buffer.add(e);
302        added++;
303      }
304    }
305    return added;
306  }
307
308  /**
309   * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, long, TimeUnit)},
310   * but with a different behavior in case it is interrupted while waiting. In that case, the
311   * operation will continue as usual, and in the end the thread's interruption status will be set
312   * (no {@code InterruptedException} is thrown).
313   *
314   * @param q the blocking queue to be drained
315   * @param buffer where to add the transferred elements
316   * @param numElements the number of elements to be waited for
317   * @param timeout how long to wait before giving up, in units of {@code unit}
318   * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter
319   * @return the number of elements transferred
320   */
321  @Beta
322  @CanIgnoreReturnValue
323  @GwtIncompatible // BlockingQueue
324  public static <E> int drainUninterruptibly(
325      BlockingQueue<E> q,
326      Collection<? super E> buffer,
327      int numElements,
328      long timeout,
329      TimeUnit unit) {
330    Preconditions.checkNotNull(buffer);
331    long deadline = System.nanoTime() + unit.toNanos(timeout);
332    int added = 0;
333    boolean interrupted = false;
334    try {
335      while (added < numElements) {
336        // we could rely solely on #poll, but #drainTo might be more efficient when there are
337        // multiple 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; // written exactly once, by a successful (uninterrupted) invocation of #poll
341          while (true) {
342            try {
343              e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
344              break;
345            } catch (InterruptedException ex) {
346              interrupted = true; // note interruption and retry
347            }
348          }
349          if (e == null) {
350            break; // we already waited enough, and there are no more elements in sight
351          }
352          buffer.add(e);
353          added++;
354        }
355      }
356    } finally {
357      if (interrupted) {
358        Thread.currentThread().interrupt();
359      }
360    }
361    return added;
362  }
363
364  /**
365   * Returns a synchronized (thread-safe) queue backed by the specified queue. In order to
366   * guarantee serial access, it is critical that <b>all</b> access to the backing queue is
367   * accomplished through the returned queue.
368   *
369   * <p>It is imperative that the user manually synchronize on the returned queue when accessing
370   * the queue's iterator: <pre>   {@code
371   *
372   *   Queue<E> queue = Queues.synchronizedQueue(MinMaxPriorityQueue.<E>create());
373   *   ...
374   *   queue.add(element);  // Needn't be in synchronized block
375   *   ...
376   *   synchronized (queue) {  // Must synchronize on queue!
377   *     Iterator<E> i = queue.iterator(); // Must be in synchronized block
378   *     while (i.hasNext()) {
379   *       foo(i.next());
380   *     }
381   *   }}</pre>
382   *
383   * <p>Failure to follow this advice may result in non-deterministic behavior.
384   *
385   * <p>The returned queue will be serializable if the specified queue is serializable.
386   *
387   * @param queue the queue to be wrapped in a synchronized view
388   * @return a synchronized view of the specified queue
389   * @since 14.0
390   */
391  public static <E> Queue<E> synchronizedQueue(Queue<E> queue) {
392    return Synchronized.queue(queue, null);
393  }
394
395  /**
396   * Returns a synchronized (thread-safe) deque backed by the specified deque. In order to
397   * guarantee serial access, it is critical that <b>all</b> access to the backing deque is
398   * accomplished through the returned deque.
399   *
400   * <p>It is imperative that the user manually synchronize on the returned deque when accessing
401   * any of the deque's iterators: <pre>   {@code
402   *
403   *   Deque<E> deque = Queues.synchronizedDeque(Queues.<E>newArrayDeque());
404   *   ...
405   *   deque.add(element);  // Needn't be in synchronized block
406   *   ...
407   *   synchronized (deque) {  // Must synchronize on deque!
408   *     Iterator<E> i = deque.iterator(); // Must be in synchronized block
409   *     while (i.hasNext()) {
410   *       foo(i.next());
411   *     }
412   *   }}</pre>
413   *
414   * <p>Failure to follow this advice may result in non-deterministic behavior.
415   *
416   * <p>The returned deque will be serializable if the specified deque is serializable.
417   *
418   * @param deque the deque to be wrapped in a synchronized view
419   * @return a synchronized view of the specified deque
420   * @since 15.0
421   */
422  public static <E> Deque<E> synchronizedDeque(Deque<E> deque) {
423    return Synchronized.deque(deque, null);
424  }
425}