Java example source code file (ConnectionMultiplexer.java)
The ConnectionMultiplexer.java Java example source code
/*
* Copyright (c) 1996, 2013, 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.*;
import java.util.*;
import java.rmi.server.LogStream;
import sun.rmi.runtime.Log;
/**
* ConnectionMultiplexer manages the transparent multiplexing of
* multiple virtual connections from one endpoint to another through
* one given real connection to that endpoint. The input and output
* streams for the the underlying real connection must be supplied.
* A callback object is also supplied to be informed of new virtual
* connections opened by the remote endpoint. After creation, the
* run() method must be called in a thread created for demultiplexing
* the connections. The openConnection() method is called to
* initiate a virtual connection from this endpoint.
*
* @author Peter Jones
*/
@SuppressWarnings("deprecation")
final class ConnectionMultiplexer {
/** "multiplex" log level */
static int logLevel = LogStream.parseLevel(getLogLevel());
private static String getLogLevel() {
return java.security.AccessController.doPrivileged(
new sun.security.action.GetPropertyAction("sun.rmi.transport.tcp.multiplex.logLevel"));
}
/* multiplex system log */
static final Log multiplexLog =
Log.getLog("sun.rmi.transport.tcp.multiplex",
"multiplex", ConnectionMultiplexer.logLevel);
/** multiplexing protocol operation codes */
private final static int OPEN = 0xE1;
private final static int CLOSE = 0xE2;
private final static int CLOSEACK = 0xE3;
private final static int REQUEST = 0xE4;
private final static int TRANSMIT = 0xE5;
/** object to notify for new connections from remote endpoint */
private TCPChannel channel;
/** input stream for underlying single connection */
private InputStream in;
/** output stream for underlying single connection */
private OutputStream out;
/** true if underlying connection originated from this endpoint
(used for generating unique connection IDs) */
private boolean orig;
/** layered stream for reading formatted data from underlying connection */
private DataInputStream dataIn;
/** layered stream for writing formatted data to underlying connection */
private DataOutputStream dataOut;
/** table holding currently open connection IDs and related info */
private Hashtable<Integer, MultiplexConnectionInfo> connectionTable = new Hashtable<>(7);
/** number of currently open connections */
private int numConnections = 0;
/** maximum allowed open connections */
private final static int maxConnections = 256;
/** ID of last connection opened */
private int lastID = 0x1001;
/** true if this mechanism is still alive */
private boolean alive = true;
/**
* Create a new ConnectionMultiplexer using the given underlying
* input/output stream pair. The run method must be called
* (possibly on a new thread) to handle the demultiplexing.
* @param channel object to notify when new connection is received
* @param in input stream of underlying connection
* @param out output stream of underlying connection
* @param orig true if this endpoint intiated the underlying
* connection (needs to be set differently at both ends)
*/
public ConnectionMultiplexer(
TCPChannel channel,
InputStream in,
OutputStream out,
boolean orig)
{
this.channel = channel;
this.in = in;
this.out = out;
this.orig = orig;
dataIn = new DataInputStream(in);
dataOut = new DataOutputStream(out);
}
/**
* Process multiplexing protocol received from underlying connection.
*/
public void run() throws IOException
{
try {
int op, id, length;
MultiplexConnectionInfo info;
while (true) {
// read next op code from remote endpoint
op = dataIn.readUnsignedByte();
switch (op) {
// remote endpoint initiating new connection
case OPEN:
id = dataIn.readUnsignedShort();
if (multiplexLog.isLoggable(Log.VERBOSE)) {
multiplexLog.log(Log.VERBOSE, "operation OPEN " + id);
}
info = connectionTable.get(id);
if (info != null)
throw new IOException(
"OPEN: Connection ID already exists");
info = new MultiplexConnectionInfo(id);
info.in = new MultiplexInputStream(this, info, 2048);
info.out = new MultiplexOutputStream(this, info, 2048);
synchronized (connectionTable) {
connectionTable.put(id, info);
++ numConnections;
}
sun.rmi.transport.Connection conn;
conn = new TCPConnection(channel, info.in, info.out);
channel.acceptMultiplexConnection(conn);
break;
// remote endpoint closing connection
case CLOSE:
id = dataIn.readUnsignedShort();
if (multiplexLog.isLoggable(Log.VERBOSE)) {
multiplexLog.log(Log.VERBOSE, "operation CLOSE " + id);
}
info = connectionTable.get(id);
if (info == null)
throw new IOException(
"CLOSE: Invalid connection ID");
info.in.disconnect();
info.out.disconnect();
if (!info.closed)
sendCloseAck(info);
synchronized (connectionTable) {
connectionTable.remove(id);
-- numConnections;
}
break;
// remote endpoint acknowledging close of connection
case CLOSEACK:
id = dataIn.readUnsignedShort();
if (multiplexLog.isLoggable(Log.VERBOSE)) {
multiplexLog.log(Log.VERBOSE,
"operation CLOSEACK " + id);
}
info = connectionTable.get(id);
if (info == null)
throw new IOException(
"CLOSEACK: Invalid connection ID");
if (!info.closed)
throw new IOException(
"CLOSEACK: Connection not closed");
info.in.disconnect();
info.out.disconnect();
synchronized (connectionTable) {
connectionTable.remove(id);
-- numConnections;
}
break;
// remote endpoint declaring additional bytes receivable
case REQUEST:
id = dataIn.readUnsignedShort();
info = connectionTable.get(id);
if (info == null)
throw new IOException(
"REQUEST: Invalid connection ID");
length = dataIn.readInt();
if (multiplexLog.isLoggable(Log.VERBOSE)) {
multiplexLog.log(Log.VERBOSE,
"operation REQUEST " + id + ": " + length);
}
info.out.request(length);
break;
// remote endpoint transmitting data packet
case TRANSMIT:
id = dataIn.readUnsignedShort();
info = connectionTable.get(id);
if (info == null)
throw new IOException("SEND: Invalid connection ID");
length = dataIn.readInt();
if (multiplexLog.isLoggable(Log.VERBOSE)) {
multiplexLog.log(Log.VERBOSE,
"operation TRANSMIT " + id + ": " + length);
}
info.in.receive(length, dataIn);
break;
default:
throw new IOException("Invalid operation: " +
Integer.toHexString(op));
}
}
} finally {
shutDown();
}
}
/**
* Initiate a new multiplexed connection through the underlying
* connection.
*/
public synchronized TCPConnection openConnection() throws IOException
{
// generate ID that should not be already used
// If all possible 32768 IDs are used,
// this method will block searching for a new ID forever.
int id;
do {
lastID = (++ lastID) & 0x7FFF;
id = lastID;
// The orig flag (copied to the high bit of the ID) is used
// to have two distinct ranges to choose IDs from for the
// two endpoints.
if (orig)
id |= 0x8000;
} while (connectionTable.get(id) != null);
// create multiplexing streams and bookkeeping information
MultiplexConnectionInfo info = new MultiplexConnectionInfo(id);
info.in = new MultiplexInputStream(this, info, 2048);
info.out = new MultiplexOutputStream(this, info, 2048);
// add to connection table if multiplexer has not died
synchronized (connectionTable) {
if (!alive)
throw new IOException("Multiplexer connection dead");
if (numConnections >= maxConnections)
throw new IOException("Cannot exceed " + maxConnections +
" simultaneous multiplexed connections");
connectionTable.put(id, info);
++ numConnections;
}
// inform remote endpoint of new connection
synchronized (dataOut) {
try {
dataOut.writeByte(OPEN);
dataOut.writeShort(id);
dataOut.flush();
} catch (IOException e) {
multiplexLog.log(Log.BRIEF, "exception: ", e);
shutDown();
throw e;
}
}
return new TCPConnection(channel, info.in, info.out);
}
/**
* Shut down all connections and clean up.
*/
public void shutDown()
{
// inform all associated streams
synchronized (connectionTable) {
// return if multiplexer already officially dead
if (!alive)
return;
alive = false;
Enumeration<MultiplexConnectionInfo> enum_ =
connectionTable.elements();
while (enum_.hasMoreElements()) {
MultiplexConnectionInfo info = enum_.nextElement();
info.in.disconnect();
info.out.disconnect();
}
connectionTable.clear();
numConnections = 0;
}
// close underlying connection, if possible (and not already done)
try {
in.close();
} catch (IOException e) {
}
try {
out.close();
} catch (IOException e) {
}
}
/**
* Send request for more data on connection to remote endpoint.
* @param info connection information structure
* @param len number of more bytes that can be received
*/
void sendRequest(MultiplexConnectionInfo info, int len) throws IOException
{
synchronized (dataOut) {
if (alive && !info.closed)
try {
dataOut.writeByte(REQUEST);
dataOut.writeShort(info.id);
dataOut.writeInt(len);
dataOut.flush();
} catch (IOException e) {
multiplexLog.log(Log.BRIEF, "exception: ", e);
shutDown();
throw e;
}
}
}
/**
* Send packet of requested data on connection to remote endpoint.
* @param info connection information structure
* @param buf array containing bytes to send
* @param off offset of first array index of packet
* @param len number of bytes in packet to send
*/
void sendTransmit(MultiplexConnectionInfo info,
byte buf[], int off, int len) throws IOException
{
synchronized (dataOut) {
if (alive && !info.closed)
try {
dataOut.writeByte(TRANSMIT);
dataOut.writeShort(info.id);
dataOut.writeInt(len);
dataOut.write(buf, off, len);
dataOut.flush();
} catch (IOException e) {
multiplexLog.log(Log.BRIEF, "exception: ", e);
shutDown();
throw e;
}
}
}
/**
* Inform remote endpoint that connection has been closed.
* @param info connection information structure
*/
void sendClose(MultiplexConnectionInfo info) throws IOException
{
info.out.disconnect();
synchronized (dataOut) {
if (alive && !info.closed)
try {
dataOut.writeByte(CLOSE);
dataOut.writeShort(info.id);
dataOut.flush();
info.closed = true;
} catch (IOException e) {
multiplexLog.log(Log.BRIEF, "exception: ", e);
shutDown();
throw e;
}
}
}
/**
* Acknowledge remote endpoint's closing of connection.
* @param info connection information structure
*/
void sendCloseAck(MultiplexConnectionInfo info) throws IOException
{
synchronized (dataOut) {
if (alive && !info.closed)
try {
dataOut.writeByte(CLOSEACK);
dataOut.writeShort(info.id);
dataOut.flush();
info.closed = true;
} catch (IOException e) {
multiplexLog.log(Log.BRIEF, "exception: ", e);
shutDown();
throw e;
}
}
}
/**
* Shut down connection upon finalization.
*/
protected void finalize() throws Throwable
{
super.finalize();
shutDown();
}
}
Other Java examples (source code examples)Here is a short list of links related to this Java ConnectionMultiplexer.java source code file: |
The search page
Other Java source code examples at this package level
Click here to learn more about this project
