servicewriter.java

移动Agent编程工具Naplet

/*
 * @<#> ServiceWriter.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.Writer;
import java.io.*;
import naplet.*;
import naplet.serviceChannel.*;

/**
 * The <code>ServiceWriter</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 ServiceWriter extends Writer
{
    NapletReader sink = null;
    boolean closedByServer = false;
    boolean closedByNaplet = false;
    boolean connected = false;
    static final int PIPE_SIZE = 1024;
    char buffer[] = new char[PIPE_SIZE];
    /**
     * The position index pair in the circular buffer at which the next
     * character of data will be stored by ServiceWriter and read by
     * NapletReader. <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>ServiceWriter</code> to be connected by
     * <code>NapletReader</code>. If the pipe stream is accessed before
     * connection, an IOException is thrown.
     */
    public ServiceWriter()
    {

    }

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

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

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

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

    /**
     * Writes the specified <code>char</code> to the piped output stream.
     * If a thread was reading data characters 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>Writer</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 ( lock )
        {
            while ( !connected )
            {
                try
                {
                    lock.wait();
                }
                catch ( InterruptedException ie )
                {
                    throw new java.io.InterruptedIOException();
                }
            }

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

                try
                {
                    lock.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++] = ( char ) c;

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

            lock.notifyAll();
        }
    }

    /**
     * Writes <code>len</code> characters from the specified character
     * array starting at offset <code>off</code> to this piped output
     * stream. If a thread was reading data characters 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 characters to write.
     * @exception  IOException  if an I/O error occurs.
     */
    public void write( char 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( "Index out of boundary" );
        }

        synchronized ( lock )
        {
            while ( !connected )
            {
                try
                {
                    lock.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.
                    lock.notifyAll();

                    try
                    {
                        lock.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;
                }
            }

            lock.notifyAll();
        }
    }

    // counter to record how many times the method get( ) is called
    // used for output of waiting message.
    static int counter = 0;

    /**
     * Called by a NapletReader to get the next character of data from this
     * piped stream. If no character 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 ( lock )
        {
            if ( !connected && !closedByServer )
            {
                // If pipe is initially disconnected
                throw new IOException(
                    "ServiceWriter( get ): Pipe is not connected" );
            }

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

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

                            case 3:
                                System.out.print(
                                    "\rNaplet waits for results from the service .... " );
                                break;

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

                    lock.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
                lock.notifyAll();
            }

            // Release waiting server
            lock.notifyAll();

            return ret;
        }
    }

    /**
     * Called by NapletReader to get up to <code>len</code> characters of
     * data from this piped stream into an array of characters. Less than
     * <code>len</code> characters will be read if the end of the data
     * stream is reached. This method blocks until at least one character
     * of input is available.
     */
    public int get( char 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( "Index out of boundary" );
        }

        /* possibly wait on the first character */
        int c = get();
        if ( c < 0 )
        {
            return -1;
        }
        else if ( len == 0 )
        {
            return 0;
        }

        cbuf[off] = ( char ) c;
        int rlen = 1;

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

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

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

            lock.notifyAll();
        }

        return rlen;
    }

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

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

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

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