// ActiveStream.java
// $Id: ActiveStream.java,v 1.9 1997/01/28 10:35:29 abaird Exp $
// (c) COPYRIGHT MIT and INRIA, 1996.
// Please first read the full copyright statement in file COPYRIGHT.html


package w3c.www.protocol.http.cache;

import java.io.*;

import w3c.util.*;

class ActiveInputStream extends InputStream {
    byte buffer[] = new byte[4096];
    int  off      = 0;
    int  len      = 0;
    
    boolean closed      = false;
    boolean interrupted = false;

    private synchronized void waitForInput()
	throws IOException
    {
	while (( ! closed) && (len == 0)) {
	    if ( interrupted )
		throw new IOException("Broken active pipe.");
	    try {
		wait();
	    } catch (InterruptedException ex) {
	    }
	}
    }

    public synchronized void receive(byte buf[], int boff, int blen) 
	throws IOException
    {
	if ( closed )
	    throw new IOException("Write to closed stream.");
	while ((! closed) && (len != 0)) {
	    try { wait(); } catch (InterruptedException ex) {} 
	}
	System.arraycopy(buf, boff, buffer, 0, blen);
	this.off    = 0;
	this.len    = blen;
	notifyAll();
    }

    public synchronized void close() {
	closed = true;
	notifyAll();
    }
    
    public synchronized void interrupt() {
	interrupted = true;
	closed      = true;
	notifyAll();
    }

    public synchronized int read()
	throws IOException
    {
	waitForInput();
	if (closed && (len == 0))
	    return -1;
	int b = buffer[off++];
	if ( off >= len ) {
	    len = 0;
	    notifyAll();
	}
	return b;
    }

    public synchronized int read(byte to[], int toff, int tlen) 
	throws IOException
    {
	waitForInput();
	// Check for exhausted stream:
	if (closed && (len == 0))
	    return -1;
	// Send the appropriate stuff:
	if (tlen > len) {
	    int snd = len;
	    System.arraycopy(buffer, off, to, toff, len);
	    len = 0;
	    notifyAll();
	    return snd ;
	} else {
	    System.arraycopy(buffer, off, to, toff, tlen);
	    len -= tlen;
	    return tlen;
	}
    }

    public synchronized int available() {
	return (closed || interrupted) ? -1 : len-off;
    }

}

/**
 * ActiveStream is used to tee a stream to the client, while caching it.
 * This class basically mimics the piped streams provided in the java library
 * in a more efficient manner (well, sort of).
 * <p>If any error occurs while writing data back to the client, then the
 * active thread finishes it works, but only streaming data into the sink,
 */

public class ActiveStream implements Runnable {
    private static ThreadCache threadcache = null;

    ActiveInputStream pout    = null;
    TeeMonitor        monitor = null;
    InputStream       src = null;
    OutputStream      dst = null;
    
    public void run() {
	byte    buffer[] = new byte[2048];
	int     chunksz  = 256;
	boolean notified = false;

	try {
	    int count = 0;
	    int total = 0;
	    while ((count = src.read(buffer, 0, chunksz)) > 0) {
		// Try to write to the pipe, is still valid:
		if ( pout != null ) {
		    try {
			pout.receive(buffer, 0, count);
		    } catch (IOException ex) {
			try { pout.close(); } catch (Exception e) {}
			pout = null;
		    }
		}
		// Always write to destination:
		dst.write(buffer, 0, count);
		total += count;
		// Increment the chunk size, for improved performance:
		chunksz = Math.min(buffer.length, chunksz << 1);
	    }
	    src.close();
	    src = null;
	    dst.close();
	    dst = null;
	    if ( pout != null ) {
		pout.close();
		pout = null;
	    }
	    monitor.notifyTeeSuccess(total);
	    notified = true;
	} catch (IOException ex) {
	    ex.printStackTrace();
	    monitor.notifyTeeFailure();
	    notified = true;
	} finally {
	    try {
		if ( src != null )
		    try { src.close(); } catch (Exception ex) {}
		if ( dst != null )
		    try {dst.close(); } catch (Exception ex) {}
		if ( pout != null ) 
		    try { pout.interrupt(); } catch (Exception ex) {}
		if ( ! notified )
		    monitor.notifyTeeFailure();
	    } catch (Exception ex) {
	    }
	}
    }

    public static InputStream createTee(TeeMonitor monitor
				 , InputStream src
				 , OutputStream dst) 
	throws IOException
    {
	// Allocate a new tee stream:
	ActiveStream tee = new ActiveStream();
	tee.monitor = monitor;
	tee.pout    = new ActiveInputStream();
	tee.src     = src;
	tee.dst     = dst;
	// Allocate a thread for this tee stream:
	if ( ! threadcache.getThread(tee, false) ) {
	    return null;
	} else {
	    return tee.pout;
	}
    }

    public static synchronized void initialize() {
	if ( threadcache == null ) {
	    threadcache = new ThreadCache("active-streams");
	    threadcache.setCachesize(10);
	    threadcache.initialize();
	}
    }

    ActiveStream() {
    }


}
