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}