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