 package scone.util.tokenstream;


import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Enumeration;
import java.util.Vector;

import com.ibm.wbi.AbortEvent;
import com.ibm.wbi.AbortIOException;
import com.ibm.wbi.MarkPreservedException;


/**
 * A Buffer to store Tokens.
 * <br>Just do not mess with this class
 * @author Volkert Buchmann (ported from code by Steve Ihde), 
 *         IBM Almaden Research Center, 
 *         (c) Copyright IBM Corp 2000
 * @version 1.0
 */
public class TokenBuffer {

    private Vector subscriptions = new Vector();
    private CircularTokenStore data;
    private long totalWritten = 0;
    private boolean done = false;
    private boolean doneReading = false;
    private AbortEvent abortEvent = null;

    private long lowWaterMark = 0;

    private boolean marked = false;
    private long lowestMark = 0;

    private static long usage = 0;
    private static long maxUsage = 0;
    private static boolean started = false;

    int dummycounter = 0;

    private static synchronized void adjustUsage(long adj) {
        usage += adj;
        if (usage > maxUsage) {
            maxUsage = usage;
        }
    }

    private static synchronized long getUsage() {
        return usage;
    }

    private static synchronized long maxUsage() {
        long rc = maxUsage;

        maxUsage = usage;
        return rc;
    }

    public TokenBuffer(int capacity) {
        data = new CircularTokenStore(capacity);
        adjustUsage(capacity);
    }

    public TokenBuffer(Token source[]) {
        data = new CircularTokenStore(source);
        totalWritten = source.length;
    }

    public void finalize() {
        adjustUsage(-(data.capacity()));
    }

    public synchronized Object subscribe() {
        if (lowWaterMark > 0) {
            // System.out.println("lowwatermark "+lowWaterMark);
            throw new TokenBufferIndexOutOfBoundsException();
        }

        Subscription s = new Subscription();

        subscriptions.addElement(s);
      
        return s;
    }

    public synchronized Object subscribe(Object other) {
        Subscription old = (Subscription) other;
        Subscription s = new Subscription();

        s.offset = old.offset;
        subscriptions.addElement(s);

        return s;
    }

    public synchronized void unsubscribe(Object subscription) {
        subscriptions.removeElement(subscription);
        checkMarks();
        checkLowWaterMark();
    }

    public synchronized void mark(Object subscription, int readlimit) {
        Subscription s = (Subscription) subscription;

        s.marked = true;
        s.mark = s.offset;
        s.readlimit = readlimit;
        if (readlimit == -1) {
            s.unlimited = true;
        }
        checkMarks();

        notifyAll();
    }
   
    public synchronized void reset(Object subscription) throws IOException {

        Subscription s = (Subscription) subscription;

        if (!s.marked) {
            throw new IOException("reset() called on unmarked stream");
        }

        // if (lowWaterMark > s.mark) {
        // System.out.println("Internal error! lowWaterMark > s.mark");
        // System.exit(1);
        // }

        s.offset = s.mark;
        s.marked = false;

        checkMarks();

    }

    public synchronized void preserveMark(Object subscription) throws IOException {

        Subscription s = (Subscription) subscription;
	
        if (!s.marked) {
            throw new IOException("preserveMark() called on unmarked stream");
        }
	
        s.preserveMark = true;

    }

    public boolean markSupported() {
        return true;
    }

    public synchronized long getMark(Object subscription) throws IOException {
        Subscription s = (Subscription) subscription;

        if (!s.marked) {
            throw new IOException("getMark() called on unmarked stream");
        }

        return s.mark;
	
    }

    public synchronized void unmark(Object subscription) {
        Subscription s = (Subscription) subscription;

        s.marked = false;
        s.mark = 0;
        s.readlimit = 0;
        s.unlimited = false;
        s.preserveMark = false;
        checkMarks();
    }

    private synchronized void checkMarks() {
        lowestMark = Long.MAX_VALUE;
        marked = false;
        Enumeration e = subscriptions.elements();

        while (e.hasMoreElements()) {
            Subscription sub = (Subscription) e.nextElement();

            if (sub.marked) {
                marked = true;
                if (sub.mark < lowestMark) {
                    lowestMark = sub.mark;
                }
            }
        }

        checkLowWaterMark();
    }

