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, in units of {@code unit} 263 * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter 264 * @return the number of elements transferred 265 * @throws InterruptedException if interrupted while waiting 266 */ 267 @Beta 268 @CanIgnoreReturnValue 269 @GwtIncompatible // BlockingQueue 270 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 271 public static <E> int drain( 272 BlockingQueue<E> q, 273 Collection<? super E> buffer, 274 int numElements, 275 long timeout, 276 TimeUnit unit) 277 throws InterruptedException { 278 Preconditions.checkNotNull(buffer); 279 /* 280 * This code performs one System.nanoTime() more than necessary, and in return, the time to 281 * execute Queue#drainTo is not added *on top* of waiting for the timeout (which could make 282 * the timeout arbitrarily inaccurate, given a queue that is slow to drain). 283 */ 284 long deadline = System.nanoTime() + unit.toNanos(timeout); 285 int added = 0; 286 while (added < numElements) { 287 // we could rely solely on #poll, but #drainTo might be more efficient when there are multiple 288 // elements already available (e.g. LinkedBlockingQueue#drainTo locks only once) 289 added += q.drainTo(buffer, numElements - added); 290 if (added < numElements) { // not enough elements immediately available; will have to poll 291 E e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS); 292 if (e == null) { 293 break; // we already waited enough, and there are no more elements in sight 294 } 295 buffer.add(e); 296 added++; 297 } 298 } 299 return added; 300 } 301 302 /** 303 * Drains the queue as {@linkplain #drain(BlockingQueue, Collection, int, long, TimeUnit)}, but 304 * with a different behavior in case it is interrupted while waiting. In that case, the operation 305 * will continue as usual, and in the end the thread's interruption status will be set (no {@code 306 * InterruptedException} is thrown). 307 * 308 * @param q the blocking queue to be drained 309 * @param buffer where to add the transferred elements 310 * @param numElements the number of elements to be waited for 311 * @param timeout how long to wait before giving up, in units of {@code unit} 312 * @param unit a {@code TimeUnit} determining how to interpret the timeout parameter 313 * @return the number of elements transferred 314 */ 315 @Beta 316 @CanIgnoreReturnValue 317 @GwtIncompatible // BlockingQueue 318 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 319 public static <E> int drainUninterruptibly( 320 BlockingQueue<E> q, 321 Collection<? super E> buffer, 322 int numElements, 323 long timeout, 324 TimeUnit unit) { 325 Preconditions.checkNotNull(buffer); 326 long deadline = System.nanoTime() + unit.toNanos(timeout); 327 int added = 0; 328 boolean interrupted = false; 329 try { 330 while (added < numElements) { 331 // we could rely solely on #poll, but #drainTo might be more efficient when there are 332 // multiple elements already available (e.g. LinkedBlockingQueue#drainTo locks only once) 333 added += q.drainTo(buffer, numElements - added); 334 if (added < numElements) { // not enough elements immediately available; will have to poll 335 E e; // written exactly once, by a successful (uninterrupted) invocation of #poll 336 while (true) { 337 try { 338 e = q.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS); 339 break; 340 } catch (InterruptedException ex) { 341 interrupted = true; // note interruption and retry 342 } 343 } 344 if (e == null) { 345 break; // we already waited enough, and there are no more elements in sight 346 } 347 buffer.add(e); 348 added++; 349 } 350 } 351 } finally { 352 if (interrupted) { 353 Thread.currentThread().interrupt(); 354 } 355 } 356 return added; 357 } 358 359 /** 360 * Returns a synchronized (thread-safe) queue backed by the specified queue. In order to guarantee 361 * serial access, it is critical that <b>all</b> access to the backing queue is accomplished 362 * through the returned queue. 363 * 364 * <p>It is imperative that the user manually synchronize on the returned queue when accessing the 365 * queue's iterator: 366 * 367 * <pre>{@code 368 * Queue<E> queue = Queues.synchronizedQueue(MinMaxPriorityQueue.<E>create()); 369 * ... 370 * queue.add(element); // Needn't be in synchronized block 371 * ... 372 * synchronized (queue) { // Must synchronize on queue! 373 * Iterator<E> i = queue.iterator(); // Must be in synchronized block 374 * while (i.hasNext()) { 375 * foo(i.next()); 376 * } 377 * } 378 * }</pre> 379 * 380 * <p>Failure to follow this advice may result in non-deterministic behavior. 381 * 382 * <p>The returned queue will be serializable if the specified queue is serializable. 383 * 384 * @param queue the queue to be wrapped in a synchronized view 385 * @return a synchronized view of the specified queue 386 * @since 14.0 387 */ 388 public static <E> Queue<E> synchronizedQueue(Queue<E> queue) { 389 return Synchronized.queue(queue, null); 390 } 391 392 /** 393 * Returns a synchronized (thread-safe) deque backed by the specified deque. In order to guarantee 394 * serial access, it is critical that <b>all</b> access to the backing deque is accomplished 395 * through the returned deque. 396 * 397 * <p>It is imperative that the user manually synchronize on the returned deque when accessing any 398 * of the deque's iterators: 399 * 400 * <pre>{@code 401 * Deque<E> deque = Queues.synchronizedDeque(Queues.<E>newArrayDeque()); 402 * ... 403 * deque.add(element); // Needn't be in synchronized block 404 * ... 405 * synchronized (deque) { // Must synchronize on deque! 406 * Iterator<E> i = deque.iterator(); // Must be in synchronized block 407 * while (i.hasNext()) { 408 * foo(i.next()); 409 * } 410 * } 411 * }</pre> 412 * 413 * <p>Failure to follow this advice may result in non-deterministic behavior. 414 * 415 * <p>The returned deque will be serializable if the specified deque is serializable. 416 * 417 * @param deque the deque to be wrapped in a synchronized view 418 * @return a synchronized view of the specified deque 419 * @since 15.0 420 */ 421 public static <E> Deque<E> synchronizedDeque(Deque<E> deque) { 422 return Synchronized.deque(deque, null); 423 } 424}