[jboss-cvs] JBossAS SVN: r81752 - branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/il/uil2.

[jboss-cvs-commits at lists.jboss.org]

Author: bershath27
Date: 2008-11-28 01:49:14 -0500 (Fri, 28 Nov 2008)
New Revision: 81752

Modified:
   branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java
Log:
Applied https://jira.jboss.org/jira/browse/ASPATCH-400

Modified: branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java
===================================================================
--- branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java	2008-11-28 03:40:30 UTC (rev 81751)
+++ branches/JBoss_4_0_4_GA_CP/messaging/src/main/org/jboss/mq/il/uil2/SocketManager.java	2008-11-28 06:49:14 UTC (rev 81752)
@@ -142,9 +142,24 @@
          writeState = STARTED;
          running.set(true);
       }
-
-      readThread.start();
-      writeThread.start();
+      try
+      {
+    	  readThread.start();
+          writeThread.start();  
+      }
+      catch (Throwable t)
+      {
+         try
+         {
+            stop();
+         }
+         catch (Throwable ignored)
+         {
+         }
+         
+         log.warn("Error starting socket manager threads", t);
+      }
+      
    }
 
    /** Stop the read and write threads by interrupting them.
@@ -153,7 +168,9 @@
    {
       synchronized (running)
       {
-         if (readState == STARTED)
+         if (trace)
+            log.trace("stop() " + readThread + " " + writeThread);
+    	 if (readState == STARTED)
          {
             readState = STOPPING;
             readThread.interrupt();
@@ -169,6 +186,13 @@
             pool.shutdownNow();
             pool = null;
          }
+         try
+         {
+            socket.close();
+         }
+         catch (Throwable ignored)
+         {
+         }
       }
    }
 
@@ -297,7 +321,7 @@
       public void run()
       {
          int msgType = 0;
-         log.debug("Begin ReadTask.run");
+         log.debug("Begin ReadTask.run " + Thread.currentThread());
          try
          {
             bufferedInput = new NotifyingBufferedInputStream(socket.getInputStream(), bufferSize, chunkSize, handler);
@@ -383,7 +407,7 @@
                break;
             }
          }
-         log.debug("End ReadTask.run");
+         log.debug("End ReadTask.run " + Thread.currentThread());
       }
 
       /**
@@ -397,7 +421,7 @@
          }
          catch (Throwable e)
          {
-            if (e instanceof JMSException)
+            if (e instanceof JMSException || running.get() == false)
                log.trace("Failed to handle: " + msg.toString(), e);
             else if (e instanceof RuntimeException || e instanceof Error)
                log.error("Failed to handle: " + msg.toString(), e);
@@ -410,7 +434,10 @@
             }
             catch (Exception ie)
             {
-               log.debug("Failed to send error reply", ie);
+               if (running.get())
+                  log.debug("Failed to send error reply", ie);
+               else
+                  log.trace("Failed to send error reply", ie);
             }
          }
       }
@@ -497,7 +524,7 @@
    {
       public void run()
       {
-         log.debug("Begin WriteTask.run");
+    	  log.debug("Begin WriteTask.run " + Thread.currentThread());
          try
          {
             bufferedOutput =
@@ -514,9 +541,17 @@
          while (true)
          {
             BaseMsg msg = null;
+            
+            synchronized (running)
+            {
+               if (writeState != STARTED)
+                  break;
+            }
             try
             {
-               msg = (BaseMsg) sendQueue.take();
+               msg = (BaseMsg) sendQueue.poll(10000l);
+               if (msg == null)
+                   continue; // Check for stop if no message for 10 seconds
                if (trace)
                   log.trace("Write msg: " + msg);
                msg.write(out);
@@ -539,7 +574,7 @@
                break;
             }
          }
-         log.debug("End WriteTask.run");
+         log.debug("End WriteTask.run " + Thread.currentThread());
       }
 
       /**