/*
 * 
 * Copyright 2004 Sun Microsystems, Inc. All Rights Reserved.
 * 
 * The contents of this file are subject to the Sun Community Source
 * License v3.0/Jini Technology Specific Attachment v 1.0 (the
 * "License"). You may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 * http://wwws.sun.com/software/jini/licensing/SCSL3_JiniTSA1.html.
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Reference Code is Jini Technology Core Platform code, v2.0. The
 * Developer of the Reference Code is Sun Microsystems, Inc.
 * 
 * Contributor(s): Sun Microsystems, Inc.
 * 
 * The contents of this file comply with the Jini Technology Core
 * Platform Compatibility Kit, v2.0A.
 * 
 * Tester(s): Sun Microsystems, Inc.
 * 
 * Test Platform(s):
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Solaris(TM)SPARC/x86
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Linux (Intel x86)
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Windows (Intel Platform)
 * 
 */

package com.sun.jini.jeri.internal.mux;

import com.sun.jini.jeri.internal.runtime.HexDumpEncoder;
import com.sun.jini.thread.Executor;
import com.sun.jini.thread.GetThreadPoolAction;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.security.AccessController;
import java.util.BitSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Mux is the abstract superclass of both client-side and server-side
 * multiplexed connections.
 *
 * @author Sun Microsystems, Inc.
 **/
abstract class Mux {

    static final int CLIENT = 0;
    static final int SERVER = 1;

    static final int MAX_SESSION_ID = 0x7F;
    public static final int MAX_REQUESTS = MAX_SESSION_ID + 1;

    static final int NoOperation	= 0x00;	// 00000000
    static final int Shutdown		= 0x02; // 00000010
    static final int Ping		= 0x04; // 00000100
    static final int PingAck		= 0x06; // 00000110
    static final int Error		= 0x08; // 00001000
    static final int IncrementRation	= 0x10; // 0001***0
    static final int Abort		= 0x20; // 001000*0
    static final int Close		= 0x30; // 00110000
    static final int Acknowledgment	= 0x40; // 00100000
    static final int Data		= 0x80; // 100****0

    static final int IncrementRation_shift	= 0x0E;
    static final int Abort_partial		= 0x02;
    static final int Data_open			= 0x10;
    static final int Data_close			= 0x08;
    static final int Data_eof			= 0x04;
    static final int Data_ackRequired		= 0x02;

    static final int ClientConnectionHeader_negotiate	= 0x01;

    private static final byte[] magic = {
	(byte) 'J', (byte) 'm', (byte) 'u', (byte) 'x'	// 0x4A6D7578
    };

    private static final int VERSION = 0x01;

    /**
     * pool of threads for executing tasks in system thread group: used for
     * I/O (reader and writer) threads and other asynchronous tasks
     */
    private static final Executor systemThreadPool =
	(Executor) AccessController.doPrivileged(
	    new GetThreadPoolAction(false));

    /** mux logger */
    private static final Logger logger =
	Logger.getLogger("net.jini.jeri.connection.mux");

    final int role;
    final int initialInboundRation;
    final int maxFragmentSize;

    private final ConnectionIO connectionIO;

    /** lock guarding all mutable instance state (below) */
    final Object muxLock = new Object();

    int initialOutboundRation;		// set from remote connection header
    private boolean clientConnectionReady = false; // server header received
    boolean serverConnectionReady = false;	   // server header sent

    boolean muxDown = false;
    String muxDownMessage;
    Throwable muxDownCause;

    final BitSet busySessions = new BitSet();
    final Map sessions = new HashMap(5);

    private int expectedPingCookie = -1;

    /**
     * Constructs a new Mux instance for a connection accessible through
     * standard (blocking) I/O streams.
     */
    Mux(OutputStream out, InputStream in,
	int role, int initialInboundRation, int maxFragmentSize)
	throws IOException
    {
	this.role = role;
	if ((initialInboundRation & ~0x00FFFF00) != 0) {
	    throw new IllegalArgumentException(
		"illegal initial inbound ration: " +
		toHexString(initialInboundRation));
	}
	this.initialInboundRation = initialInboundRation;
	this.maxFragmentSize = maxFragmentSize;

	this.connectionIO = new StreamConnectionIO(this, out, in);
    }