    private synchronized void unmarkLowestMark() {
        long tmpLowestMark = Long.MAX_VALUE;
        Subscription tmpLowestMarker = null;
	
        Enumeration s = subscriptions.elements();

        while (s.hasMoreElements()) {
            Subscription sub = (Subscription) s.nextElement();

            if (sub.marked) {
                if (sub.mark < tmpLowestMark) {
                    tmpLowestMark = sub.mark;
                    tmpLowestMarker = sub;
                }
            }
        }

        if (tmpLowestMarker != null) {
            // We found a mark to unmark....
            unmark(tmpLowestMarker);
        }
    }

    private synchronized void checkLowWaterMark() {
        long oldLowWaterMark = lowWaterMark;

        lowWaterMark = (marked) ? lowestMark : Long.MAX_VALUE;
        Enumeration e = subscriptions.elements();

        while (e.hasMoreElements()) {
            Subscription sub = (Subscription) e.nextElement();

            if (sub.offset < lowWaterMark) {
                lowWaterMark = sub.offset;
            }
        }
	
        int discard = (int) (lowWaterMark - oldLowWaterMark);

        // dummycounter+=discard;
        // System.out.print(this);
        // System.out.print(" discarded ");
        // System.out.println(dummycounter);
        if (discard > 0) {
            data.discardTokens(discard);
            notifyAll();
        }
    }

    public synchronized int read(Token buf[], int offset, int size, Object context, long timeout, boolean peek) throws AbortIOException, InterruptedIOException, MarkPreservedException {

        // System.out.print("lowWaterMark ");
        // System.out.println(lowWaterMark);
        if (context == null) {
            throw new NullPointerException();
        }

        Subscription s = (Subscription) context;

        long startWait = System.currentTimeMillis();
        long remainingWait = timeout;

        while (s.offset >= totalWritten && !isDone() && !isAborted()) {
            try {
                if (timeout == 0) {
                    wait();
                } else {
                    wait(remainingWait);
                    remainingWait = timeout
                            - (System.currentTimeMillis() - startWait);
                    if (remainingWait < 1) {
                        break;
                    }
                }
            } catch (InterruptedException e) {}
        }
	
        if (isAborted()) {
            throw new AbortIOException(abortEvent);
        }

        int available = Math.min((int) (totalWritten - s.offset), size);

        if (available > 0) {
            data.peekTokens((int) (totalWritten - s.offset), buf, offset, available, mustClone());

            if (!peek) {
                if (s.marked && !s.unlimited && s.preserveMark
                        && available > s.readlimit) {
                    // If the reader has asked to preserve the mark,
                    // give him as much as we can without passing the
                    // mark.  If that amount is zero, throw an
                    // IOException.
                    if (s.readlimit > 0) {
                        available = s.readlimit;
                    } else {
                        throw new MarkPreservedException(
                                "Read would exceed readlimit");
                    }
		    
                }

                s.offset += available;
                if (s.marked && !s.unlimited) {

                    s.readlimit -= available;
                    if (s.readlimit < 0) {
                        unmark(s);
                    }
                }

                checkLowWaterMark();
            }

            return available;
        }

        if (isDone()) {
            // System.out.println("yeah, that's it!");
            return -1;
        }

        throw new InterruptedIOException();

    }

    private void ensureSpace(int size) {
        int oldCapacity = data.capacity();

        data.ensureCapacity(size + data.length());
        int diff = data.capacity() - oldCapacity;

        if (diff != 0) {
            adjustUsage(diff);
        }
    }

    private boolean getSpaceIfMarked(int size) {
        boolean ensured = false;

        while (marked && !ensured) {
            try {
                ensureSpace(size);
                ensured = true;
            } catch (OutOfMemoryError e) {
                unmarkLowestMark();
            }
        }
        return true;
    }

    public synchronized int write(Token buf[], int offset, int size)
            throws TokenBufferClosedException, AbortIOException {

        if (isDone()) {
            throw new TokenBufferClosedException();
        }

        while (getSpaceIfMarked(size) && data.space() == 0 && !isAborted()
                && !(isDoneReading() && subscriptions.size() == 0)) {

            try {
                wait();
            } catch (InterruptedException e) {}

            /*
             try {
             wait(5000);
             if (data.space()==0) {
             System.out.println("circle: " + this.toString() + " Still waiting to write..." + subscriptions.size()+ this.toString() + " " + marked);
             }
             } catch (InterruptedException e) {
             e.printStackTrace();
             }
             */


        }

        if (isAborted()) {
            throw new AbortIOException(abortEvent);
        }

        if (isDoneReading() && subscriptions.size() == 0) {
            return size;
        }

        int written = Math.min(data.space(), size);

        data.appendTokens(buf, offset, written);

        totalWritten += written;

        notifyAll();

        return written;

    }

