1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one or more
3 * contributor license agreements. See the NOTICE file distributed with
4 * this work for additional information regarding copyright ownership.
5 * The ASF licenses this file to You under the Apache License, Version 2.0
6 * (the "License"); you may not use this file except in compliance with
7 * the License. You may obtain a copy of the License at
8 *
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17 package org.apache.commons.collections4.queue;
18
19 import java.io.IOException;
20 import java.io.InvalidObjectException;
21 import java.io.ObjectInputStream;
22 import java.io.ObjectOutputStream;
23 import java.io.Serializable;
24 import java.util.AbstractCollection;
25 import java.util.Arrays;
26 import java.util.Collection;
27 import java.util.Iterator;
28 import java.util.NoSuchElementException;
29 import java.util.Objects;
30 import java.util.Queue;
31
32 import org.apache.commons.collections4.BoundedCollection;
33
34 /**
35 * CircularFifoQueue is a first-in first-out queue with a fixed size that
36 * replaces its oldest element if full.
37 * <p>
38 * The removal order of a {@link CircularFifoQueue} is based on the
39 * insertion order; elements are removed in the same order in which they
40 * were added. The iteration order is the same as the removal order.
41 * </p>
42 * <p>
43 * The {@link #add(Object)}, {@link #remove()}, {@link #peek()}, {@link #poll()},
44 * {@link #offer(Object)} operations all perform in constant time.
45 * All other operations perform in linear time or worse.
46 * </p>
47 * <p>
48 * This queue prevents null objects from being added.
49 * </p>
50 *
51 * @param <E> The type of elements in this collection
52 * @since 4.0
53 */
54 public class CircularFifoQueue<E> extends AbstractCollection<E>
55 implements Queue<E>, BoundedCollection<E>, Serializable {
56
57 /** Serialization version. */
58 private static final long serialVersionUID = -8423413834657610406L;
59
60 /** Underlying storage array. */
61 private transient E[] elements;
62
63 /** Array index of first (oldest) queue element. */
64 private transient int start;
65
66 /**
67 * Index mod maxElements of the array position following the last queue
68 * element. Queue elements start at elements[start] and "wrap around"
69 * elements[maxElements-1], ending at elements[decrement(end)].
70 * For example, elements = {c,a,b}, start=1, end=1 corresponds to
71 * the queue [a,b,c].
72 */
73 private transient int end;
74
75 /** Flag to indicate if the queue is currently full. */
76 private transient boolean full;
77
78 /** Capacity of the queue. */
79 private final int maxElements;
80
81 /**
82 * Constructor that creates a queue with the default size of 32.
83 */
84 public CircularFifoQueue() {
85 this(32);
86 }
87
88 /**
89 * Constructor that creates a queue from the specified collection.
90 * The collection size also sets the queue size.
91 *
92 * @param coll The collection to copy into the queue, may not be null
93 * @throws NullPointerException if the collection is null
94 */
95 public CircularFifoQueue(final Collection<? extends E> coll) {
96 this(coll.size());
97 addAll(coll);
98 }
99
100 /**
101 * Constructor that creates a queue with the specified size.
102 *
103 * @param size The size of the queue (cannot be changed)
104 * @throws IllegalArgumentException if the size is < 1
105 */
106 @SuppressWarnings("unchecked")
107 public CircularFifoQueue(final int size) {
108 if (size <= 0) {
109 throw new IllegalArgumentException("The size must be greater than 0");
110 }
111 elements = (E[]) new Object[size];
112 maxElements = elements.length;
113 }
114
115 /**
116 * Adds the given element to this queue. If the queue is full, the least recently added
117 * element is discarded so that a new element can be inserted.
118 *
119 * @param element The element to add
120 * @return true, always
121 * @throws NullPointerException if the given element is null
122 */
123 @Override
124 public boolean add(final E element) {
125 Objects.requireNonNull(element, "element");
126
127 if (isAtFullCapacity()) {
128 remove();
129 }
130
131 elements[end++] = element;
132
133 if (end >= maxElements) {
134 end = 0;
135 }
136
137 if (end == start) {
138 full = true;
139 }
140
141 return true;
142 }
143
144 /**
145 * Clears this queue.
146 */
147 @Override
148 public void clear() {
149 full = false;
150 start = 0;
151 end = 0;
152 Arrays.fill(elements, null);
153 }
154
155 /**
156 * Decrements the internal index.
157 *
158 * @param index The index to decrement
159 * @return The updated index
160 */
161 private int decrement(int index) {
162 index--;
163 if (index < 0) {
164 index = maxElements - 1;
165 }
166 return index;
167 }
168
169 @Override
170 public E element() {
171 if (isEmpty()) {
172 throw new NoSuchElementException("queue is empty");
173 }
174 return peek();
175 }
176
177 /**
178 * Gets the element at the specified position in this queue.
179 *
180 * @param index The position of the element in the queue
181 * @return The element at position {@code index}
182 * @throws NoSuchElementException if the requested position is outside the range [0, size)
183 */
184 public E get(final int index) {
185 final int sz = size();
186 if (index < 0 || index >= sz) {
187 throw new NoSuchElementException(
188 String.format("The specified index %1$d is outside the available range [0, %2$d)",
189 Integer.valueOf(index), Integer.valueOf(sz)));
190 }
191
192 final int idx = (start + index) % maxElements;
193 return elements[idx];
194 }
195
196 /**
197 * Increments the internal index.
198 *
199 * @param index The index to increment
200 * @return The updated index
201 */
202 private int increment(int index) {
203 index++;
204 if (index >= maxElements) {
205 index = 0;
206 }
207 return index;
208 }
209
210 /**
211 * Returns {@code true} if the capacity limit of this queue has been reached,
212 * i.e. the number of elements stored in the queue equals its maximum size.
213 *
214 * @return {@code true} if the capacity limit has been reached, {@code false} otherwise
215 * @since 4.1
216 */
217 public boolean isAtFullCapacity() {
218 return size() == maxElements;
219 }
220
221 /**
222 * Returns true if this queue is empty; false otherwise.
223 *
224 * @return true if this queue is empty
225 */
226 @Override
227 public boolean isEmpty() {
228 return size() == 0;
229 }
230
231 /**
232 * {@inheritDoc}
233 * <p>
234 * A {@code CircularFifoQueue} can never be full, thus this returns always
235 * {@code false}.
236 *
237 * @return always returns {@code false}
238 */
239 @Override
240 public boolean isFull() {
241 return false;
242 }
243
244 /**
245 * Returns an iterator over this queue's elements.
246 *
247 * @return An iterator over this queue's elements
248 */
249 @Override
250 public Iterator<E> iterator() {
251 return new Iterator<E>() {
252
253 private int index = start;
254 private int lastReturnedIndex = -1;
255 private boolean isFirst = full;
256
257 @Override
258 public boolean hasNext() {
259 return isFirst || index != end;
260 }
261
262 @Override
263 public E next() {
264 if (!hasNext()) {
265 throw new NoSuchElementException();
266 }
267 isFirst = false;
268 lastReturnedIndex = index;
269 index = increment(index);
270 return elements[lastReturnedIndex];
271 }
272
273 @Override
274 public void remove() {
275 if (lastReturnedIndex == -1) {
276 throw new IllegalStateException();
277 }
278
279 // First element can be removed quickly
280 if (lastReturnedIndex == start) {
281 CircularFifoQueue.this.remove();
282 lastReturnedIndex = -1;
283 return;
284 }
285
286 int pos = lastReturnedIndex + 1;
287 if (start < lastReturnedIndex && pos < end) {
288 // shift in one part
289 System.arraycopy(elements, pos, elements, lastReturnedIndex, end - pos);
290 } else {
291 // Other elements require us to shift the subsequent elements
292 while (pos != end) {
293 if (pos >= maxElements) {
294 elements[pos - 1] = elements[0];
295 pos = 0;
296 } else {
297 elements[decrement(pos)] = elements[pos];
298 pos = increment(pos);
299 }
300 }
301 }
302
303 lastReturnedIndex = -1;
304 end = decrement(end);
305 elements[end] = null;
306 full = false;
307 index = decrement(index);
308 }
309
310 };
311 }
312
313 /**
314 * Gets the maximum size of the collection (the bound).
315 *
316 * @return The maximum number of elements the collection can hold
317 */
318 @Override
319 public int maxSize() {
320 return maxElements;
321 }
322
323 /**
324 * Adds the given element to this queue. If the queue is full, the least recently added
325 * element is discarded so that a new element can be inserted.
326 *
327 * @param element The element to add
328 * @return true, always
329 * @throws NullPointerException if the given element is null
330 */
331 @Override
332 public boolean offer(final E element) {
333 return add(element);
334 }
335
336 @Override
337 public E peek() {
338 if (isEmpty()) {
339 return null;
340 }
341 return elements[start];
342 }
343
344 @Override
345 public E poll() {
346 if (isEmpty()) {
347 return null;
348 }
349 return remove();
350 }
351
352 /**
353 * Deserializes the queue in using a custom routine.
354 *
355 * @param in The input stream
356 * @throws IOException Thrown if an I/O error occurs while writing to the output stream
357 * @throws ClassNotFoundException if the class of a serialized object cannot be found
358 */
359 @SuppressWarnings("unchecked")
360 private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
361 in.defaultReadObject();
362 if (maxElements < 1) {
363 throw new InvalidObjectException("maxElements must be greater than 0");
364 }
365 elements = (E[]) new Object[maxElements];
366 final int size = in.readInt();
367 if (size < 0 || size > maxElements) {
368 throw new InvalidObjectException("size is out of range: " + size);
369 }
370 for (int i = 0; i < size; i++) {
371 elements[i] = (E) in.readObject();
372 }
373 start = 0;
374 full = size == maxElements;
375 if (full) {
376 end = 0;
377 } else {
378 end = size;
379 }
380 }
381
382 @Override
383 public E remove() {
384 if (isEmpty()) {
385 throw new NoSuchElementException("queue is empty");
386 }
387
388 final E element = elements[start];
389 if (element != null) {
390 elements[start++] = null;
391
392 if (start >= maxElements) {
393 start = 0;
394 }
395 full = false;
396 }
397 return element;
398 }
399
400 /**
401 * Returns the number of elements stored in the queue.
402 *
403 * @return this queue's size
404 */
405 @Override
406 public int size() {
407 int size = 0;
408
409 if (end < start) {
410 size = maxElements - start + end;
411 } else if (end == start) {
412 size = full ? maxElements : 0;
413 } else {
414 size = end - start;
415 }
416
417 return size;
418 }
419
420 /**
421 * Serializes this object to an ObjectOutputStream.
422 *
423 * @param out The target ObjectOutputStream.
424 * @throws IOException thrown when an I/O errors occur writing to the target stream.
425 */
426 private void writeObject(final ObjectOutputStream out) throws IOException {
427 out.defaultWriteObject();
428 out.writeInt(size());
429 for (final E e : this) {
430 out.writeObject(e);
431 }
432 }
433
434 }