    Mux(SocketChannel channel,
	int role, int initialInboundRation, int maxFragmentSize)
	throws IOException
    {
	this.role = role;
	if ((initialInboundRation & ~0x00FFFF00) != 0) {
	    throw new IllegalArgumentException(
		"illegal initial inbound ration: " +
		toHexString(initialInboundRation));
	}
	this.initialInboundRation = initialInboundRation;
	this.maxFragmentSize = maxFragmentSize;

	this.connectionIO = new SocketChannelConnectionIO(this, channel);
    }

    /**
     * Starts I/O processing.
     *
     * This method should be invoked only after this instance has
     * been completely initialized, so that subclasses will not
     * see uninitialized state.
     */
    public void start() throws IOException {
	if (role == CLIENT) {
	    readState = READ_SERVER_CONNECTION_HEADER;
	} else {
	    assert role == SERVER;
	    readState = READ_CLIENT_CONNECTION_HEADER;
	}

	connectionIO.start();

	if (role == CLIENT) {
	    asyncSendClientConnectionHeader();
	    synchronized (muxLock) {
		while (!muxDown && !clientConnectionReady) {
		    try {
			muxLock.wait();		// REMIND: timeout?
		    } catch (InterruptedException e) {
			String message =
			    "interrupt waiting for connection header";
			setDown(message, e);
			throw new IOException(message);
		    }
		}
		if (muxDown) {
		    IOException ioe = new IOException(muxDownMessage);
		    ioe.initCause(muxDownCause);
		    throw ioe;
		}
	    }
	}
    }

    /**
     * Handles indication that this multiplexed connection has
     * gone down, either through normal operation or failure.
     *
     * This method should be overridden by subclasses that want to
     * implement custom behavior when this connection has gone down.
     */
    protected void handleDown() {
    }

    /**
     * This method is invoked internally and is intended to be
     * overridden by subclasses.
     */
    void handleOpen(int sessionID) throws ProtocolException {
	throw new ProtocolException(
	    "remote endpoint attempted to open session");
    }

    /**
     *
     * This method is intended to be invoked by subclasses only.
     *
     * This method must ONLY be invoked while synchronized on muxLock
     * and while muxDown is false.
     */
    final void addSession(int sessionID, Session session) {
	assert Thread.holdsLock(muxLock);
	assert !muxDown;
	assert !busySessions.get(sessionID);
	assert sessions.get(new Integer(sessionID)) == null;

	busySessions.set(sessionID);
	sessions.put(new Integer(sessionID), session);
    }

    /**
     *
     * This method is intended to be invoked by this class and
     * subclasses only.
     *
     * This method MAY be invoked while synchronized on muxLock.
     */
    final void setDown(final String message, final Throwable cause) {
	synchronized (muxLock) {
	    if (muxDown) {
		return;
	    }
	    muxDown = true;
	    muxDownMessage = message;
	    muxDownCause = cause;
	    muxLock.notifyAll();
	}

	/*
	 * The following should be safe because we just left the
	 * synchonized block, and after setting the muxDown latch
	 * therein, no other thread should ever touch the "sessions"
	 * data structure.
	 *
	 * The following code is executed asynchronously in a separate
	 * thread to avoid deadlock, in case our caller holds muxLock,
	 * because individual session locks must never be acquired while
	 * holding muxLock.
	 */
	systemThreadPool.execute(new Runnable() {
	    public void run() {
		Iterator iter = sessions.values().iterator();
		while (iter.hasNext()) {
		    Session session = (Session) iter.next();
		    session.setDown(message, cause);
		}
	    }
	}, "Mux session shutdown");

	handleDown();
    }

    /**
     * Removes the identified session from the session table.
     *
     * This method is intended to be invoked by the associated Session
     * object only.
     */
    final void removeSession(int sessionID) {
	synchronized (muxLock) {
	    if (muxDown) {
		return;
	    }
	    assert busySessions.get(sessionID);
	    busySessions.clear(sessionID);
	    sessions.remove(new Integer(sessionID));
	}
    }

