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