ojluni/src/main/java/sun/rmi/transport/tcp/MultiplexInputStream.java

/*
 * Copyright (c) 1996, 1997, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code 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
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
package sun.rmi.transport.tcp;

import java.io.*;

/**
 * MultiplexInputStream manages receiving data over a connection managed
 * by a ConnectionMultiplexer object.  This object is responsible for
 * requesting more bytes of data as space in its internal buffer becomes
 * available.
 *
 * @author Peter Jones
 */
final class MultiplexInputStream extends InputStream {

    /** object managing multiplexed connection */
    private ConnectionMultiplexer manager;

    /** information about the connection this is the input stream for */
    private MultiplexConnectionInfo info;

    /** input buffer */
    private byte buffer[];

    /** number of real data bytes present in buffer */
    private int present = 0;

    /** current position to read from in input buffer */
    private int pos = 0;

    /** pending number of bytes this stream has requested */
    private int requested = 0;

    /** true if this connection has been disconnected */
    private boolean disconnected = false;

    /**
     * lock acquired to access shared variables:
     * buffer, present, pos, requested, & disconnected
     * WARNING:  Any of the methods manager.send*() should not be
     * invoked while this lock is held, since they could potentially
     * block if the underlying connection's transport buffers are
     * full, and the manager may need to acquire this lock to process
     * and consume data coming over the underlying connection.
     */
    private Object lock = new Object();

    /** level at which more data is requested when read past */
    private int waterMark;

    /** data structure for holding reads of one byte */
    private byte temp[] = new byte[1];

    /**
     * Create a new MultiplexInputStream for the given manager.
     * @param manager object that manages this connection
     * @param info structure for connection this stream reads from
     * @param bufferLength length of input buffer
     */
    MultiplexInputStream(
        ConnectionMultiplexer    manager,
        MultiplexConnectionInfo  info,
        int                      bufferLength)
    {
        this.manager = manager;
        this.info    = info;

        buffer = new byte[bufferLength];
        waterMark = bufferLength / 2;
    }

    /**
     * Read a byte from the connection.
     */
    public synchronized int read() throws IOException
    {
        int n = read(temp, 0, 1);
        if (n != 1)
            return -1;
        return temp[0] & 0xFF;
    }

    /**
     * Read a subarray of bytes from connection.  This method blocks for
     * at least one byte, and it returns the number of bytes actually read,
     * or -1 if the end of the stream was detected.
     * @param b array to read bytes into
     * @param off offset of beginning of bytes to read into
     * @param len number of bytes to read
     */
    public synchronized int read(byte b[], int off, int len) throws IOException
    {
        if (len <= 0)
            return 0;

        int moreSpace;
        synchronized (lock) {
            if (pos >= present)
                pos = present = 0;
            else if (pos >= waterMark) {
                System.arraycopy(buffer, pos, buffer, 0, present - pos);
                present -= pos;
                pos = 0;
            }
            int freeSpace = buffer.length - present;
            moreSpace = Math.max(freeSpace - requested, 0);
        }
        if (moreSpace > 0)
            manager.sendRequest(info, moreSpace);
        synchronized (lock) {
            requested += moreSpace;
            while ((pos >= present) && !disconnected) {
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                }
            }
            if (disconnected && pos >= present)
                return -1;

            int available = present - pos;
            if (len < available) {
                System.arraycopy(buffer, pos, b, off, len);
                pos += len;
                return len;
            }
            else {
                System.arraycopy(buffer, pos, b, off, available);
                pos = present = 0;
                // could send another request here, if len > available??
                return available;
            }
        }
    }

    /**
     * Return the number of bytes immediately available for reading.
     */
    public int available() throws IOException
    {
        synchronized (lock) {
            return present - pos;
        }
    }

    /**
     * Close this connection.
     */
    public void close() throws IOException
    {
        manager.sendClose(info);
    }

    /**
     * Receive bytes transmitted from connection at remote endpoint.
     * @param length number of bytes transmitted
     * @param in input stream with those bytes ready to be read
     */
    void receive(int length, DataInputStream in)
        throws IOException
    {
        /* TO DO: Optimize so that data received from stream can be loaded
         * directly into user's buffer if there is a pending read().
         */
        synchronized (lock) {
            if ((pos > 0) && ((buffer.length - present) < length)) {
                System.arraycopy(buffer, pos, buffer, 0, present - pos);
                present -= pos;
                pos = 0;
            }
            if ((buffer.length - present) < length)
                throw new IOException("Receive buffer overflow");
            in.readFully(buffer, present, length);
            present += length;
            requested -= length;
            lock.notifyAll();
        }
    }

    /**
     * Disconnect this stream from all connection activity.
     */
    void disconnect()
    {
        synchronized (lock) {
            disconnected = true;
            lock.notifyAll();
        }
    }
}