ring-buffer

package jet;

import com.hazelcast.ringbuffer.StaleSequenceException;


public class RingBuffer<T> {

    private T[] ringItems;

    private long tailSequence = -1;
    private long headSequence = tailSequence + 1;
    private int capacity;


    public RingBuffer(int capacity) {
        this.capacity = capacity;
        this.ringItems = (T[]) new Object[capacity];
    }

    public long tailSequence() {
        return tailSequence;
    }

    public long headSequence() {
        return headSequence;
    }

    public int getCapacity() {
        return capacity;
    }

    public long size() {
        return tailSequence - headSequence + 1;
    }

    public boolean isEmpty() {
        return size() == 0;
    }

    private int toIndex(long sequence) {
        return (int) (sequence % ringItems.length);
    }

    private void checkReadSequence(long sequence) {
        if (sequence > tailSequence) {
            throw new IllegalArgumentException("sequence:" + sequence
                    + " is too large. The current tailSequence is:" + tailSequence);
        }

        if (sequence < headSequence) {
            throw new StaleSequenceException("sequence:" + sequence
                    + " is too small. The current headSequence is:" + headSequence
                    + " tailSequence is:" + tailSequence, headSequence);
        }
    }

    public long add(T item) {
        tailSequence++;

        if (tailSequence - capacity == headSequence) {
            headSequence++;
        }

        int index = toIndex(tailSequence);

        // first we write the dataItem in the ring.
        ringItems[index] = item;

        return tailSequence;
    }

    public long addAll(T[] items) {
        long result = -1;
        for (T item : items) {
            result = add(item);
        }
        return result;
    }

    public T read(long sequence) {
        checkReadSequence(sequence);

        int index = toIndex(sequence);
        return ringItems[index];
    }
}