svn commit: r638596 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java

[chirino]

Author: chirino
Date: Tue Mar 18 14:43:42 2008
New Revision: 638596

URL: http://svn.apache.org/viewvc?rev=638596&view=rev
Log:
access to the writeThread was not safe.. Plus interrupting another thread is 
not recommended anyways.
also fail new oneway() operations once inactivity is detected.


Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java?rev=638596&r1=638595&r2=638596&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java
 Tue Mar 18 14:43:42 2008
@@ -23,6 +23,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.activemq.command.KeepAliveInfo;
 import org.apache.activemq.command.WireFormatInfo;
@@ -51,12 +52,13 @@
 
     private final AtomicBoolean commandSent = new AtomicBoolean(false);
     private final AtomicBoolean inSend = new AtomicBoolean(false);
+    private final AtomicBoolean inactive = new AtomicBoolean(false);
 
     private final AtomicBoolean commandReceived = new AtomicBoolean(true);
     private final AtomicBoolean inReceive = new AtomicBoolean(false);
     private SchedulerTimerTask writeCheckerTask;
     private SchedulerTimerTask readCheckerTask;
-    private Thread writeThread;
+    
     private long readCheckTime;
     private long writeCheckTime;
     
@@ -151,20 +153,17 @@
             return;
         }
         if (!commandReceived.get()) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("No message received since last read check for " + 
toString() + "! Throwing InactivityIOException.");
+            if( inactive.getAndSet(false) ) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("No message received since last read check for " 
+ toString() + "! Throwing InactivityIOException.");
+                }
+                ASYNC_TASKS.execute(new Runnable() {  
+                    public void run() {
+                        onException(new InactivityIOException("Channel was 
inactive for too long: "+next.getRemoteAddress()));
+                    };
+                    
+                });
             }
-            ASYNC_TASKS.execute(new Runnable() {  
-                public void run() {
-                    Thread t = writeThread;
-                    if (t != null) {
-                        t.interrupt();
-                    }
-                    onException(new InactivityIOException("Channel was 
inactive for too long: "+next.getRemoteAddress()));
-                        
-                };
-                
-            });
 
         } else {
             if (LOG.isTraceEnabled()) {
@@ -223,12 +222,11 @@
                     startMonitorThreads();
                 }
             }
-            synchronized (writeChecker) {
-                writeThread=Thread.currentThread();
-                next.oneway(o);
+            if( inactive.get() ) {
+                throw new InactivityIOException("Channel was inactive for too 
long: "+next.getRemoteAddress());
             }
+            next.oneway(o);
         } finally {
-            writeThread=null;
             commandSent.set(true);
             inSend.set(false);
         }