svn commit: r1412087 - /activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java

[chirino]

Author: chirino
Date: Wed Nov 21 13:28:18 2012
New Revision: 1412087

URL: http://svn.apache.org/viewvc?rev=1412087&view=rev
Log:
Trying to fix CI test failures: Don't send anymore data down the transport once 
it's closed/disconnected.

Modified:
    
activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java

Modified: 
activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java?rev=1412087&r1=1412086&r2=1412087&view=diff
==============================================================================
--- 
activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java
 (original)
+++ 
activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java
 Wed Nov 21 13:28:18 2012
@@ -18,6 +18,7 @@ package org.apache.activemq.transport.mq
 
 import java.io.IOException;
 import java.security.cert.X509Certificate;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.jms.JMSException;
 import org.apache.activemq.broker.BrokerContext;
@@ -44,6 +45,7 @@ public class MQTTTransportFilter extends
     private final MQTTProtocolConverter protocolConverter;
     private MQTTInactivityMonitor monitor;
     private MQTTWireFormat wireFormat;
+    private final AtomicBoolean stopped = new AtomicBoolean();
 
     private boolean trace;
 
@@ -87,12 +89,21 @@ public class MQTTTransportFilter extends
     }
 
     public void sendToMQTT(MQTTFrame command) throws IOException {
-        if (trace) {
-            TRACE.trace("Sending: \n" + command);
+        if( !stopped.get() ) {
+            if (trace) {
+                TRACE.trace("Sending: \n" + command);
+            }
+            Transport n = next;
+            if (n != null) {
+                n.oneway(command);
+            }
         }
-        Transport n = next;
-        if (n != null) {
-            n.oneway(command);
+    }
+
+    @Override
+    public void stop() throws Exception {
+        if( stopped.compareAndSet(false, true) ) {
+            super.stop();
         }
     }