    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 *      http://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.ObjectInputStream;
    021import java.io.ObjectOutputStream;
    022import java.io.Serializable;
    023import java.util.AbstractCollection;
    024import java.util.Arrays;
    025import java.util.Collection;
    026import java.util.Iterator;
    027import java.util.NoSuchElementException;
    028import java.util.Queue;
    029
    030import org.apache.commons.collections4.BoundedCollection;
    031
    032/**
    033 * CircularFifoQueue is a first-in first-out queue with a fixed size that
    034 * replaces its oldest element if full.
    035 * <p>
    036 * The removal order of a {@link CircularFifoQueue} is based on the
    037 * insertion order; elements are removed in the same order in which they
    038 * were added.  The iteration order is the same as the removal order.
    039 * <p>
    040 * The {@link #add(Object)}, {@link #remove()}, {@link #peek()}, {@link #poll},
    041 * {@link #offer(Object)} operations all perform in constant time.
    042 * All other operations perform in linear time or worse.
    043 * <p>
    044 * This queue prevents null objects from being added.
    045 *
    046 * @param <E> the type of elements in this collection
    047 * @since 4.0
    048 */
    049public class CircularFifoQueue<E> extends AbstractCollection<E>
    050    implements Queue<E>, BoundedCollection<E>, Serializable {
    051
    052    /** Serialization version. */
    053    private static final long serialVersionUID = -8423413834657610406L;
    054
    055    /** Underlying storage array. */
    056    private transient E[] elements;
    057
    058    /** Array index of first (oldest) queue element. */
    059    private transient int start = 0;
    060
    061    /**
    062     * Index mod maxElements of the array position following the last queue
    063     * element.  Queue elements start at elements[start] and "wrap around"
    064     * elements[maxElements-1], ending at elements[decrement(end)].
    065     * For example, elements = {c,a,b}, start=1, end=1 corresponds to
    066     * the queue [a,b,c].
    067     */
    068    private transient int end = 0;
    069
    070    /** Flag to indicate if the queue is currently full. */
    071    private transient boolean full = false;
    072
    073    /** Capacity of the queue. */
    074    private final int maxElements;
    075
    076    /**
    077     * Constructor that creates a queue with the default size of 32.
    078     */
    079    public CircularFifoQueue() {
    080        this(32);
    081    }
    082
    083    /**
    084     * Constructor that creates a queue with the specified size.
    085     *
    086     * @param size  the size of the queue (cannot be changed)
    087     * @throws IllegalArgumentException  if the size is &lt; 1
    088     */
    089    @SuppressWarnings("unchecked")
    090    public CircularFifoQueue(final int size) {
    091        if (size <= 0) {
    092            throw new IllegalArgumentException("The size must be greater than 0");
    093        }
    094        elements = (E[]) new Object[size];
    095        maxElements = elements.length;
    096    }
    097
    098    /**
    099     * Constructor that creates a queue from the specified collection.
    100     * The collection size also sets the queue size.
    101     *
    102     * @param coll  the collection to copy into the queue, may not be null
    103     * @throws NullPointerException if the collection is null
    104     */
    105    public CircularFifoQueue(final Collection<? extends E> coll) {
    106        this(coll.size());
    107        addAll(coll);
    108    }
    109
    110    //-----------------------------------------------------------------------
    111    /**
    112     * Write the queue out using a custom routine.
    113     *
    114     * @param out  the output stream
    115     * @throws IOException if an I/O error occurs while writing to the output stream
    116     */
    117    private void writeObject(final ObjectOutputStream out) throws IOException {
    118        out.defaultWriteObject();
    119        out.writeInt(size());
    120        for (final E e : this) {
    121            out.writeObject(e);
    122        }
    123    }
    124
    125    /**
    126     * Read the queue in using a custom routine.
    127     *
    128     * @param in  the input stream
    129     * @throws IOException if an I/O error occurs while writing to the output stream
    130     * @throws ClassNotFoundException if the class of a serialized object can not be found
    131     */
    132    @SuppressWarnings("unchecked")
    133    private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
    134        in.defaultReadObject();
    135        elements = (E[]) new Object[maxElements];
    136        final int size = in.readInt();
    137        for (int i = 0; i < size; i++) {
    138            elements[i] = (E) in.readObject();
    139        }
    140        start = 0;
    141        full = size == maxElements;
    142        if (full) {
    143            end = 0;
    144        } else {
    145            end = size;
    146        }
    147    }
    148
    149    //-----------------------------------------------------------------------
    150    /**
    151     * Returns the number of elements stored in the queue.
    152     *
    153     * @return this queue's size
    154     */
    155    @Override
    156    public int size() {
    157        int size = 0;
    158
    159        if (end < start) {
    160            size = maxElements - start + end;
    161        } else if (end == start) {
    162            size = full ? maxElements : 0;
    163        } else {
    164            size = end - start;
    165        }
    166
    167        return size;
    168    }
    169
    170    /**
    171     * Returns true if this queue is empty; false otherwise.
    172     *
    173     * @return true if this queue is empty
    174     */
    175    @Override
    176    public boolean isEmpty() {
    177        return size() == 0;
    178    }
    179
    180    /**
    181     * {@inheritDoc}
    182     * <p>
    183     * A {@code CircularFifoQueue} can never be full, thus this returns always
    184     * {@code false}.
    185     *
    186     * @return always returns {@code false}
    187     */
    188    @Override
    189    public boolean isFull() {
    190        return false;
    191    }
    192
    193    /**
    194     * Returns {@code true} if the capacity limit of this queue has been reached,
    195     * i.e. the number of elements stored in the queue equals its maximum size.
    196     *
    197     * @return {@code true} if the capacity limit has been reached, {@code false} otherwise
    198     * @since 4.1
    199     */
    200    public boolean isAtFullCapacity() {
    201        return size() == maxElements;
    202    }
    203
    204    /**
    205     * Gets the maximum size of the collection (the bound).
    206     *
    207     * @return the maximum number of elements the collection can hold
    208     */
    209    @Override
    210    public int maxSize() {
    211        return maxElements;
    212    }
    213
    214    /**
    215     * Clears this queue.
    216     */
    217    @Override
    218    public void clear() {
    219        full = false;
    220        start = 0;
    221        end = 0;
    222        Arrays.fill(elements, null);
    223    }
    224
    225    /**
    226     * Adds the given element to this queue. If the queue is full, the least recently added
    227     * element is discarded so that a new element can be inserted.
    228     *
    229     * @param element  the element to add
    230     * @return true, always
    231     * @throws NullPointerException  if the given element is null
    232     */
    233    @Override
    234    public boolean add(final E element) {
    235        if (null == element) {
    236            throw new NullPointerException("Attempted to add null object to queue");
    237        }
    238
    239        if (isAtFullCapacity()) {
    240            remove();
    241        }
    242
    243        elements[end++] = element;
    244
    245        if (end >= maxElements) {
    246            end = 0;
    247        }
    248
    249        if (end == start) {
    250            full = true;
    251        }
    252
    253        return true;
    254    }
    255
    256    /**
    257     * Returns the element at the specified position in this queue.
    258     *
    259     * @param index the position of the element in the queue
    260     * @return the element at position {@code index}
    261     * @throws NoSuchElementException if the requested position is outside the range [0, size)
    262     */
    263    public E get(final int index) {
    264        final int sz = size();
    265        if (index < 0 || index >= sz) {
    266            throw new NoSuchElementException(
    267                    String.format("The specified index (%1$d) is outside the available range [0, %2$d)",
    268                                  Integer.valueOf(index), Integer.valueOf(sz)));
    269        }
    270
    271        final int idx = (start + index) % maxElements;
    272        return elements[idx];
    273    }
    274
    275    //-----------------------------------------------------------------------
    276
    277    /**
    278     * Adds the given element to this queue. If the queue is full, the least recently added
    279     * element is discarded so that a new element can be inserted.
    280     *
    281     * @param element  the element to add
    282     * @return true, always
    283     * @throws NullPointerException  if the given element is null
    284     */
    285    @Override
    286    public boolean offer(final E element) {
    287        return add(element);
    288    }
    289
    290    @Override
    291    public E poll() {
    292        if (isEmpty()) {
    293            return null;
    294        }
    295        return remove();
    296    }
    297
    298    @Override
    299    public E element() {
    300        if (isEmpty()) {
    301            throw new NoSuchElementException("queue is empty");
    302        }
    303        return peek();
    304    }
    305
    306    @Override
    307    public E peek() {
    308        if (isEmpty()) {
    309            return null;
    310        }
    311        return elements[start];
    312    }
    313
    314    @Override
    315    public E remove() {
    316        if (isEmpty()) {
    317            throw new NoSuchElementException("queue is empty");
    318        }
    319
    320        final E element = elements[start];
    321        if (null != element) {
    322            elements[start++] = null;
    323
    324            if (start >= maxElements) {
    325                start = 0;
    326            }
    327            full = false;
    328        }
    329        return element;
    330    }
    331
    332    //-----------------------------------------------------------------------
    333    /**
    334     * Increments the internal index.
    335     *
    336     * @param index  the index to increment
    337     * @return the updated index
    338     */
    339    private int increment(int index) {
    340        index++;
    341        if (index >= maxElements) {
    342            index = 0;
    343        }
    344        return index;
    345    }
    346
    347    /**
    348     * Decrements the internal index.
    349     *
    350     * @param index  the index to decrement
    351     * @return the updated index
    352     */
    353    private int decrement(int index) {
    354        index--;
    355        if (index < 0) {
    356            index = maxElements - 1;
    357        }
    358        return index;
    359    }
    360
    361    /**
    362     * Returns an iterator over this queue's elements.
    363     *
    364     * @return an iterator over this queue's elements
    365     */
    366    @Override
    367    public Iterator<E> iterator() {
    368        return new Iterator<E>() {
    369
    370            private int index = start;
    371            private int lastReturnedIndex = -1;
    372            private boolean isFirst = full;
    373
    374            @Override
    375            public boolean hasNext() {
    376                return isFirst || index != end;
    377            }
    378
    379            @Override
    380            public E next() {
    381                if (!hasNext()) {
    382                    throw new NoSuchElementException();
    383                }
    384                isFirst = false;
    385                lastReturnedIndex = index;
    386                index = increment(index);
    387                return elements[lastReturnedIndex];
    388            }
    389
    390            @Override
    391            public void remove() {
    392                if (lastReturnedIndex == -1) {
    393                    throw new IllegalStateException();
    394                }
    395
    396                // First element can be removed quickly
    397                if (lastReturnedIndex == start) {
    398                    CircularFifoQueue.this.remove();
    399                    lastReturnedIndex = -1;
    400                    return;
    401                }
    402
    403                int pos = lastReturnedIndex + 1;
    404                if (start < lastReturnedIndex && pos < end) {
    405                    // shift in one part
    406                    System.arraycopy(elements, pos, elements, lastReturnedIndex, end - pos);
    407                } else {
    408                    // Other elements require us to shift the subsequent elements
    409                    while (pos != end) {
    410                        if (pos >= maxElements) {
    411                            elements[pos - 1] = elements[0];
    412                            pos = 0;
    413                        } else {
    414                            elements[decrement(pos)] = elements[pos];
    415                            pos = increment(pos);
    416                        }
    417                    }
    418                }
    419
    420                lastReturnedIndex = -1;
    421                end = decrement(end);
    422                elements[end] = null;
    423                full = false;
    424                index = decrement(index);
    425            }
    426
    427        };
    428    }
    429
    430}