public SocketManager(Socket s) throws IOException 
{
       
      socket = s;
      sendQueue = new LinkedQueue();
      replyMap = new ConcurrentHashMap();
      trace = log.isTraceEnabled();
}

 public  void sendMessage(BaseMsg msg) throws Exception 
{
      internalSendMessage(msg, true);
      if (msg.error != null)
      {
         if (trace)
            log.trace("sendMessage will throw error", msg.error);
         throw msg.error;
      }
}

Send a two-way message and block the calling thread until the msg reply is received. This enques the msg to the sendQueue, places the msg in the replyMap and waits on the msg. The msg is notified by the read task thread when it finds a msg with a msgID that maps to the msg in the msgReply map.

 public  void sendOneWay(BaseMsg msg) throws Exception 
{
      msg.getMsgID();
      internalSendMessage(msg, false);
}

Send a one-way.

 public  void sendReply(BaseMsg msg) throws Exception 
{
      msg.trimReply();
      internalSendMessage(msg, false);
}

Send a reply.

 public  void setBufferSize(int size) 
{
      this.bufferSize = size;
}

Sets the buffer size

 public  void setChunkSize(int size) 
{
      this.chunkSize = size;
}

Sets the chunk size

 public  void setHandler(SocketManagerHandler handler) 
{
      this.handler = handler;
      if (bufferedInput != null)
         bufferedInput.setStreamListener(handler);
      if (bufferedOutput != null)
         bufferedOutput.setStreamListener(handler);
}

Set the callback handler for msgs that were not originated by the socket manager. This is any msgs read that was not sent via the sendMessage method.

 public  void start(ThreadGroup tg) 
{
      if (trace)
         log.trace("start called", new Exception("Start stack trace"));
      InetAddress inetAddr = socket.getInetAddress();
      String ipAddress = (inetAddr != null) ? inetAddr.getHostAddress() : "< unknown >";
      ipAddress += ":" + socket.getPort();
      if (pool == null)
      {
         // TODO: Check the validity of this config
         pool = new PooledExecutor(5);
         pool.setMinimumPoolSize(1);
         pool.setKeepAliveTime(1000 * 60);
         pool.runWhenBlocked();
         String id = "SocketManager.MsgPool@"+
            Integer.toHexString(System.identityHashCode(this))
            + " client=" +  ipAddress;
         pool.setThreadFactory(new UILThreadFactory(id));
      }
      ReadTask readTask = new ReadTask();
      readThread = new Thread(tg, readTask, "UIL2.SocketManager.ReadTask#" + taskID.increment() + " client=" +  ipAddress);
      readThread.setDaemon(true);
      WriteTask writeTask = new WriteTask();
      writeThread = new Thread(tg, writeTask, "UIL2.SocketManager.WriteTask#" + taskID.increment() + " client=" + ipAddress);
      writeThread.setDaemon(true);
      synchronized (running)
      {
         readState = STARTED;
         writeState = STARTED;
         running.set(true);
      }
      try
      {
         readThread.start();
         writeThread.start();
      }
      catch (Throwable t)
      {
         try
         {
            stop();
         }
         catch (Throwable ignored)
         {
         }
         log.warn("Error starting socket manager threads", t);
      }
}

Start the read and write threads using the given thread group and names of "UIL2.SocketManager.ReadTask" and "UIL2.SocketManager.WriteTask".

 public  void stop() 
{
      synchronized (running)
      {
         if (trace)
            log.trace("stop() " + readThread + " " + writeThread);
         if (readState == STARTED)
         {
            readState = STOPPING;
            readThread.interrupt();
         }
         if (writeState == STARTED)
         {
            writeState = STOPPING;
            writeThread.interrupt();
         }
         running.set(false);
         if (pool != null)
         {
            pool.shutdownNow();
            pool = null;
         }
         try
         {
            socket.close();
         }
         catch (Throwable ignored)
         {
         }
      }
}

Stop the read and write threads by interrupting them.