Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Trying to fix CI test failures: Don't send anymore data down the transport once it's closed/disconnected. 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1412087 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2012-11-21 13:28:18 +00:00
parent 2751497e38
commit d350ce6f61
1 changed files with 16 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.transport.mqtt;
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 TransportFilter implements MQTTTranspor
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 TransportFilter implements MQTTTranspor
}
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();
}
}