/*
 *        JacORB - a free Java ORB
 *
 *   Copyright (C) 1997-98  Gerald Brose.
 *
 *   This library is free software; you can redistribute it and/or
 *   modify it under the terms of the GNU Library General Public
 *   License as published by the Free Software Foundation; either
 *   version 2 of the License, or (at your option) any later version.
 *
 *   This library 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
 *   Library General Public License for more details.
 *
 *   You should have received a copy of the GNU Library General Public
 *   License along with this library; if not, write to the Free
 *   Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

package jacorb.orb;

/**
 * This class manages connections.<br>
 * It writes to an OutputStream or receives from an InputStream,<br>
 * converting type representations from internal to external or<br>
 * vice versa<p>
 *
 * @author Gerald Brose, FU Berlin
 * @version $Id: Connection.java,v 1.42 2000/04/04 14:05:39 brose Exp $
 *
 */


import java.io.*;
import java.net.*;
import java.util.*;
import jacorb.orb.giop.*;
import jacorb.util.*;

public final class Connection 
    implements ClientSideConnection, ServerSideConnection
{
    private int id_count = 0;
    private String connection_info = null;
    private BufferedInputStream in_stream;
    BufferedOutputStream out_stream;

    protected jacorb.orb.ORB orb;

    /** write lock */
    public Object writeLock = new Object();


    // repID -> object
    private Hashtable objects = new Hashtable();

  //contains RequestOutputStreams instead of byte[]
  private Hashtable buffers = new Hashtable();

    private Hashtable replies = new Hashtable();
    private boolean littleEndian;
    private boolean is_server;
    private boolean is_ssl;
	
    /**
     * Connection OSF character formats, if zero for client side connection
     * then prefered tcs was not computed yet; use TCS for zero testing.
     * Note that zero values are invalid when tcsNegotiated=true.
     * It's protected because it's directly read by CDR classes defined
     * in this package.
     * @author devik
     */

    protected int TCS=0,TCSW=0;	
    /**
     * Client's information whether the codeSet has been already sent to server.
     * It's important because there is possibility that we add the context
     * to new request but the request will never be sent because of exception
     * during marshaling in stub. TCS will be considered sent at point when
     * at least one request with tcs context was sent successfully. Also
     * note there is possibility that another requests with tcs context
     * can be pending and also sent AFTER tcsNegotiated was set to true. But it's
     * not problem because server side will ignore all tcs contexts recieved
     * after first one (but OMG doc is not clear here).<br>
     * When set for server side connection (is_server=true) it means that
     * at least one request with tcs context was recieved and tcs information
     * is valid.
     * @author devik
     */

    private boolean tcsNegotiated=false;
	
    /**
     * IIOP version active on the channel.
     * @author devik
     */

    protected org.omg.IIOP.Version IIOPVersion=new org.omg.IIOP.Version((byte)1,(byte)0);

    /* how many clients use this connection? */
    private int client_count = 0;

    private Socket mysock = null;
    private ReplyReceptor rep;
    private byte [] header = new byte[ Messages.MSG_HEADER_SIZE ];

    ORB getORB()
    {
	return orb;
    }


    /**
     * Called by Delegate or setServerCodeSet in order to mark tcs on this
     * connection as negotiated. It's public because it has to be called
     * also from dii.Request.
     */

    public void markTcsNegotiated()
    {
	if(tcsNegotiated) 
	    return;
	tcsNegotiated = true;
	jacorb.util.Debug.output(4,"TCS marked as negotiated");
    }
	
    public boolean isTCSNegotiated()
    {
	return tcsNegotiated;
    }
	
    public boolean isSSL()
    {
	return is_ssl;
    }

    protected int selectCodeSet(org.omg.CONV_FRAME.CodeSetComponent scs,
				int ourNative)
    {
	// check if we support server's native sets
	if(scs.native_code_set==ourNative) 
	    return ourNative;
		
	// is our native CS supported at server ?
	for(int i=0;i<scs.conversion_code_sets.length;i++)
	{
	    if(scs.conversion_code_sets[i]==ourNative) 
		return ourNative;
	}
		
	// can't find supported set ..
	return 0;
    }

    /**
     * This code selects appropriate codeset for connection from information
     * contained in some IOR. Code is called by client side connection.
     * Returns true if common codeset was found and so that it should be
     * send in context.
     */
    protected boolean selectCodeSet(ParsedIOR pior)
    {
	for(int i=0;i<pior.taggedComponents.length;i++)
	{
	    if(pior.taggedComponents[i].tag != org.omg.IOP.TAG_CODE_SETS.value) 
		continue;

	    jacorb.util.Debug.output(4,"TAG_CODE_SETS found");			

	    // get server cs from IOR 
	    CDRInputStream is =
		new CDRInputStream(orb,pior.taggedComponents[i].component_data);
	    is.setLittleEndian(is.read_boolean());

	    org.omg.CONV_FRAME.CodeSetComponentInfo inf;
	    inf = org.omg.CONV_FRAME.CodeSetComponentInfoHelper.read(is);
		
	    // char data, try UTF8 and ISO8859_1
	    TCS = selectCodeSet(inf.ForCharData,CodeSet.UTF8);
	    if(TCS == 0) 
		TCS = CodeSet.ISO8859_1;
			
	    // wchar data, UTF8 or UTF16 can be used
	    TCSW = selectCodeSet(inf.ForWcharData,CodeSet.UTF8);
	    if(TCSW == 0) 
		TCSW = selectCodeSet(inf.ForWcharData,CodeSet.UTF16);

	    jacorb.util.Debug.output(4,"TCS selected: "+CodeSet.csName(TCS)+","+CodeSet.csName(TCSW));
	    return true;
	}

	// if we are here then it means that no tagged component was in IOR, 
	// probably some old server so set default parameters
	// TODO(devik): we ends up here also for MICO server which seems to have some
	//  strange IOR - have to investigate

	TCS = CodeSet.ISO8859_1;
	TCSW = CodeSet.UTF16; // not as in OMG spec!
		
	// mark as negotiated, why if it's not true ? because we don't
	// want to try negotiate on each IOR until IOR with codeset is found.
	// TODO(devik): Or should we ???
	markTcsNegotiated();
		
	jacorb.util.Debug.output(4,"default TCS selected: "+CodeSet.csName(TCS)+","+CodeSet.csName(TCSW));
	return false;
    }
	
    /**
     * Called on server side connection to evaluate requext contexts and
     * set TCS if one found. Returns true if TCS was correctly set.
     * Currently it's called from BasicAdapter.
     */
    public boolean setServerCodeSet(org.omg.IOP.ServiceContext [] ctx)
    {
        if(!Environment.charsetScanCtx()) return false;
	// search all contexts until TAG_CODE_SETS found
	for(int i=0;i<ctx.length;i++)
	{
	    if(ctx[i].context_id != org.omg.IOP.TAG_CODE_SETS.value) continue;
			
	    // TAG_CODE_SETS found, demarshall
	    CDRInputStream is=new CDRInputStream(orb,ctx[i].context_data);
	    is.setLittleEndian(is.read_boolean());
	    org.omg.CONV_FRAME.CodeSetContext csx;
	    csx=org.omg.CONV_FRAME.CodeSetContextHelper.read(is);
	    TCSW=csx.wchar_data;
	    TCS=csx.char_data;
	    markTcsNegotiated();
	    jacorb.util.Debug.output(4,"TCS set: "+CodeSet.csName(TCS)+","+CodeSet.csName(TCSW));
	    return true;
	}
	return false; // no TAG_CODE_SETS here
    }
	
    /**
     * Adds code set service context to another contexts if needed.
     */
    public org.omg.IOP.ServiceContext [] addCodeSetContext(org.omg.IOP.ServiceContext [] ctx,
							   ParsedIOR pior)
    {		
	// if already negotiated, don't send any further cs contexts
	// we should test it also directly before calling this method
	// as performance optimization

	if(tcsNegotiated || !Environment.charsetSendCtx()) 
	    return ctx;
		
	// not negotiated but also TCS is not selected, so select one
	// if it can't be selected (ior doesn't contain codesets) don't change ctx
	if(TCS==0) 
	    if(!selectCodeSet(pior)) 
		return ctx;
		
	// encapsulate context

	CDROutputStream os = new CDROutputStream(orb);
	os.write_boolean(false);
	org.omg.CONV_FRAME.CodeSetContextHelper.write(os,
				     new org.omg.CONV_FRAME.CodeSetContext(TCS,TCSW));
		
	org.omg.IOP.ServiceContext [] ncx = new org.omg.IOP.ServiceContext[ctx.length+1];
	System.arraycopy(ctx,0,ncx,0,ctx.length);
	ncx[ctx.length]=new org.omg.IOP.ServiceContext(org.omg.IOP.TAG_CODE_SETS.value,
						       os.getBufferCopy());
	// jacorb.util.Debug.output(4,"TCS ctx added: "+CodeSet.csName(TCS)+","+CodeSet.csName(TCSW));
	return ncx;
    }
	
    /**
     * class Constructor, establishes a connection over a given socket
     * this constructor is called from orb.getConnection for clients and
     * from BasicAdapter for servers
     *
     * @param <code>Socket s</code>
     * @param <code>boolean _is_server</code> - if opened from the server side.<br>
     * @except <code>java.io.IOException</code>
     */

    public Connection(jacorb.orb.ORB orb, java.net.Socket s, boolean _is_server )
	throws IOException
    {
	this.orb = orb;
	mysock = s;
	is_server = _is_server;
        is_ssl     = jacorb.security.ssl.SSLSetup.isSSL ( s ); // bnv
	in_stream = 
	    new BufferedInputStream(mysock.getInputStream());
	out_stream = 
	    new BufferedOutputStream(mysock.getOutputStream(), Environment.outBufSize() );

	String ip = mysock.getInetAddress().getHostAddress();
	if( ip.indexOf('/') > 0)
	    ip = ip.substring( ip.indexOf('/') + 1 );
	String host_and_port = ip + ":"+ mysock.getPort();
	String ssl = is_ssl ? "SSL " : ""; //bnv
	connection_info = host_and_port;
	client_count = 1;

	if( ! _is_server ) // for clients only:
	{
	    jacorb.util.Debug.output(1,"New " + ssl + "connection to " + host_and_port);
	    rep = new ReplyReceptor( this );
	}
	else
	    jacorb.util.Debug.output(1,"Accepted " + ssl + "connection from " + host_and_port);
    }


    /**
     *	Close a connection. Should only be done directly if the connection
     *  is broken or idCount has reached zero. Otherwise, use releaseConnection.
     */

    synchronized void closeConnection()
    {
	jacorb.util.Debug.output(1,"Closing connection to " + connection_info);
	try
	{
	    if( rep != null )
		rep.stopReceptor();

	    jacorb.util.Debug.output(4,"Closing connection to " + connection_info + " (reply receptor closed)");
	    try 
	    {
		if( mysock != null )
		    mysock.close();
	    } 
	    catch(IOException e) 
	    {
		jacorb.util.Debug.output(4, e);
	    }
	    
	    if( in_stream != null )
		in_stream.close();

	    jacorb.util.Debug.output(4,"Closing connection to " + connection_info + " (in streams closed)");

	    if( out_stream != null )
		out_stream.close();

	jacorb.util.Debug.output(4,"Closing connection to " + connection_info + " (out streams closed)");

	} 
	catch ( IOException e)
	{
	    jacorb.util.Debug.output(4,e);
	}

	in_stream = null;
	out_stream = null;
	rep = null;

	mysock = null;

	jacorb.util.Debug.output(3,"Closing connection to " + connection_info + " (sockets closed)");

	orb.removeConnection( this );

	// connection_info = null;

	if( buffers.size() > 0 )
	{
	    try 
	    {
		reconnect();
	    } 
	    catch ( org.omg.CORBA.COMM_FAILURE exe )
	    {
		jacorb.util.Debug.output(1,"Could not reconnect to " + 
				   connection_info + " (loss of outstanding replies)");
	    }
	}
    }

    public BufferManager getBufferManager()
    {
	return ((jacorb.orb.ORB)orb).getBufferManager();
    }

    public boolean connected()
    {
	return mysock != null;
    }

    /** 
     * @returns a new request id 
     */

    public synchronized int getId()
    {
	return id_count++; /* */
    }

    protected void incUsers()
    {
	client_count++;
    }

    /** called by delegate when the delegate is duplicated */

    public void duplicate()
    {
	incUsers();
    }

    /** 
     * called to notify all waiting threads that the connection is
     * broken
     */

    private void abort()
	throws  EOFException
    {
	jacorb.util.Debug.output(3,"Connection to " + connection_info + " aborting...");
	Enumeration keys = replies.keys();
	int lost_replies = 0;

	while(keys.hasMoreElements()) 
	{
	  Object key = keys.nextElement();

	  ReplyInputStream client = (ReplyInputStream)replies.get(key);
	  client.cancel();
	  lost_replies++;

	  CDROutputStream os = (CDROutputStream) buffers.get(key);
	  if (os != null)
	    os.release();
	}

	replies.clear();
	buffers.clear();
	objects.clear();

	if( lost_replies > 0 )
	    jacorb.util.Debug.output(2,"Lost " + lost_replies + " outstanding replies");
	throw new EOFException();
    }

    /**
     * low-level input routine, not MT-safe, used by the ReplyReceptor object
     * for this connection
     */

    public  byte[] readBuffer() 
	throws IOException 
    {
	for( int i = 0; i < Messages.MSG_HEADER_SIZE; i++ )
	{
	    int input = -1;

	    try
	    {
		input = in_stream.read();
	    } 
	    catch( java.io.InterruptedIOException ioint )
	    {
		/* if a client-side time out has been set we have to
		   abort. NOTE: outstanding replies are lost! */
		
		jacorb.util.Debug.output(3,"Connection timed out");
	    }
	    catch( Exception e )
	    {
		jacorb.util.Debug.output(3,e);
	    }
	    if( input < 0 )		
	    {
		abort();
	    }
	    header[i]=(byte)input;
	}

	

	/* check for SECIOP vs. GIOP headers */

	if( (char)header[0] == 'S' && (char)header[1] == 'E' && 
	    (char)header[2] =='C' && (char)header[3] == 'P' )
	{
	    throw new IOException("Cannot handle SECIOP messages yet...");
	}
	else if( (char)header[0] == 'G' && (char)header[1] == 'I' && 
		 (char)header[2] =='O' && (char)header[3] == 'P')
	{
	    /* determine message size, but respect byte order! */
	    int msg_size = 0;
	    
	    if( header[6] == 0 ) // big-endian
		msg_size =  ((0xff & header[8]) << 24) + 
		    ((0xff & header[9]) << 16) + 
		    ((0xff & header[10])<< 8) + 
		    ((0xff & header[11]));
	    else	// little-endian
		msg_size =  ((0xff & header[11]) << 24) + 
		    ((0xff & header[10]) << 16) + 
		    ((0xff & header[9])  << 8) + 
		    ((0xff & header[8]));
	    
	    if( msg_size < 0 )
	    {
		abort();
	    }
	    
	    int bufSize = msg_size + Messages.MSG_HEADER_SIZE;
	    byte[] inbuf = orb.getBufferManager().getBuffer(bufSize);
	    
	    /* copy header */
	    
	    for( int ii = 0; ii < Messages.MSG_HEADER_SIZE; ii++ )
	    {
		inbuf[ii] = (byte)header[ii];
	    }

	    int read = Messages.MSG_HEADER_SIZE;

	    //	    while( read < inbuf.length )
	    while( read < bufSize )
	    {
		int n = -1;
		try
		{
		    n = in_stream.read(inbuf, read, bufSize-read);
		} 
		catch( IOException ie )
		{
		    jacorb.util.Debug.output(4,ie);
		}

		if( n<0 )
		{
		    abort();
		}  
		read += n;
	    }
	    return inbuf;
	}
	else 
	{
	    jacorb.util.Debug.output( 0, "Unknown message header type detected by " + 
				      Thread.currentThread().getName() + ", discarding ", header );
	    return readBuffer();
	    //	    throw new IOException("Unknown message header type!");
	}
    }

    /**
     * Receive a GIOP reply over this connection
     * and return an appropriate CDRInputStream
     */

    public void  receiveReply() 
	throws IOException, CloseConnectionException
    {
	if( !connected())
	    throw new EOFException();
	byte [] buf = readBuffer();
	int msg_type = buf[7];

	switch ( msg_type )
	{
	case  org.omg.GIOP.MsgType_1_0._Reply: 
	    {
		jacorb.util.Debug.output( 8, "receiveReply", buf );
		Integer key = new Integer(Messages.getRequestId( buf, msg_type ));
		org.omg.CORBA.Object o = (org.omg.CORBA.Object)objects.remove( key );

		/* retrieve the ReplyInputStream that is waiting for this reply */
		CDROutputStream os = (CDROutputStream) buffers.remove( key );
		if (os != null)
		  os.release();

		ReplyInputStream pending = (ReplyInputStream)replies.remove( key );
      
		if( pending != null ) 
		    pending.init( buf, o );
		else 
		    System.err.println("Serious Error! No pending request for reply " + key );
		break;
	    }
	case org.omg.GIOP.MsgType_1_0._LocateReply:
	    { 
		Integer key = new Integer(Messages.getRequestId( buf, msg_type ));

		/* retrieve the ReplyInputStream that is waiting for this reply */
		CDROutputStream os = (CDROutputStream) buffers.remove( key );
		if (os != null)
		  os.release();

		jacorb.orb.giop.LocateReplyInputStream pending = 
		    (jacorb.orb.giop.LocateReplyInputStream)replies.remove( key );

		if( pending != null ) 
		{
		    pending.init( buf );
		} 
		else 
		    System.err.println("Serious Error! No pending request for reply " + key );
		break;
	    }
	case org.omg.GIOP.MsgType_1_0._CloseConnection:
	    {
		jacorb.util.Debug.output(3,"got close connection message");
		throw new CloseConnectionException("Received <CloseConnection> from Server");
	    }
	default:
	    throw new IOException("Received Message of unrecognizable type "+ msg_type);
	}
   }

    /**
     *
     */

    public synchronized void reconnect()
	throws org.omg.CORBA.COMM_FAILURE
    {	
	jacorb.util.Debug.output(1,"Trying to reconnect to " + connection_info);

	int retries = Environment.noOfRetries();
	while( retries > 0 ) 
	{
	    try 
	    {
		mysock = new java.net.Socket( connection_info.substring(0,connection_info.indexOf(":")), 
					      new Integer(connection_info.substring(
							     connection_info.indexOf(":")+1)).intValue());
		mysock.setTcpNoDelay(true);
		in_stream = new BufferedInputStream(mysock.getInputStream());
		out_stream = new BufferedOutputStream(mysock.getOutputStream(), Environment.outBufSize());
		
		String ip = mysock.getInetAddress().getHostAddress();
		if( ip.indexOf('/') > 0)
		    ip = ip.substring( ip.indexOf('/') + 1 );
		String host_and_port = ip + ":"+ mysock.getPort();
		orb.addConnection( this );
		connection_info = host_and_port;
		jacorb.util.Debug.output(1,"Reconnected to " + host_and_port);
		rep = new ReplyReceptor( this );
		for( Enumeration e = buffers.elements(); e.hasMoreElements();)
		{
		  CDROutputStream os = (CDROutputStream) e.nextElement();
		  os.write_to(out_stream);
		  os.release();
		}
		buffers.clear();
		return;
	    } 
	    catch ( IOException c ) 
	    { 
		jacorb.util.Debug.output(1,"Retrying to reconnect to " + connection_info);
		try 
		{
		    Thread.sleep( Environment.retryInterval() );
		} 
		catch ( InterruptedException i ){}
		retries--;
	    }
	}
	if( retries == 0 )
	    throw new org.omg.CORBA.COMM_FAILURE("Retries exceeded, couldn't reconnect to " + 
						 connection_info );
    }

    /**
     *	Release a connection. If called by the last client using
     *	this connection (using Stub._release()), it is closed.
     */

    public synchronized void releaseConnection()
    {
	client_count--;
	if( client_count == 0 )
	{
	    closeConnection();
	}
	else
	    jacorb.util.Debug.output(2,"Releasing one connection to " + 
			       connection_info );		
    }

    /**
     *
     */

    public String getInfo()
    {
	return connection_info;
    }



    /** writes the arguments of method calls to the output stream */
    
    public org.omg.CORBA.portable.InputStream sendRequest( org.omg.CORBA.Object o,
							   RequestOutputStream os ) 
	throws org.omg.CORBA.COMM_FAILURE
    {

	if( !connected() )
	{
	    jacorb.util.Debug.output(3, "not connected" );
	    reconnect();
	}

	synchronized( writeLock )
	{
	    ReplyInputStream rep = null;
	    
	    try
	    {
//  		byte buf [];
//  		int size = -1;

		if( Environment.clientInterceptMessages())
		{
		    jacorb.util.Debug.output(4, " passing buffer to interceptors " );
		    os.setBuffer(orb.client_messageIntercept_pre( o, os.getBufferCopy()));
		}
//  		else
//  		{
//  		    buf = os.getInternalBuffer();
//  		    size = os.size();
//  		}

		if( os.response_expected() )
		{
		    rep = new ReplyInputStream(this, os.requestId());
		    Integer key = new Integer( os.requestId() );
		    buffers.put( key, os );
		    replies.put( key, rep );
		    objects.put( key, o );
		}

		if (jacorb.util.Environment.verbosityLevel() > 4)
		  //This is a costly op, since it involves buffer copying!
		  jacorb.util.Debug.output(5,"send request", os.getBufferCopy());

		os.write_to(out_stream);
		
		/** release the stream so its buffer can go back into the pool */
		if (! os.response_expected() )
		  os.release();		

		//jacorb.util.Debug.output(5,"sendreq",buf,size);
	    } 
	    catch ( Exception e )
	    {
		jacorb.util.Debug.output(2,e);
		throw new org.omg.CORBA.COMM_FAILURE();
	    }		
	    return rep;
	}
    }

    public synchronized LocateReplyInputStream sendLocateRequest( LocateRequestOutputStream os ) 
	throws org.omg.CORBA.COMM_FAILURE
    {
	if( !connected() )
	{
	    reconnect();
	}

	synchronized( writeLock )
	{
	    LocateReplyInputStream rep = null;	    
	    try
	    {
		rep = new LocateReplyInputStream(this, os.requestId());
		Integer key = new Integer( os.requestId() );
		buffers.put( key, os );
		replies.put( key, rep );
		
		os.write_to(out_stream);
	    } 
	    catch ( Exception e )
	    {
		jacorb.util.Debug.output(2,e);
		throw new org.omg.CORBA.COMM_FAILURE();
	    }		
	    return rep;
	}
    }


    /**
     * send a "close connection" message to the other side.
     */

    public synchronized void sendCloseConnection() 
	throws org.omg.CORBA.COMM_FAILURE 
    {

	    try
	    {
		out_stream.write( Messages.closeConnectionMessage());
		out_stream.flush();
		closeConnection();
	    } 
	    catch ( Exception e )
	    {
		throw new org.omg.CORBA.COMM_FAILURE();
	    }

    }


    /**
     *
     */

    public void sendLocateReply( int request_id, int status, org.omg.CORBA.Object arg ) 
	throws org.omg.CORBA.COMM_FAILURE 
    {
	synchronized( writeLock )
	{
	    try
	    {
		out_stream.write( Messages.locateReplyMessage(request_id,status,arg,this));
		out_stream.flush();
	    } 
	    catch ( Exception e )
	    {
		jacorb.util.Debug.output(2,e);
		throw new org.omg.CORBA.COMM_FAILURE();
	    }
	}
    }

    /**
     * called from dsi/ServerRequest
     */


    public void sendReply( jacorb.orb.giop.ReplyOutputStream os ) 
	throws IOException 
    {
	synchronized( writeLock )
	{
	    try
	    {	     
		if( Environment.serverInterceptMessages())
		{
		    byte[] buf = orb.server_messageIntercept_post(os.getBufferCopy());
		    out_stream.write( buf, 0, buf.length);
		    out_stream.flush();
		}
		else
		    // jacorb.util.Debug.output(3,"bypassing any message-level interceptors");
		    os.write_to(out_stream);
		
		os.release();
	    } 
	    catch ( Exception e )
	    {
		jacorb.util.Debug.output(2,e);
		throw new org.omg.CORBA.COMM_FAILURE();
	    }
	}
    }


    public void setTimeOut(int timeout)
	throws SocketException
    {
	mysock.setSoTimeout(timeout);
    }
   
    /**
     * utility method, doesn't actually belong here
     */
    public static void dumpBA(byte bs[])
    {
	int len = bs.length;
	for (int i=0; i<len; i++)
	{
	    if (0 == i%20)
	    {
		java.lang.System.out.println();
	    }
	    dumpHex(bs[i]);
	}
	java.lang.System.out.println();
    }

    /**
     * utility method, doesn't actually belong here
     */

    public static void dumpHex(byte b)
    {
	int n1 = (b & 0xff) / 16;
	int n2 = (b & 0xff) % 16;
	char c1 = (char)(n1>9 ? ('A'+(n1-10)) : ('0'+n1));
	char c2 = (char)(n2>9 ? ('A'+(n2-10)) : ('0'+n2));
	char c3;
	if (b>(byte)31 && b<(byte)127)
	    c3 = (char)b;
	else
	    c3 = ' ';
	java.lang.System.out.print(c1+(c2+" ")+c3+"  ");
    }

    public BufferedOutputStream get_out_stream(){
	return out_stream;
    }

    public Hashtable get_buffers(){
	return buffers;
    }
	
    public Hashtable get_replies(){
	return replies;
    }

    public Hashtable get_objects(){
	return objects;
    }

    public void writeDirectly(byte[] buffer,int len) 
	throws IOException
    {
	get_out_stream().write(buffer,0, len);
	get_out_stream().flush();
    }
}