    /**
     * Sends the ClientConnectionHeader message for this connection.
     */
    final void asyncSendClientConnectionHeader() {
	assert role == CLIENT;

	ByteBuffer header = ByteBuffer.allocate(8);
	header.put(magic)
	      .put((byte) VERSION)
	      .putShort((short) (initialInboundRation >> 8))
	      .put((byte) 0)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends the ServerConnectionHeader message for this connection.
     */
    final void asyncSendServerConnectionHeader() {
	assert role == SERVER;

	ByteBuffer header = ByteBuffer.allocate(8);
	header.put(magic)
	      .put((byte) VERSION)
	      .putShort((short) (initialInboundRation >> 8))
	      .put((byte) 0)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends a NoOperation message with the contents of the supplied buffer
     * as the data.
     *
     * The "length" of the NoOperation message will be the number of bytes
     * remaining in the buffer, and the data sent will be the contents
     * of the buffer between its current position and its limit.  Or if
     * the buffer argument is null, "length" will simply be zero...
     * REMIND: split into two methods instead?
     *
     * The actual writing to the underlying connection, including access to
     * the buffer's content and other state, is asynchronous with the
     * invocation of this method; therefore, the supplied buffer must not
     * be mutated even after this method has returned.
     */
    final void asyncSendNoOperation(ByteBuffer buffer) {
	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) NoOperation)
	      .put((byte) 0);

	if (buffer != null) {
	    assert buffer.remaining() <= 0xFFFF;
	    header.putShort((short) buffer.remaining())
		  .flip();
	    connectionIO.asyncSend(header, buffer);
	} else {
	    header.putShort((short) 0)
		  .flip();
	    connectionIO.asyncSend(header);
	}
    }

    /**
     * Sends a Shutdown message with the UTF-8 encoding of the supplied
     * message as the data.  If message is null, then zero bytes of data
     * will be sent with the message header.
     */
    final void asyncSendShutdown(String message) {
	ByteBuffer data = (message != null ?
			   getUTF8BufferFromString(message) : null);

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Shutdown)
	      .put((byte) 0);

