free/util/PipedStreams.java

/**
 * The utillib library.
 * More information is available at http://www.jinchess.com/.
 * Copyright (C) 2002 Alexander Maryanovsky.
 * All rights reserved.
 *
 * The utillib library is free software; you can redistribute
 * it and/or modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 *
 * The utillib library is distributed in the hope that it will
 * be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
 * General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with utillib library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package free.util;

import java.io.IOException;
import java.io.InterruptedIOException;

/**
 * This class is responsible for creating and managing pairs of PipedStreams.
 * A single instance of this class consists of one PipedInputStream and one
 * PipedOutputStream connected to each other.
 */

public class PipedStreams{


  /**
   * The default buffer size.
   */

  private static final int DEFAULT_BUFFER_SIZE = 2048;


  /**
   * The PipedInputStream.
   */

  private final PipedInputStream in;


  /**
   * The PipedOutputStream.
   */

  private final PipedOutputStream out;


  /**
   * The value of the soTimeout.
   */

  private volatile int soTimeout = 0;


  /**
   * The buffer.
   */

  private byte [] buf;


  /**
   * Whether the buffer is allowed to grow.
   */

  private final boolean growBuf;


  /**
   * The index of the byte that will be read next.
   */

  private int readIndex = 0;


  /**
   * The index of the byte that that will be written next.
   */

  private int writeIndex = 0;


  /**
   * Becomes true when the PipedOutputStream gets closed.
   */

  private boolean writerClosed = false;


  /**
   * Becomes true when the PipedInputStream gets closed.
   */

  private boolean readerClosed = false;


  /**
   * The lock protecting writing.
   */

  private Object writeLock = new String("Write Lock for PipedStreams");


  /**
   * The lock protecting reading.
   */

  private Object readLock = new String("Read Lock for PipedStream");


  /**
   * Creates new <code>PipedStreams</code>.
   */

  public PipedStreams(){
    this(2048, false);
  }


  /**
   * Creates new <code>PipedStreams</code> with the specified buffer size. Once
   * the specified amount of bytes have been written into the,
   * <code>OutputStream</code> attempting to write more data will block until
   * enough data has been read to allow writing into the buffer again.
   */

  public PipedStreams(int bufSize){
    this(bufSize, false);
  }


  /**
   * Creates new <code>PipedStreams</code>. If <code>growBuf</code> is
   * <code>true</code>, the internal buffer will be grown indefinitely when more
   * space is required for the written data. This means that writing into the
   * <code>OutputStream</code> will never block. Note that there is currently no
   * mechanism to cause the internal buffer to shrink.
   */

  public PipedStreams(boolean growBuf){
    this(DEFAULT_BUFFER_SIZE, growBuf);
  }


  /**
   * Creates new <code>PipedStreams</code> with the specified initial buffer
   * size, potentially allowing the buffer to grow indefinitely.
   */

  public PipedStreams(int bufSize, boolean growBuf){
    if (bufSize <= 0)
      throw new IllegalArgumentException("The buffer size must be a positive integer");

    in = new PipedInputStream(this);
    out = new PipedOutputStream(this);

    this.growBuf = growBuf;
    this.buf = new byte[bufSize];
  }


  /**
   * Sets the SO_TIMEOUT for the PipedInputStream. A read operation on the
   * PipedInputStream will only block for the given amount of milliseconds, after
   * that, an InterruptedIOException will be thrown, but the streams will remain
   * valid. A value of 0 implies this option is off (read() can block indefinitely).
   * NOTE: This method should not be called while a read() operation is in progress
   * (it will block until the read() is done, which may be a long time, or never).
   */

  public void setSoTimeout(int timeout){
    synchronized(readLock){ // Don't modify this while a read is in progress.
      soTimeout = timeout;
    }
  }


  /**
   * Returns the value of SO_TIMEOUT. A value of 0 implies this option is off
   * (read() can block indefinitely).
   */

  public int getSoTimeout(){
    return soTimeout;
  }


  /**
   * Returns the PipedInputStream.
   */

  public PipedInputStream getInputStream(){
    return in;
  }


  /**
   * Returns the PipedOutputStream.
   */

  public PipedOutputStream getOutputStream(){
    return out;
  }


  /**
   * Returns the amount of bytes available to be read immediately (without
   * blocking) by the PipedInputStream.
   */

  synchronized int available(){
    if (readerClosed)
      return 0;

    return availableImpl();
  }


  /**
   * Returns the amount of bytes available to be read immediately (without
   * blocking) by the PipedInputStream.
   */

  private int availableImpl(){
    if (writeIndex >= readIndex) // On the same lap.
      return writeIndex - readIndex;
    else // On different laps.
      return writeIndex + buf.length - readIndex;
  }


  /**
   * Returns the amount of bytes that can be written into the buffer without
   * blocking.
   */

  private int availableSpace(){
    return buf.length - availableImpl() - 1;
  }


  /**
   * Increases the size of the internal buffer by at least the specified amount
   * of bytes. The caller must take care of proper synchronization.
   */

