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