/*

 * 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();

        }

    }

}