serviceinputstream.java

移动Agent编程工具Naplet

/*
 * @<#> ServiceInputStream.java, version 0.0.1, 1/1/2001
 *
 * THIS PROGRAM IS FREE SOFTWARE; YOU CAN DISTRIBUTE IT AND/OR
 * MODIFY IT UNDER THE TERMS OF THE GNU GENERAL PUBLIC LICENSE 
 * AS PUBLISHED BY THE FREE SOFTWARE FOUNDATION.
 *
 * THIS PROGRAM 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 GENERAL PUBLIC LICENSE FOR MORE DETAILS.
 *
 * Copyright (c) 2000 Wayne State University. All Rights Reserved.
 */

package naplet.serviceChannel;

import java.io.InputStream;
import java.io.IOException;

import naplet.*;
import naplet.serviceChannel.*;

/**
 * Piped byte-input streams between naplets and priviledged services.
 * The pipe is non-symmetric in the sense that a ServiceInputStream may exist 
 * alone and can be connected by NapletOutputStreams at different instances. 
 * By contrast, Java's built-in <code>PipedInputStream</code> and 
 * <code> PipedOutputStream</code> are symmetric and they are dependent on 
 * each other.  
 *
 * @version 0.0.1, 1/1/2001
 * @author C. Xu ([e-mail])
 */

public class ServiceInputStream extends InputStream 
{
      boolean closedByServer = false;
      boolean closedByNaplet = false;
      boolean connected = false;

      /**
       * Non-symmetric piped reader/writer. The ServiceInputStream is a 
       * primary party which can exist alone and be connected to secondary 
       * parties at different time.
       */

      NapletOutputStream source = null;
	
      static final int PIPE_SIZE = 1024;
      byte buffer[] = new byte[PIPE_SIZE];

      /**
       * The position index pair in the circular buffer at which the 
       * next byte of data will be stored by NapletOutputStream and read
       * by ServiceInputStream. <code>in&lt;0</code> implies the buffer is 
       * empty, <code>in==out</code> implies the buffer is full
       */
      int in = -1;
      int out = 0;

      /**
       * Creates a <code>ServiceInputStream</code> with an open input channel.
       * A <code>NapletOutputStream<code> must be connected to it before it 
       * being used.  
       *
       * @see     naplet.ServiceInputStream#connect(naplet.NapletOutputStream)
       */
      public ServiceInputStream( )
      {
      }

      /**
       * Connect a NapletOutputStream to the ServiceInputStream. 
       * If the sink is already connected, throws an IOException
       */
      protected void connect( NapletOutputStream src )
	 throws IOException 
      {
	 if ( src == null )
	 {
            throw new NullPointerException( );
	 }

	 synchronized ( this )
	 {
	    while ( connected )
	    {
	       if ( source == null )
	       {
		  throw new NapletInternalError( 
		     "Null source in a connected pipe" );
	       }

	       if ( source.equals( src ) ) 
	       { 
		  closedByNaplet = false;
		  in = -1;
		  out = 0;
	       } // Already connected by itself
	       else 
	       {
		  System.out.println( "Already connected by other source" );
		 
		  try 
		  {
		     wait( );
		  } 
		  catch ( InterruptedException ie ) 
		  {
		     throw new java.io.InterruptedIOException( );
		  }
	       }
	    }

	    source = src;
	    src.sink = this;
	    in = -1;
	    out = 0;
	    connected = true;

	    // To release any waiting ServiceInputStream
	    notifyAll();
	 }
      }

 
      /**
       * Close this piped stream and releases any system resources 
       * associated with the stream. A question is how to safely close 
       * the pipe. 
       *
       * @exception  IOException  if an I/O error occurs.
       */
      public void close( )  
	 throws IOException 
      {
	 synchronized ( this )
	 {
	    in = -1;
	    closedByServer = true;
	    connected = false;
	 }	
      }

      /**
       * Write a char of data to the pipe stream.  If the pipe is unconnected,
       * an IOException is thrown. This method will block if the pipe is full.
       */
      void receive( int c )
	 throws IOException
      {
	 synchronized ( this )
	 {
	    if ( !connected )
	    {
	       if ( closedByServer )
	       {
		  throw new IOException( "Service is discontinued" );
	       }
	       else
	       {
		  throw new IOException( "Pipe not connected" );
	       }
            } 
			
	    while ( in == out ) 
	    {    
	       // Pipe is full, wake up readers and wait for it.
	       notifyAll( );	
	    
	       try
	       {
		  wait( );
	       } 
	       catch ( InterruptedException ex )
	       {
		  throw new java.io.InterruptedIOException( );
	       }

	       if ( closedByServer )
	       {
		  throw new IOException( "Pipe closed by Server" );
	       }
	    }

	    if ( in < 0 ) 
	    {			
	       in = 0;
	    }  // If pipe is empty

	    buffer[in++] = ( byte ) c;

	    if ( in >= buffer.length ) 
	    {
	       in = 0;
	    }

	    notifyAll();
	 }
      }