    public synchronized void done() {
        if (!done) {
            done = true;
            notifyAll();
        }
    }

    public synchronized boolean isDone() {
        return done;
    }

    public synchronized void doneReading() {
        if (!doneReading) {
            doneReading = true;
            notifyAll();
        }
    }

    public synchronized boolean isDoneReading() { 
        return doneReading;
    }

    public synchronized int available(Object context) {
        Subscription s = (Subscription) context;

        return (int) (totalWritten - s.offset);
    }

    public synchronized void handleAbort(AbortEvent ae) {
        abortEvent = ae;
        notifyAll();
    }

    public synchronized boolean isAborted() {
        return abortEvent != null;
    }

    public synchronized AbortEvent getAbortEvent() {
        return abortEvent;
    }

    public long getOffset(Object context) {
        Subscription s = (Subscription) context;

        return s.offset;
    }

    // returns true, if the Objects in the buffer must be cloned
    // when reading
    private boolean mustClone() {
        return marked || (subscriptions.size() > 1);
    }

    class Subscription {
        public long offset = 0;
        public boolean marked = false;
        public long mark = 0;
        public int readlimit = 0;
        public boolean unlimited = false;
        public boolean preserveMark = false;
    }


    public class CircularTokenStore {

        private Token data[] = null;
        private boolean empty = true;
        private int first = 0;
        private int next = 0;

        public CircularTokenStore(int capacity) {
            data = new Token[capacity];
        }

        public CircularTokenStore(Token source[]) {
            data = source;
            empty = false;
            first = 0;
            next = source.length;
        }

        public void appendTokens(Token src[], int offset, int length) {

            int firstCopy = Math.min(data.length - next, length);

            System.arraycopy(src, offset, data, next, firstCopy);

            if (firstCopy < length) {
                System.arraycopy(src, offset + firstCopy, data, 0, length - firstCopy);
            }

            next = (next + length) % data.length;
            empty = false;

        }

        public void discardTokens(int num) {
            first = (first + num) % data.length;
            if (first == next) {
                first = 0;
                next = 0;
                empty = true;
            }
      
        }

        public void peekTokens(int backwardsOffset, Token dest[], int destOffset, int length, boolean mustClone) {

            int srcOffset = next - backwardsOffset;

            if (srcOffset < 0) {
                srcOffset += data.length;
            }

            copy(srcOffset, dest, destOffset, length, mustClone);

        }
  
        public int length() {
            if (empty) {
                return 0;
            }

            if (first < next) {
                return next - first;
            } else { 
                return data.length - first - next;
            }
        }
  
        public int space() {
            if (!empty && next == first) {
                return 0;
            }

            if (first <= next) {
                return data.length - (next - first);
            } else {
                return first - next;
            }
        }

        public int capacity() {
            return data.length;
        }

        public void ensureCapacity(int c) {
            if (c < data.length) {
                return;
            }
    
            int newCapacity = data.length;

            while (newCapacity < c) {
                newCapacity *= 2;
            }
    
            Token newData[] = new Token[newCapacity];

            int oldLength = length();

            copy(first, newData, 0, oldLength, false);
    
            data = newData;
            first = 0;
            next = oldLength;

        }

        private void copy(int srcOffset, Token dest[], int destOffset, int length, boolean mustClone) {
            int firstCopy = Math.min(data.length - srcOffset, length);

            if (mustClone) {
                cloneCopy(data, srcOffset, dest, destOffset, firstCopy);
                if (firstCopy < length) {
                    cloneCopy(data, 0, dest, destOffset + firstCopy, length - firstCopy);
                }
            } else {
                System.arraycopy(data, srcOffset, dest, destOffset, firstCopy);
                if (firstCopy < length) {
                    System.arraycopy(data, 0, dest, destOffset + firstCopy, length - firstCopy);
                }
            }
        }

        // works like System.arraycopy(), but the Tokens in the array are cloned
        private void cloneCopy(Token[] src, int srcPos, Token[] dest, int destPos, int length) {
            for (int i = 0; i < length; i++) {
                dest[destPos++] = src[srcPos++].getClone();
            }
        }

    }
}