	if (data != null) {
	    assert data.remaining() <= 0xFFFF;
	    header.putShort((short) data.remaining())
		  .flip();
	    connectionIO.asyncSend(header, data);
	} else {
	    header.putShort((short) 0)
		  .flip();
	    connectionIO.asyncSend(header);
	}
    }

    /**
     * Sends a Ping message with the specified "cookie".
     */
    final void asyncSendPing(int cookie) {
	assert cookie >= 0 && cookie <= 0xFFFF;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Ping)
              .put((byte) 0)
	      .putShort((short) cookie)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends a PingAck message with the specified "cookie".
     */
    final void asyncSendPingAck(int cookie) {
	assert cookie >= 0 && cookie <= 0xFFFF;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) PingAck)
	      .put((byte) 0)
	      .putShort((short) cookie)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends an Error message with the UTF-8 encoding of the supplied
     * message as the data.  If message is null, then zero bytes of data
     * will be sent with the message header.
     */
    final void asyncSendError(String message) {
	ByteBuffer data = (message != null ?
			   getUTF8BufferFromString(message) : null);

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Error)
	      .put((byte) 0);

	if (data != null) {
	    assert data.remaining() <= 0xFFFF;
	    header.putShort((short) data.remaining())
		  .flip();
	    connectionIO.asyncSend(header, data);
	} else {
	    header.putShort((short) 0)
		  .flip();
	    connectionIO.asyncSend(header);
	}
    }

    /**
     * Sends an Error message with the UTF-8 encoding of the supplied
     * message as the data.  If message is null, then zero bytes of data
     * will be sent with the message header.
     */
    final IOFuture futureSendError(String message) {
	ByteBuffer data = getUTF8BufferFromString(message);

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Error)
	      .put((byte) 0);

	assert data.remaining() <= 0xFFFF;
	header.putShort((short) data.remaining())
	      .flip();
	return connectionIO.futureSend(header, data);
    }

    /**
     * Sends an IncrementRation message for the specified "sessionID" and
     * the specified "increment".
     */
    final void asyncSendIncrementRation(/*****int op, *****/int sessionID,
					int increment)
    {
	final int op = IncrementRation;
//	assert (op & 0xF1) == IncrementRation;	// validate operation code
//	assert (op & 0xE0) == 0;		// NYI: support use of shift
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;
	assert increment >= 0 && increment <= 0xFFFF;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) op)
	      .put((byte) sessionID)
	      .putShort((short) increment)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends an Abort message for the specified "sessionID" with the contents
     * of the specified buffer as the data.
     *
     * The "length" of the Abort message will be the number of bytes
     * remaining in the buffer, and the data sent will be the contents
     * of the buffer between its current position and its limit.  Or if
     * the buffer argument is null, "length" will simply be zero...
     * REMIND: split into two methods instead?
     *
     * For efficiency, the caller is responsible for pre-computing the first
     * byte of the message, including any control flags if appropriate.
     */
    final void asyncSendAbort(int op, int sessionID, ByteBuffer data) {
	assert (op & 0xFD) == Abort;		// validate operation code
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) op)
	      .put((byte) sessionID);

	if (data != null) {
	    assert data.remaining() <= 0xFFFF;
	    header.putShort((short) data.remaining())
		  .flip();
	    connectionIO.asyncSend(header, data);
	} else {
	    header.putShort((short) 0)
		  .flip();
	    connectionIO.asyncSend(header);
	}
    }

    /**
     * Sends a Close message for the specified "sessionID".
     */
    final void asyncSendClose(int sessionID) {
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Close)
	      .put((byte) sessionID)
	      .putShort((short) 0)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends an Acknowledgment message for the specified "sessionID".
     */
    final void asyncSendAcknowledgment(int sessionID) {
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) Acknowledgment)
	      .put((byte) sessionID)
	      .putShort((short) 0)
	      .flip();
	connectionIO.asyncSend(header);
    }

    /**
     * Sends a Data message for the specified "sessionID" with the contents
     * of the supplied buffer as the data.
     *
     * The "length" of the Data message will be the number of bytes
     * remaining in the buffer, and the data sent will be the contents
     * of the buffer between its current position and its limit.  Or if
     * the buffer argument is null, "length" will simply be zero...
     * REMIND: split into two methods instead?
     *
     * For efficiency, the caller is responsible for pre-computing the first
     * byte of the Data message, including any control flags if appropriate.
     *
     * The actual writing to the underlying connection, including access to
     * the buffer's content and other state, is asynchronous with the
     * invocation of this method; therefore, the supplied buffer must not
     * be mutated even after this method has returned.
     */
    final void asyncSendData(int op, int sessionID, ByteBuffer data) {
	assert (op & 0xE1) == Data;	// validate operation code
	assert (op & Data_eof) != 0 ||	// close and ackRequired require eof
	    (op & Data_close & Data_ackRequired) == 0;
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) op)
	      .put((byte) sessionID);

	if (data != null) {
	    assert data.remaining() <= 0xFFFF;
	    header.putShort((short) data.remaining())
		  .flip();
	    connectionIO.asyncSend(header, data);
	} else {
	    header.putShort((short) 0)
		  .flip();
	    connectionIO.asyncSend(header);
	}
    }

    /**
     * Sends a Data message for the specified sessionID with the contents
     * of the supplied buffer as the data.
     *
     * The "length" of the Data message will be the number of bytes
     * remaining in the buffer, and the data sent will be the contents
     * of the buffer between its current position and its limit.
     *
     * For efficiency, the caller is responsible for pre-computing the first
     * byte of the Data message, including any control flags if appropriate.
     *
     * The actual writing to the underlying connection, including access to
     * the buffer's content and other state, is asynchronous with the
     * invocation of this method; therefore, the supplied buffer must not
     * be mutated even after this method has returned, until it is guaranteed
     * that use of the buffer has completed.
     *
     * The returned IOFuture object can be used to wait until the write has
     * definitely completed (or will definitely not complete due to some
     * failure).  After the write has completed, the buffer's position will
     * have been incremented to its limit (which will not have changed).
     */
    final IOFuture futureSendData(int op, int sessionID, ByteBuffer data) {
	assert (op & 0xE1) == Data;	// verify operation code
	assert (op & Data_eof) != 0 ||	// close and ackRequired require eof
	    (op & Data_close & Data_ackRequired) == 0;
	assert sessionID >= 0 && sessionID <= MAX_SESSION_ID;
	assert data.remaining() <= 0xFFFF;

	ByteBuffer header = ByteBuffer.allocate(4);
	header.put((byte) op)
	      .put((byte) sessionID)
	      .putShort((short) data.remaining())
	      .flip();
	return connectionIO.futureSend(header, data);
    }

    /*
     * read states
     */
    private static final int READ_CLIENT_CONNECTION_HEADER	= 0;
    private static final int READ_SERVER_CONNECTION_HEADER	= 1;
    private static final int READ_MESSAGE_HEADER		= 2;
    private static final int READ_MESSAGE_BODY			= 3;

    /*
     * current read state lock and variables
     */
    private final Object readStateLock = new Object();
    private int readState;
    private int currentOp;
    private int currentSessionID;
    private int currentLengthRemaining;
    private ByteBuffer currentDataBuffer = null;

    void processIncomingData(ByteBuffer buffer) throws ProtocolException {
	buffer.flip();	// process data that has been read into buffer
	assert buffer.hasRemaining();

	synchronized (readStateLock) {
	  stateLoop:
	    do {
		switch (readState) {
		  case READ_CLIENT_CONNECTION_HEADER:
		    if (!readClientConnectionHeader(buffer)) {
			break stateLoop;
		    }
		    break;

		  case READ_SERVER_CONNECTION_HEADER:
		    if (!readServerConnectionHeader(buffer)) {
			break stateLoop;
		    }
		    break;

		  case READ_MESSAGE_HEADER:
		    if (!readMessageHeader(buffer)) {
			break stateLoop;
		    }
		    break;

		  case READ_MESSAGE_BODY:
		    if (!readMessageBody(buffer)) {
			break stateLoop;
		    }
		    break;

		  default:
		    throw new AssertionError();
		}
	    } while (buffer.hasRemaining());
	}

	buffer.compact();
    }

    private boolean readClientConnectionHeader(ByteBuffer buffer)
	throws ProtocolException
    {
	assert role == SERVER;

	validatePartialMagicNumber(buffer);
	if (buffer.remaining() < 8) {
	    return false;		// wait for complete header to arrive
	}
	int headerPosition = buffer.position();
	buffer.position(headerPosition + 4);	// skip header already checked
	int version = (buffer.get() & 0xFF);
	int ration = (buffer.getShort() & 0xFFFF) << 8;
	int flags = (buffer.get() & 0xFF);
	boolean negotiate = (flags & ClientConnectionHeader_negotiate) != 0;

	synchronized (muxLock) {
	    initialOutboundRation = ration;
	    asyncSendServerConnectionHeader();

	    if (version == 0) {
		throw new ProtocolException(
		    "bad protocol version: " + version);
	    }
	    if (version > VERSION) {
		if (!negotiate) {
		    setDown("unsupported protocol version: " + version, null);
		    throw new ProtocolException(
			"unsupported protocol version: " + version);
		}
	    }

	    serverConnectionReady = true;
	}

	readState = READ_MESSAGE_HEADER;
	return true;
    }

    private boolean readServerConnectionHeader(ByteBuffer buffer)
	throws ProtocolException
    {
	assert role == CLIENT;

	validatePartialMagicNumber(buffer);

	if (buffer.remaining() < 8) {
	    return false;
	}
	int headerPosition = buffer.position();
	buffer.position(headerPosition + 4);	// skip header already checked
	int version = (buffer.get() & 0xFF);
	int ration = (buffer.getShort() & 0xFFFF) << 8;
	int flags = (buffer.get() & 0xFF);

	synchronized (muxLock) {
	    initialOutboundRation = ration;

	    if (version == 0) {
		throw new ProtocolException(
		    "bad protocol version: " + version);
	    }
	    if (version > VERSION) {
		throw new ProtocolException(
		    "unexpected protocol version: " + version);
	    }

	    clientConnectionReady = true;
	    muxLock.notifyAll();
	}

	readState = READ_MESSAGE_HEADER;
	return true;
    }

    private void validatePartialMagicNumber(ByteBuffer buffer)
	throws ProtocolException
    {
	if (buffer.remaining() > 0) {
	    byte[] temp = new byte[Math.min(buffer.remaining(), magic.length)];
	    buffer.mark();
	    buffer.get(temp);
	    buffer.reset();
	    for (int i = 0; i < temp.length; i++) {
		if (temp[i] != magic[i]) {
		    setDown((role == CLIENT ? "server" : "client") +
			" sent bad magic number: " + toHexString(temp), null);
		    throw new ProtocolException("bad magic number: " +
						toHexString(temp));
		}
	    }
	}
    }

    private boolean readMessageHeader(ByteBuffer buffer)
	throws ProtocolException
    {
	if (buffer.remaining() < 4) {
	    return false;		// wait for complete header to arrive
	}
	int headerPosition = buffer.position();
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST,
		       "message header: " +
		       toHexString(buffer.getInt(headerPosition)));
	}

	int op = (buffer.get() & 0xFF);
	if ((op & 0xE1) == Data) {
	    int sessionID = (buffer.get() & 0xFF);
	    if (sessionID > MAX_SESSION_ID) {
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    currentOp = op;
	    currentSessionID = sessionID;
	    currentLengthRemaining = (buffer.getShort() & 0xFFFF);
	    if (currentLengthRemaining > 0) {
		currentDataBuffer =
		    ByteBuffer.allocate(currentLengthRemaining);
		readState = READ_MESSAGE_BODY;
	    } else {
		dispatchCurrentMessage();
	    }
	    return true;

	} else if ((op & 0xF1) == IncrementRation) {
	    int sessionID = (buffer.get() & 0xFF);
	    if (sessionID > MAX_SESSION_ID) {
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    int increment = (buffer.getShort() & 0xFFFF);
	    int shift = op & IncrementRation_shift;
	    increment <<= shift;
	    handleIncrementRation(sessionID, increment);
	    return true;

	} else if ((op & 0xFD) == Abort) {
	    int sessionID = (buffer.get() & 0xFF);
	    if (sessionID > MAX_SESSION_ID) {
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    currentOp = op;
	    currentSessionID = sessionID;
	    currentLengthRemaining = (buffer.getShort() & 0xFFFF);
	    if (currentLengthRemaining > 0) {
		currentDataBuffer =
		    ByteBuffer.allocate(currentLengthRemaining);
		readState = READ_MESSAGE_BODY;
	    } else {
		dispatchCurrentMessage();
	    }
	    return true;

	}
	switch (op) {
	  case NoOperation: {
	    if (buffer.get() != 0) {	// ignore sign extension
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    currentOp = op;
	    currentLengthRemaining = (buffer.getShort() & 0xFFFF);
	    currentDataBuffer = null;	// ignore data for NoOperation
	    if (currentLengthRemaining > 0) {
		readState = READ_MESSAGE_BODY;
	    } else {
		dispatchCurrentMessage();
	    }
	    return true;
	  }

	  case Shutdown: {
	    if (buffer.get() != 0) {	// ignore sign extension
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    currentOp = op;
	    currentLengthRemaining = (buffer.getShort() & 0xFFFF);
	    if (currentLengthRemaining > 0) {
		currentDataBuffer =
		    ByteBuffer.allocate(currentLengthRemaining);
		readState = READ_MESSAGE_BODY;
	    } else {
		dispatchCurrentMessage();
	    }
	    return true;
	  }

	  case Ping: {
	    if (buffer.get() != 0) {	// ignore sign extension
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    int cookie = (buffer.getShort() & 0xFFFF);
	    handlePing(cookie);
	    return true;
	  }

	  case PingAck: {
	    if (buffer.get() != 0) {	// ignore sign extension
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    int cookie = (buffer.getShort() & 0xFFFF);
	    handlePingAck(cookie);
	    return true;
	  }

	  case Error: {
	    if (buffer.get() != 0) {	// ignore sign extension
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    currentOp = op;
	    currentLengthRemaining = (buffer.getShort() & 0xFFFF);
	    if (currentLengthRemaining > 0) {
		currentDataBuffer =
		    ByteBuffer.allocate(currentLengthRemaining);
		readState = READ_MESSAGE_BODY;
	    } else {
		dispatchCurrentMessage();
	    }
	    return true;
	  }

	  case Close: {
	    int sessionID = (buffer.get() & 0xFF);
	    if (sessionID > MAX_SESSION_ID ||
		buffer.getShort() != 0)		// ignore sign extension
	    {
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    handleClose(sessionID);
	    return true;
	  }

	  case Acknowledgment: {
	    int sessionID = (buffer.get() & 0xFF);
	    if (sessionID > MAX_SESSION_ID ||
		buffer.getShort() != 0)		// ignore sign extension
	    {
		throw new ProtocolException("bad message header: " +
		    toHexString(buffer.getInt(headerPosition)));
	    }
	    handleAcknowledgment(sessionID);
	    return true;
	  }

	  default:
	    throw new ProtocolException("bad message header: " +
		toHexString(buffer.getInt(headerPosition)));
	}
    }

    private boolean readMessageBody(ByteBuffer buffer)
	throws ProtocolException
    {
	assert currentLengthRemaining > 0;
	assert currentDataBuffer == null ||
	    currentDataBuffer.remaining() == currentLengthRemaining;

	if (buffer.remaining() > currentLengthRemaining) {
	    int origLimit = buffer.limit();
	    buffer.limit(buffer.position() + currentLengthRemaining);
	    if (currentDataBuffer != null) {
		currentDataBuffer.put(buffer);
	    } else {
		buffer.position(buffer.position() + currentLengthRemaining);
	    }
	    currentLengthRemaining = 0;
	    buffer.limit(origLimit);
	} else {
	    currentLengthRemaining -= buffer.remaining();
	    if (currentDataBuffer != null) {
		currentDataBuffer.put(buffer);
	    } else {
		buffer.position(buffer.limit());
	    }
	}

	if (currentLengthRemaining > 0) {
	    return false;
	} else {
	    currentDataBuffer.flip();
	    dispatchCurrentMessage();
	    currentDataBuffer = null;		// don't let this linger
	    readState = READ_MESSAGE_HEADER;
	    return true;
	}
    }

    private void dispatchCurrentMessage() throws ProtocolException {
	assert currentDataBuffer == null || currentDataBuffer.hasRemaining();

	int op = currentOp;
	if ((op & 0xE1) == Data) {
	    boolean open	= (op & Data_open) != 0;
	    boolean close	= (op & Data_close) != 0;
	    boolean eof		= (op & Data_eof) != 0;
	    boolean ackRequired	= (op & Data_ackRequired) != 0;
	    handleData(currentSessionID, open, close, eof, ackRequired,
		       (currentDataBuffer != null ?
			currentDataBuffer : ByteBuffer.allocate(0)));
	    return;

	} else if ((op & 0xFD) == Abort) {
	    boolean partial = (op & Abort_partial) != 0;
	    handleAbort(currentSessionID, partial,
			(currentDataBuffer != null ?
			 getStringFromUTF8Buffer(currentDataBuffer) : ""));
	    return;

	}
	switch (op) {
	  case NoOperation:
	    handleNoOperation();
	    return;

	  case Shutdown:
	    handleShutdown(currentDataBuffer != null ?
			   getStringFromUTF8Buffer(currentDataBuffer) : "");
	    return;

	  case Error:
	    handleError(currentDataBuffer != null ?
			getStringFromUTF8Buffer(currentDataBuffer) : "");
	    return;

	  default:
	    throw new AssertionError(Integer.toHexString((byte) op));
	}
    }

    private void handleNoOperation() throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "NoOperation");
	}

	// do nothing
    }

    private void handleShutdown(String message) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "Shutdown");
	}

	if (role != CLIENT) {
	    throw new ProtocolException("Shutdown sent by client");
	}
	setDown("mux connection shut down gracefully", null);
	throw new ProtocolException("received Shutdown message");
    }

    private void handlePing(int cookie) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "Ping: cookie=" + cookie);
	}

	asyncSendPingAck(cookie);
    }

    private void handlePingAck(int cookie) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "PingAck: cookie=" + cookie);
	}

	synchronized (muxLock) {
	    if (cookie != expectedPingCookie) {
		throw new ProtocolException(
		    "unexpected ping cookie: " + cookie);
	    } else {
		expectedPingCookie = -1;
		// NYI: rest of ping machinery
	    }
	}
    }

    private void handleError(String message) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "Error");
	}

	setDown((role == CLIENT ? "server" : "client") +
		" reported protocol error: " + message, null);
	throw new ProtocolException("received Error message");
    }

    private void handleIncrementRation(int sessionID, int increment)
	throws ProtocolException
    {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST,
		"IncrementRation: sessionID=" + sessionID +
		",increment=" + increment);
	}

	getSession(sessionID).handleIncrementRation(increment);
    }

    private void handleAbort(int sessionID, boolean partial, String message)
	throws ProtocolException
    {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST,
		"Abort: sessionID=" + sessionID +
		",partial=" + partial);
	}

	getSession(sessionID).handleAbort(partial);
    }

    private void handleClose(int sessionID) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "Close: sessionID=" + sessionID);
	}

	getSession(sessionID).handleClose();
    }

    private void handleAcknowledgment(int sessionID) throws ProtocolException {
	if (logger.isLoggable(Level.FINEST)) {
	    logger.log(Level.FINEST, "Acknowledgment: sessionID=" + sessionID);
	}

	getSession(sessionID).handleAcknowledgment();
    }

    private void handleData(int sessionID, boolean open, boolean close,
			    boolean eof, boolean ackRequired, ByteBuffer data)
	throws ProtocolException
    {
	if (logger.isLoggable(Level.FINEST)) {
	    int length = data.remaining();
	    HexDumpEncoder encoder = new HexDumpEncoder();
	    byte[] bytes = new byte[data.remaining()];
	    data.mark();
	    data.get(bytes);
	    data.reset();
	    logger.log(Level.FINEST,
		"Data: sessionID=" + sessionID +
		(open ? ",open" : "") +
		(close ? ",close" : "") +
		(eof ? ",eof" : "") +
		(ackRequired ? ",ackRequired" : "") +
		",length=" + length +
		(length > 0 ? ",data=\n" + encoder.encode(bytes) : ""));
	}

	if (!eof && (close || ackRequired)) {
	    throw new ProtocolException("Data: eof=" + eof +
					",close=" + close +
					",ackRequired=" + ackRequired);
	}

	if (open) {
	    handleOpen(sessionID);
	}

	getSession(sessionID).handleData(data, eof, close, ackRequired);
    }

    private Session getSession(int sessionID) throws ProtocolException {
	synchronized (muxLock) {
	    if (!busySessions.get(sessionID)) {
		throw new ProtocolException(
		    "inactive sessionID: " + sessionID);
	    }
	    return (Session) sessions.get(new Integer(sessionID));
	}
    }

    private static ByteBuffer getUTF8BufferFromString(String s) {
	CharsetEncoder encoder = Charset.forName("UTF-8").newEncoder();
	try {
	    return encoder.encode(CharBuffer.wrap(s));
	} catch (CharacterCodingException e) {
	    return null;
	}
    }

    private static String getStringFromUTF8Buffer(ByteBuffer buffer) {
	CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
	try {
	    return decoder.decode(buffer).toString();
	} catch (CharacterCodingException e) {
	    return "(error decoding UTF-8 message: " + e.toString() + ")";
	}
    }

    private static String toHexString(byte x) {
	char[] buf = new char[2];
	buf[0] = toHexChar((x >> 4) & 0xF);
	buf[1] = toHexChar(x & 0xF);
	return new String(buf);
    }

    private static String toHexString(int x) {
	char[] buf = new char[8];
	for (int i = 0; i < 8; i++) {
	    buf[i] = toHexChar((x >> ((7 - i) * 4)) & 0xF);
	}
	return new String(buf);
    }

    private static String toHexString(byte[] b) {
	char[] buf = new char[b.length * 2];
	int j = 0;
	for (int i = 0; i < b.length; i++) {
	    buf[j++] = toHexChar((b[i] >> 4) & 0xF);
	    buf[j++] = toHexChar(b[i] & 0xF);
	}
	return new String(buf);
    }

    private static char toHexChar(int x) {
	return x < 10 ? (char) ('0' + x) : (char) ('A' - 10 + x);
    }
}
