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, in units of {@code unit} 269 * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter 270 * @return the number of elements transferred 271 * @throws InterruptedException if interrupted while waiting 272 */ 273 @Beta 274 @CanIgnoreReturnValue 275 @GwtIncompatible // BlockingQueue 276 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 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)}, but 310 * with a different behavior in case it is interrupted while waiting. In that case, the operation 311 * will continue as usual, and in the end the thread's interruption status will be set (no {@code 312 * 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 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 325 public static <E> int drainUninterruptibly( 326 BlockingQueue<E> q, 327 Collection<? super E> buffer, 328 int numElements, 329 long timeout, 330 TimeUnit unit) { 331 Preconditions.checkNotNull(buffer); 332 long deadline = System.nanoTime() + unit.toNanos(timeout); 333 int added = 0; 334 boolean interrupted = false; 335 try { 336 while (added < numElements) { 337 // we could rely solely on #poll, but #drainTo might be more efficient when there are 338 // multiple elements already available (e.g. LinkedBlockingQueue#drainTo locks only once) 339 added += q.drainTo(buffer, numElements - added); 340 if (added < numElements) { // not enough elements immediately available; will have to poll 341 E e; // written exactly once, by a successful (uninterrupted) invocation of #poll 342 while (true) { 343 try { 344 e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS); 345 break; 346 } catch (InterruptedException ex) { 347 interrupted = true; // note interruption and retry 348 } 349 } 350 if (e == null) { 351 break; // we already waited enough, and there are no more elements in sight 352 } 353 buffer.add(e); 354 added++; 355 } 356 } 357 } finally { 358 if (interrupted) { 359 Thread.currentThread().interrupt(); 360 } 361 } 362 return added; 363 } 364 365 /** 366 * Returns a synchronized (thread-safe) queue backed by the specified queue. In order to guarantee 367 * serial access, it is critical that <b>all</b> access to the backing queue is accomplished 368 * through the returned queue. 369 * 370 * <p>It is imperative that the user manually synchronize on the returned queue when accessing the 371 * queue's iterator: 372 * 373 * <pre>{@code 374 * Queue<E> queue = Queues.synchronizedQueue(MinMaxPriorityQueue.<E>create()); 375 * ... 376 * queue.add(element); // Needn't be in synchronized block 377 * ... 378 * synchronized (queue) { // Must synchronize on queue! 379 * Iterator<E> i = queue.iterator(); // Must be in synchronized block 380 * while (i.hasNext()) { 381 * foo(i.next()); 382 * } 383 * } 384 * }</pre> 385 * 386 * <p>Failure to follow this advice may result in non-deterministic behavior. 387 * 388 * <p>The returned queue will be serializable if the specified queue is serializable. 389 * 390 * @param queue the queue to be wrapped in a synchronized view 391 * @return a synchronized view of the specified queue 392 * @since 14.0 393 */ 394 public static <E extends @Nullable Object> Queue<E> synchronizedQueue(Queue<E> queue) { 395 return Synchronized.queue(queue, null); 396 } 397 398 /** 399 * Returns a synchronized (thread-safe) deque backed by the specified deque. In order to guarantee 400 * serial access, it is critical that <b>all</b> access to the backing deque is accomplished 401 * through the returned deque. 402 * 403 * <p>It is imperative that the user manually synchronize on the returned deque when accessing any 404 * of the deque's iterators: 405 * 406 * <pre>{@code 407 * Deque<E> deque = Queues.synchronizedDeque(Queues.<E>newArrayDeque()); 408 * ... 409 * deque.add(element); // Needn't be in synchronized block 410 * ... 411 * synchronized (deque) { // Must synchronize on deque! 412 * Iterator<E> i = deque.iterator(); // Must be in synchronized block 413 * while (i.hasNext()) { 414 * foo(i.next()); 415 * } 416 * } 417 * }</pre> 418 * 419 * <p>Failure to follow this advice may result in non-deterministic behavior. 420 * 421 * <p>The returned deque will be serializable if the specified deque is serializable. 422 * 423 * @param deque the deque to be wrapped in a synchronized view 424 * @return a synchronized view of the specified deque 425 * @since 15.0 426 */ 427 public static <E extends @Nullable Object> Deque<E> synchronizedDeque(Deque<E> deque) { 428 return Synchronized.deque(deque, null); 429 } 430}