    001/*
    002 * Licensed to the Apache Software Foundation (ASF) under one or more
    003 * contributor license agreements.  See the NOTICE file distributed with
    004 * this work for additional information regarding copyright ownership.
    005 * The ASF licenses this file to You under the Apache License, Version 2.0
    006 * (the "License"); you may not use this file except in compliance with
    007 * the License.  You may obtain a copy of the License at
    008 *
    009 *      https://www.apache.org/licenses/LICENSE-2.0
    010 *
    011 * Unless required by applicable law or agreed to in writing, software
    012 * distributed under the License is distributed on an "AS IS" BASIS,
    013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    014 * See the License for the specific language governing permissions and
    015 * limitations under the License.
    016 */
    017package org.apache.commons.collections4.queue;
    018
    019import java.io.IOException;
    020import java.io.InvalidObjectException;
    021import java.io.ObjectInputStream;
    022import java.io.ObjectOutputStream;
    023import java.io.Serializable;
    024import java.util.AbstractCollection;
    025import java.util.Arrays;
    026import java.util.Collection;
    027import java.util.Iterator;
    028import java.util.NoSuchElementException;
    029import java.util.Objects;
    030import java.util.Queue;
    031
    032import org.apache.commons.collections4.BoundedCollection;
    033
    034/**
    035 * CircularFifoQueue is a first-in first-out queue with a fixed size that
    036 * replaces its oldest element if full.
    037 * <p>
    038 * The removal order of a {@link CircularFifoQueue} is based on the
    039 * insertion order; elements are removed in the same order in which they
    040 * were added.  The iteration order is the same as the removal order.
    041 * </p>
    042 * <p>
    043 * The {@link #add(Object)}, {@link #remove()}, {@link #peek()}, {@link #poll()},
    044 * {@link #offer(Object)} operations all perform in constant time.
    045 * All other operations perform in linear time or worse.
    046 * </p>
    047 * <p>
    048 * This queue prevents null objects from being added.
    049 * </p>
    050 *
    051 * @param <E> The type of elements in this collection
    052 * @since 4.0
    053 */
    054public class CircularFifoQueue<E> extends AbstractCollection<E>
    055    implements Queue<E>, BoundedCollection<E>, Serializable {
    056
    057    /** Serialization version. */
    058    private static final long serialVersionUID = -8423413834657610406L;
    059
    060    /** Underlying storage array. */
    061    private transient E[] elements;
    062
    063    /** Array index of first (oldest) queue element. */
    064    private transient int start;
    065
    066    /**
    067     * Index mod maxElements of the array position following the last queue
    068     * element.  Queue elements start at elements[start] and "wrap around"
    069     * elements[maxElements-1], ending at elements[decrement(end)].
    070     * For example, elements = {c,a,b}, start=1, end=1 corresponds to
    071     * the queue [a,b,c].
    072     */
    073    private transient int end;
    074
    075    /** Flag to indicate if the queue is currently full. */
    076    private transient boolean full;
    077
    078    /** Capacity of the queue. */
    079    private final int maxElements;
    080
    081    /**
    082     * Constructor that creates a queue with the default size of 32.
    083     */
    084    public CircularFifoQueue() {
    085        this(32);
    086    }
    087
    088    /**
    089     * Constructor that creates a queue from the specified collection.
    090     * The collection size also sets the queue size.
    091     *
    092     * @param coll  The collection to copy into the queue, may not be null
    093     * @throws NullPointerException if the collection is null
    094     */
    095    public CircularFifoQueue(final Collection<? extends E> coll) {
    096        this(coll.size());
    097        addAll(coll);
    098    }
    099
    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 &lt; 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}