[jboss-cvs] JBossCache/src/org/jboss/cache/loader/tcp ...

[Manik Surtani]

  User: msurtani
  Date: 06/10/25 08:49:31

  Modified:    src/org/jboss/cache/loader/tcp  TcpCacheServer.java
  Log:
  JBCACHE-690
  JBCACHE-800
  JBCACHE-810
  
  Revision  Changes    Path
  1.21      +67 -41    JBossCache/src/org/jboss/cache/loader/tcp/TcpCacheServer.java
  
  (In the diff below, changes in quantity of whitespace are not shown.)
  
  Index: TcpCacheServer.java
  ===================================================================
  RCS file: /cvsroot/jboss/JBossCache/src/org/jboss/cache/loader/tcp/TcpCacheServer.java,v
  retrieving revision 1.20
  retrieving revision 1.21
  diff -u -b -r1.20 -r1.21
  --- TcpCacheServer.java	24 Oct 2006 12:36:41 -0000	1.20
  +++ TcpCacheServer.java	25 Oct 2006 12:49:31 -0000	1.21
  @@ -12,15 +12,18 @@
   import org.jboss.cache.loader.DelegatingCacheLoader;
   
   import javax.management.MBeanServer;
  +import javax.management.MBeanServerInvocationHandler;
   import javax.management.MalformedObjectNameException;
   import javax.management.ObjectName;
  -import javax.management.MBeanServerInvocationHandler;
  +import java.io.BufferedInputStream;
  +import java.io.BufferedOutputStream;
   import java.io.IOException;
   import java.io.ObjectInputStream;
   import java.io.ObjectOutputStream;
   import java.net.InetAddress;
   import java.net.ServerSocket;
   import java.net.Socket;
  +import java.net.SocketException;
   import java.net.UnknownHostException;
   import java.util.ArrayList;
   import java.util.HashMap;
  @@ -34,7 +37,7 @@
    * TCP-IP based CacheServer, configure TcpDelegatingCacheLoader with host and port of this server
    *
    * @author Bela Ban
  - * @version $Id: TcpCacheServer.java,v 1.20 2006/10/24 12:36:41 msurtani Exp $
  + * @version $Id: TcpCacheServer.java,v 1.21 2006/10/25 12:49:31 msurtani Exp $
    */
   public class TcpCacheServer implements TcpCacheServerMBean
   {
  @@ -46,8 +49,13 @@
      ObjectName cache_name;
      String config;
      boolean running = true;
  -   List conns = new LinkedList();
  +   final List conns = new LinkedList();
      String agendId;
  +   Thread serverThread;
  +   /**
  +    * whether or not to start the server thread as a daemon.  Should be false if started from the command line, true if started as an MBean.
  +    */
  +   boolean daemon = true;
      static Log mylog = LogFactory.getLog(TcpCacheServer.class);
   
   
  @@ -155,16 +163,49 @@
   
         srv_sock = new ServerSocket(port, 10, bind_addr);
         System.out.println("TcpCacheServer listening on : " + srv_sock.getInetAddress() + ":" + srv_sock.getLocalPort());
  +      mylog.info("TcpCacheServer listening on : " + srv_sock.getInetAddress() + ":" + srv_sock.getLocalPort());
  +
  +      running = true;
  +
  +      serverThread = new Thread("TcpCacheServer")
  +      {
  +         public void run()
  +         {
  +            try
  +            {
         while (running)
         {
  -         client_sock = srv_sock.accept();
  -         conn = new Connection(client_sock, cache);
  +                  Socket client_sock = srv_sock.accept();
  +                  Connection conn = new Connection(client_sock, cache);
            conns.add(conn);
            conn.start();
         }
      }
  +            catch (SocketException se)
  +            {
  +               if (!running)
  +               {
  +                  // this is because of the stop() lifecycle method being called.
  +                  // ignore.
  +                  mylog.info("Shutting down TcpCacheServer");
  +               }
  +               else
  +               {
  +                  mylog.error("Caught exception! Shutting down server thread.", se);
  +               }
  +            }
  +            catch (IOException e)
  +            {
  +               mylog.error("Caught exception! Shutting down server thread.", e);
  +            }
  +         }
  +      };
  +      serverThread.setDaemon(daemon);
  +      serverThread.start();
   
  -   public void stopService()
  +   }
  +
  +   public void stop()
      {
         running = false;
         for (Iterator it = conns.iterator(); it.hasNext();)
  @@ -173,11 +214,13 @@
            conn.close();
         }
         conns.clear();
  +
         if (srv_sock != null)
         {
            try
            {
               srv_sock.close();
  +            srv_sock = null;
            }
            catch (IOException e)
            {
  @@ -203,30 +246,28 @@
      {
      }
   
  -   public void stop()
  -   {
  -   }
  -
      public void destroy()
      {
      }
   
   
  -   private static class Connection implements Runnable
  +   private class Connection implements Runnable
      {
         Socket sock = null;
         ObjectInputStream input = null;
         ObjectOutputStream output = null;
  -      //TreeCacheMBean     c;
         TreeCache c;
         Thread t = null;
   
         public Connection(Socket sock, TreeCache cache) throws IOException
         {
  -         //public Connection(Socket sock, TreeCacheMBean cache) throws IOException {
            this.sock = sock;
  -         output = new ObjectOutputStream(sock.getOutputStream());
  -         input = new ObjectInputStream(sock.getInputStream());
  +
  +         output = new ObjectOutputStream(new BufferedOutputStream(sock.getOutputStream()));
  +         output.flush();
  +
  +         input = new ObjectInputStream(new BufferedInputStream(sock.getInputStream()));
  +
            c = cache;
         }
   
  @@ -262,6 +303,9 @@
            catch (Throwable th)
            {
            }
  +
  +         // remove self from connections list
  +         conns.remove(this);
         }
   
         public void run()
  @@ -282,33 +326,14 @@
               }
               catch (IOException e)
               {
  -               mylog.warn("failed reading data, thread will terminate", e);
  -               try
  -               {
  -                  if (output != null) output.close();
  -               }
  -               catch (Throwable th)
  -               {
  -               }
  -               try
  -               {
  -                  if (input != null) input.close();
  -               }
  -               catch (Throwable th)
  -               {
  -               }
  -               try
  -               {
  -                  if (sock != null) sock.close();
  -               }
  -               catch (Throwable th)
  -               {
  -               }
  +               mylog.debug("Client closed socket");
  +               close();
                  break;
               }
   
               try
               {
  +               output.reset();
                  switch (op)
                  {
                     case DelegatingCacheLoader.delegateGetChildrenNames:
  @@ -509,6 +534,7 @@
            return;
         }
         server = new TcpCacheServer();
  +      server.daemon = false;
         server.setBindAddress(bind_addr);
         server.setPort(port);
         server.setConfig(config);