  private void growBuf(int minGrowSize){
    int growSize = minGrowSize < buf.length ? buf.length : minGrowSize;
    byte [] newBuf = new byte[buf.length + growSize];
    System.arraycopy(buf, 0, newBuf, 0, buf.length);
    buf = newBuf;
  }


  /**
   * Writes a single byte to the buffer. Note that this method can block if the
   * buffer is full.
   */

  synchronized void write(int b) throws IOException{
    synchronized(writeLock){
      if (readerClosed || writerClosed)
        throw new IOException("Stream closed");

      while (availableSpace() == 0){
        if (growBuf)
          growBuf(1);
        else try{
          wait();
        } catch (InterruptedException e){
            throw new InterruptedIOException();
          }
      }

      if (readerClosed || writerClosed)
        throw new IOException("Stream closed");

      buf[writeIndex++] = (byte)(b&0xff);
      if (writeIndex == buf.length)
        writeIndex = 0;

      notifyAll();
    }
  }


  /**
   * Writes bytes according to the contract of OutputStream.write(byte [], int, int)
   * with the only difference that this might block if the buffer is full.
   */

  synchronized void write(byte [] arr, int offset, int length) throws IOException{
    synchronized(writeLock){
      if (readerClosed||writerClosed)
        throw new IOException("Stream closed");

      if (growBuf && (length > availableSpace()))
        growBuf(length - availableSpace());

      while(length > 0){
        while (availableSpace() == 0){
          try{
            wait();
          } catch (InterruptedException e){
              throw new InterruptedIOException();
            }
        }

        int availableSpace = availableSpace();

        int amountToWrite = length > availableSpace ? availableSpace : length;
        int part1Size = buf.length-writeIndex >= amountToWrite ? amountToWrite : buf.length-writeIndex;
        int part2Size = amountToWrite-part1Size > 0 ? amountToWrite - part1Size : 0;

        System.arraycopy(arr, offset, buf, writeIndex, part1Size);
        System.arraycopy(arr, offset + part1Size, buf, 0, part2Size);

        offset += amountToWrite;
        length -= amountToWrite;

        writeIndex = (writeIndex + amountToWrite) % buf.length;

        notifyAll();
      }
    }
  }


  /**
   * Reads a single byte from the buffer according to the contract of
   * InputStream.read().
   */

  synchronized int read() throws IOException{
    synchronized(readLock){
      if (readerClosed)
        throw new IOException("Stream closed");

      final long startedWaitingTS = System.currentTimeMillis();
      while (available() == 0){
        if (writerClosed)
          return -1;

        long curTime = System.currentTimeMillis();
        if ((soTimeout != 0) && (curTime - startedWaitingTS >= soTimeout))
          throw new InterruptedIOException();

        try{
          if (soTimeout == 0)
            wait();
          else{
            wait(soTimeout + curTime - startedWaitingTS);
          }
        } catch (InterruptedException e){
            throw new InterruptedIOException();
          }
        if (readerClosed)
          throw new IOException("Stream closed");

      }

      int b = buf[readIndex++];
      if (readIndex == buf.length)
        readIndex = 0;

      notifyAll();

      return b < 0 ? b+256 : b;
    }
  }


  /**
   * Reads bytes from the buffer into the given array according to the contract
   * of InputStream.read(byte [], int, int)
   */

  synchronized int read(byte [] arr, int offset, int length) throws IOException{
    synchronized(readLock){
      if (readerClosed)
        throw new IOException("Stream closed");

      final long startedWaitingTS = System.currentTimeMillis();
      while (available() == 0){
        if (writerClosed)
          return -1;

        long curTime = System.currentTimeMillis();
        if ((soTimeout != 0) && (curTime - startedWaitingTS >= soTimeout))
          throw new InterruptedIOException();

        try{
          if (soTimeout == 0)
            wait();
          else{
            wait(soTimeout + curTime - startedWaitingTS);
          }
        } catch (InterruptedException e){
            throw new InterruptedIOException();
          }
        if (readerClosed)
          throw new IOException("Stream closed");
      }

      int available = available();
      int amountToRead = length > available ? available : length;
      int part1Size = buf.length-readIndex > amountToRead ? amountToRead : buf.length-readIndex;
      int part2Size = amountToRead-part1Size > 0 ? amountToRead-part1Size : 0;

      System.arraycopy(buf, readIndex, arr, offset, part1Size);
      System.arraycopy(buf, 0, arr, offset + part1Size, part2Size);

      readIndex = (readIndex + amountToRead) % buf.length;

      notifyAll();

      return amountToRead;
    }
  }


  /**
   * Closes down the streams as far as the PipedOutputStream is concerned.
   */

  synchronized void closeWriter(){
    if (writerClosed)
      throw new IllegalStateException("Already closed");
    writerClosed = true;
    notifyAll();
  }


  /**
   * Closes down the streams as far as the PipedInputStream is concerned
   */

  synchronized void closeReader(){
    if (readerClosed)
      throw new IllegalStateException("Already closed");

    readerClosed = true;
    notifyAll();
  }

}