You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2012/11/21 14:28:19 UTC
svn commit: r1412087 -
/activemq/trunk/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java
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();
}
}