      /**
       * Receives data into an array of bytes.  This method will
       * block until some input is available. 
       */
      void receive( byte c[], int off, int len )
	 throws IOException
      {
	 synchronized ( this )
	 {
	    if ( !connected )
	    {
	       if ( closedByServer )
	       {
		  throw new IOException( "Service is discontinued" );
	       }
	       else
	       {
		  throw new IOException( "Pipe not connected" );
	       }
            } 
	    
	    while ( --len >= 0 )
	    {
	       while ( in == out ) 
	       {    
		  // Pipe is full, wake up readers and wait for it.
		  notifyAll( );	
	    
		  try
		  {
		     wait( );
		  } 
		  catch ( InterruptedException ex )
		  {
		     throw new java.io.InterruptedIOException( );
		  }

		  if ( closedByServer )
		  {
		     throw new IOException( "Pipe closed by Server" );
		  }
	       }

	       if ( in < 0 ) 
	       {			
		  in = 0;
	       }  // If pipe is empty

	       buffer[in++] = c[off++];

	       if ( in >= buffer.length ) 
	       {
		  in = 0;
	       }
	    }

	    notifyAll( );
	 }
      }

      /**
       * Reads the next byte of data from this piped stream.
       * If no byte is available because the end of the stream 
       * has been reached, the value <code>-1</code> is returned. 
       * 
       * This method blocks until input data is available, the end of
       * the stream is detected, or an exception is thrown. 
       *
       * If a thread was providing data bytes
       * to the connected piped writer, but
       * the  thread is no longer alive, then an
       * <code>IOException</code> is thrown.
       *
       * @return     the next byte of data, or <code>-1</code> 
       *             if the end of the stream is reached.
       * @exception  IOException  if the pipe is broken.
       */
      public int read( )
	 throws IOException
      {
	 synchronized ( this )
	 {
	    while ( !connected && !closedByNaplet )
	    {
	       // If pipe is initially disconnected  
	       try
	       { 
		  wait( );
	       } 
	       catch ( InterruptedException ie )
	       {
		  throw new java.io.InterruptedIOException( );
	       }
	    }

	    // If pipe is connected, or recently disconnected by Naplet
	    // Since a recently disconnected pipe may not be empty, it must
	    // be flushed before it is reset to the initial condition.

	    try 
	    {
	       while ( in < 0 )
	       {  	
		  if ( !connected && closedByNaplet )
		  { 
		     // reset to initial state 
		     closedByNaplet = false; 
		     return -1;
		  }
		  
		  notifyAll( );
		  wait( );
	       }  // The pipe is empty 		
	    }
	    catch ( InterruptedException ie )
	    {
	       close( );
	       throw new java.io.InterruptedIOException( 
		  "Service is discontinued" );
	    }

	    int ret = buffer[out++];
	    if ( out >= buffer.length ) 
	    {
	       out = 0;
	    }

	    if ( in == out )
	    {   
	       in = -1;		
	       out = 0;
	    }  // Pipe is now empty 

	    notifyAll( );

	    return ret;
	 }
      }

      /**
       * Reads up to <code>len</code> bytes of data from this piped
       * stream into an array of bytes. Less than <code>len</code> 
       * bytes will be read if the end of the data stream is reached. 
       * This method blocks until at least one byte of input is 
       * available. If a thread was providing data bytes to the 
       * connected piped output, but the thread is no longer alive, then 
       * an <code>IOException</code> is thrown.
       *
       * @param      cbuf     the buffer into which the data is read.
       * @param      off   the start offset of the data.
       * @param      len   the maximum number of bytes read.
       * @return     the total number of bytes read into the buffer, or
       *             <code>-1</code> if there is no more data because the end 
       *             of the stream has been reached.
       * @exception  IOException  if an I/O error occurs.
       */
      public int read( byte cbuf[], int off, int len ) 
	 throws IOException 
      {
	 if ( ( off < 0 ) || ( off >= cbuf.length ) || ( len < 0 )
	      || ( ( off + len ) > cbuf.length ) || ( ( off + len ) < 0 ) )
	 {
	    throw new IndexOutOfBoundsException( );
	 } 
	
	 /* possibly wait on the first byte */
	 int c = read( );		
	 if ( c < 0 )
	 {
	    return -1;
	 }
	 else if ( len == 0 )
	 {
	    return 0;
	 }

	 cbuf[off] = ( byte ) c;

	 int rlen = 1;
	 synchronized ( this )
	 {
	    while ( ( in >= 0 ) && ( --len > 0 ) )
	    {
	       cbuf[off + rlen] = buffer[out++];
	       rlen++;

	       if ( out >= buffer.length )
	       {
		  out = 0;
	       }

	       if (in == out)
	       {  
		  in = -1;
		  out = 0;
	       }  // Empty buffer
	    }
	 }

	 return rlen;
      }
}