import java.io.IOException;
import java.io.Reader;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Pipe extends Reader {

    private final Lock monitorLock = new ReentrantLock();
    private final Queue<Condition> waitingForFirstRead = new LinkedList<Condition>();
    private Map<Thread, Condition> waitingForSecondRead = new HashMap<Thread, Condition>();

    private final Queue<Character> writeQueue = new LinkedList<Character>();
    private final Condition notEmpty = monitorLock.newCondition();

    private boolean closed = false;
    private boolean eof = false;

    public void write(char ch) {
        write(ch, false);
    }

    //NB: combine write() with eof() because the last write() will block, making it hard to call eof()
    public void writeFinal(char ch) {
        write(ch, true);
    }

    private void write(char ch, boolean isFinal) {
        monitorLock.lock();
        try {
            if(closed) {
                throw new IllegalStateException("Cannot write after the read end has been closed.");
            }
            if(eof) {
                throw new IllegalStateException("Cannot write after EOF.");
            }
            if(isFinal) {
                eof = true;
            }
            writeQueue.add(ch);
            Condition cond = monitorLock.newCondition();
            waitingForFirstRead.add(cond);
            notEmpty.signalAll();
            cond.awaitUninterruptibly();
        } finally {
            monitorLock.unlock();
        }
    }

    @Override
    //NB: ignore length and always write 1 byte (unless at EOF)
    public int read(char[] buf, int offset, int length) throws IOException {
        monitorLock.lock();
        try {
            if(closed) {
                throw new IllegalStateException("Cannot read after the read end has been closed.");
            }
            boolean freedWriter = freeNextWriter();
            while(!eof && writeQueue.isEmpty()) {
                notEmpty.awaitUninterruptibly();
                if(!freedWriter) {
                    freedWriter = freeNextWriter();
                }
            }
            if(eof && writeQueue.isEmpty()) {
                //no writers waiting, so none to free
                return -1;
            }
            waitingForSecondRead.put(Thread.currentThread(), waitingForFirstRead.poll());
            buf[offset] = writeQueue.poll();
            return 1;
        } finally {
            monitorLock.unlock();
        }
    }

    @Override
    public void close() throws IOException {
        monitorLock.lock();
        try {
            closed = true;
            freeAllWriters();
        } finally {
            monitorLock.unlock();
        }
    }

    //NB: didn't bother with monitorLock because this method is always called safely
    private boolean freeNextWriter() {
        Condition cond = waitingForSecondRead.get(Thread.currentThread());
        if(cond != null) {
            cond.signal();
            waitingForSecondRead.remove(Thread.currentThread());
            return true;
        }
        return false;
    }

    //NB: didn't bother with monitorLock because this method is always called safely
    private void freeAllWriters() {
        for(Condition cond : waitingForSecondRead.values()) {
            cond.signal();
        }
        waitingForSecondRead.clear();
        for(Condition cond : waitingForFirstRead) {
            cond.signal();
        }
        waitingForFirstRead.clear();
    }
}
