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