serviceoutputstream.java

移动Agent编程工具Naplet

/*
 * @<#> ServiceOutputStream.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.IOException;
import java.io.OutputStream;
import java.io.*;

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

/**
 * The <code>ServiceOutputStream</code> class defines an non-symmetric output
 * piped stream to naplets.
 *
 * @version 0.0.1, 1/1/2001
 * @author C. Xu ([e-mail])
 */
public class ServiceOutputStream extends OutputStream
{

      NapletInputStream sink = null;

      boolean closedByServer = false;
      boolean closedByNaplet = false;
      boolean connected = false;

      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 ServiceOutputStream and read by
       * NapletInputStream. <code>in&lt;0</code> implies the buffer is empty,
       * and <code>in==out</code> implies the buffer is full.
       */
      int in = -1;
      int out = 0;

      /**
       * Create a <code>ServiceOutputStream</code> to be connected by
       * <code>NapletInputStream</code>. If the pipe stream is accessed before
       * connection, an IOException is thrown.
       */
      public ServiceOutputStream( )
      {
      }

      /**
       * Called by a <code>NapletInputStream</code> to establish a piped
       * stream with this <code>ServiceOutputStream</code>.
       *
       * @param snk The <code>NapletInputStream</code> connected this
       *            <code>SerivceOutputStream</code>
       */
      public void connect( NapletInputStream snk )
	 throws IOException
      {
	 if ( snk == null)
	 {
            throw new NullPointerException( );
	 }

	 synchronized ( this )
	 {
	    while ( connected )
	    {
	       if ( snk.equals( sink ) )
	       { // Already connected by the same NapletInputStream
		  // No effect
	       }
	       else
	       {        // Already connected by other naplet
		  try
		  {
		     wait( );
		  }
		  catch (InterruptedException ie)
		  {
		     throw new IOException("Already connected");
		  }
	       }
	    }

	    sink = snk;
	    snk.source = this;
	    connected = true;

	    notifyAll( );
	    in = -1;
	    out = 0;
	 }
      }

      /**
       * Writes the specified <code>byte</code> to the piped output stream.
       * If a thread was reading data bytes from the connected piped
       * input stream, but the thread is no longer alive, then an
       * <code>IOException</code> is thrown.
       * <p>
       * Implements the <code>write</code> method of <code>OutputStream
       * </code>.
       *
       * @param      c   the <code>char</code> to be written.
       * @exception  IOException  if an I/O error occurs.
       */
      public void write( int c )
	 throws IOException
      {
	 synchronized ( this )
	 {
	    while ( !connected )
	    {
	       try
	       {
		  wait( );
	       }
	       catch (InterruptedException ie)
	       {
		  throw new java.io.InterruptedIOException( );
	       }
	    }

	    while ( in == out )
	    {
	       // Pipe is full, wake up readers and wait for it.
	       notifyAll( );

	       try
	       {
		  wait( );
	       }
	       catch (InterruptedException ie)
	       {
		  throw new java.io.InterruptedIOException( );
	       }

	       // During server waiting, the pipe might be closed by naplet
	       if ( closedByNaplet )
	       {
		  closedByNaplet = false;
		  throw new IOException(
		     "Pipe closed by naplet during server waitng");
	       }
	    }

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

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

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

	    notifyAll( );
	 }
      }

      /**
       * Writes <code>len</code> bytes from the specified byte
       * array starting at offset <code>off</code> to this piped output
       * stream. If a thread was reading data bytes from the connected
       * piped input stream, but the thread is no longer alive, then an
       * <code>IOException</code> is thrown.
       *
       * @param      cbuf  the data.
       * @param      off   the start offset in the data.
       * @param      len   the number of bytes to write.
       * @exception  IOException  if an I/O error occurs.
       */
      public void write( 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( );
	 }

	 synchronized( this )
	 {
	    while ( !connected )
	    {
	       try
	       {
		  wait( );
	       }
	       catch ( InterruptedException ie )
	       {
		  throw new java.io.InterruptedIOException( );
	       }
	    }

	    while ( --len >= 0 )
	    {
 	       while ( in == out )
	       {
		  //  Pipe is full, wake up readers and wait for it.
		  notifyAll( );

		  try
		  {
		     wait( );
		  }
		  catch ( InterruptedException ie )
		  {
		     throw new java.io.InterruptedIOException( );
		  }
	       }

	       if ( closedByNaplet )
	       {
		  closedByNaplet = false;
		  throw new IOException(
		     "Pipe closed by naplet during server waiting" );
	       }

	       if ( in < 0 )
	       {
		  in = 0;
	       }

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

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

	    notifyAll( );
 	 }
      }

      /**
       * Called by a NapletInputStream to get 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.
       */
      protected int get( )
	 throws IOException
      {
	 synchronized ( this )
	 {
            if ( !connected && !closedByServer )
	    {
	       // If pipe is initially disconnected
	       throw new IOException( "Pipe is not connected" );
            }

	    try
	    {
                int counter = 0;
                 while ( in < 0 )
                 {
                     if ( closedByServer && !connected )
                     {
                         closedByServer = false;
                         return -1; // return EOF
                     }

                     counter++;
                     if ( counter % 5 == 0 )
                     {
                         switch ( ( counter / 5 ) % 5 )
                         {
                             case 0:
                                 System.out.println(
                                     "\rNaplet wait for results from service .    " );
                                 break;
                             case 1:
                                 System.out.println(
                                     "\rNaplet wait for results from service ..   " );
                                 break;
                             case 2:
                                 System.out.println(
                                     "\rNaplet wait for results from service ...  " );
                                 break;

                             case 3:
                                 System.out.println(
                                     "\rNaplet wait for results from service .... " );
                                 break;

                             case 4:
                                 System.out.println(
                                     "\rNaplet wait for results from service ....." );
                                 break;
                         }
                     } // Output the waiting message

                     wait();
                 } // The pipe is empty
	    }
	    catch ( InterruptedException ie )
	    {
	       throw new java.io.InterruptedIOException( );
	    }

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

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

	       //  Release the closing of ServiceWriter
	       notifyAll( );
	    }

	    // Release waiting server
	    notifyAll( );

	    return ret;
	 }
      }

      /**
       * Called by NapletInputStream to get 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.
       */
      public int get( 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 = get( );
	 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;
	       }
	    }

	    notifyAll( );
	 }

	 return rlen;
      }

      public synchronized void flush( )
	 throws IOException
      {
	 if ( sink == null )
	 {
            throw new IOException( "Pipe is not connected" );
	 }

	 synchronized ( this )
	 {
            this.notifyAll( );
	 }
      }

      /**
       * Closes this piped output stream and releases any system resources
       * associated with this stream. This stream cannot be used for
       * writing bytes, without being connected by a
       * <code>NapletInputStream</code>.
       *
       * @exception  IOException  if an I/O error occurs.
       */
      public void close( )
	 throws IOException
      {
	 synchronized ( this )
	 {
	    while ( in >= 0 )
	    {
	       try
	       {
		  wait( );
	       }
	       catch ( InterruptedException ie )
	       {
		  throw new java.io.InterruptedIOException( );
	       }
	    }

	    closedByServer = true;
	    notifyAll();
	    connected = false;
	 }
      }
}