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>(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 /** 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>(Collections2.cast(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>(Collections2.cast(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>(Collections2.cast(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>(Collections2.cast(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>(Collections2.cast(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}