fusesource/mqtt-client — Issue #8: Client listener does not seem to function in multiple threads
Author: gjckis
Created: 2012-02-29 09:11:30+00:00
State: open

Hi,

Based on the client example I wrote a Listener class to listen to topics from my MQTT server. When I start the client it listens to a topic by starting it in a thread. If I start one thread then everything is fine. If I start 2 threads then something goes wrong and onConnected / onDisconnected starts flowing over me :-)

Everything in my listener class is private so I started to wonder if I am not allowed to use the org.fusesource.mqtt.client library in threads.

Has anyone tried this?

Thanks in advance,
- Gummi

PS: Here is the source I have

////////////ListenerMain.java

public class ListenerMain {

public static void main(String[] args) throws Exception {
//Listener mqList1 = new Listener("tcp://mq.mydev.local:1883", "topic1/#", "c:/gj.txt", true);
//new Thread(mqList1).start( );

//Listener mqList2 = new Listener("tcp://mq.mydev.local:1883", "topic2/#", "c:/gj2.txt", true);
//new Thread(mqList2).start( );

Listener mqList3 = new Listener("tcp://mq.mydev.local:1883", "topic3/#", "c:/gj3.txt", true);
new Thread(mqList3).start( );

}

}

////////////// Listener.java

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.logging.*;
import java.io.*;
import java.net.URISyntaxException;

public class Listener implements Runnable{
private static final long DEFAULT_SLEEP_BEFORE_RE_ATTEMPT_IN_SECONDS = 5000;
private static final long DEFAULT_MAX_RE_ATTEMPT_DURATION_IN_SECONDS = 3600 * 3;

private long listenerSleepBeforeReAttemptInSeconds;
private long listenerMaxReAttemptDurationInSeconds;
private MQTT mqtt;

private ArrayList topics;
private boolean listenerDebug;
private String listenerHostURI;
private String listenerTopic;
private String listenerLogFile;
private long listenerLastSuccessfulSubscription;

private Logger fLogger;
private String NEW_LINE = System.getProperty("line.separator");

public Listener(String listenerHostURI, String listenerTopic, String logFile, boolean debug) {
this(listenerHostURI, listenerTopic, logFile, DEFAULT_SLEEP_BEFORE_RE_ATTEMPT_IN_SECONDS, DEFAULT_MAX_RE_ATTEMPT_DURATION_IN_SECONDS, debug);
}

public Listener(String listenerHostURI, String listenerTopic, String logFile, long listenerSleepBeforeReAttemptInSeconds, long listenerMaxReAttemptDurationInSeconds, boolean debug) {
init(listenerHostURI, listenerTopic, logFile, listenerSleepBeforeReAttemptInSeconds, listenerMaxReAttemptDurationInSeconds, debug);
}

private void init(String listenerHostURI, String listenerTopic, String logFile,	long listenerSleepBeforeReAttemptInSeconds, long listenerMaxReAttemptDurationInSeconds, boolean debug) {
this.listenerHostURI = listenerHostURI;
this.listenerTopic = listenerTopic;
this.listenerLogFile = logFile;
this.listenerSleepBeforeReAttemptInSeconds = listenerSleepBeforeReAttemptInSeconds;
this.listenerMaxReAttemptDurationInSeconds = listenerMaxReAttemptDurationInSeconds;
this.listenerDebug = debug;
initMQTT();
}

private void initMQTT() {
mqtt = new MQTT();
listenerLastSuccessfulSubscription = System.currentTimeMillis();

try {
fLogger = Logger.getLogger("eTactica.mqtt.listener");
FileHandler handler = new FileHandler(listenerLogFile);
fLogger.addHandler(handler);
} catch (IOException e) {
System.out.println("Logger - Failed");
}

try {
mqtt.setHost(listenerHostURI);
} catch (URISyntaxException e) {
stderr("setHost failed: " + e);
stderr(e);
}
QoS qos = QoS.AT_MOST_ONCE;
topics = new ArrayList();
topics.add(new Topic(listenerTopic, qos));
}

private void stdout(String x) {
if (listenerDebug) {
fLogger.log(Level.INFO, x + NEW_LINE);
}
}

private void stderr(String x) {
if (listenerDebug) {
fLogger.log(Level.SEVERE, x + NEW_LINE);
}
}

private void stderr(Throwable e) {
if (listenerDebug) {
StringWriter sw = new StringWriter();
PrintWriter pw = new PrintWriter(sw);
e.printStackTrace(pw);

fLogger.log(Level.SEVERE, sw.toString() + NEW_LINE);
}
}

private void subscriptionSuccessful() {
listenerLastSuccessfulSubscription = System.currentTimeMillis();
}

private boolean tryToListen() {
return ((System.currentTimeMillis() - listenerLastSuccessfulSubscription) < listenerMaxReAttemptDurationInSeconds * 1000);
}

private void sleepBeforeReAttempt() throws InterruptedException {
stdout(String.format(("Listener stopped, re-attempt in %s seconds."), listenerSleepBeforeReAttemptInSeconds));
Thread.sleep(listenerSleepBeforeReAttemptInSeconds);
}

private void listenerReAttemptsOver() {
stdout(String.format(("Listener stopped since reattempts have failed for %s seconds."), listenerMaxReAttemptDurationInSeconds));
}

private void listen() {
final CallbackConnection connection = mqtt.callbackConnection();
final CountDownLatch done = new CountDownLatch(1);

/* Runtime.getRuntime().addShutdownHook(new Thread(){
@Override
public void run() {
setName("MQTT client shutdown");
stderr("Disconnecting the client.");

connection.getDispatchQueue().execute(new Runnable() {
public void run() {
connection.disconnect(new Callback() {
public void onSuccess(Void value) {
stdout("Disconnecting onSuccess.");
done.countDown();
}
public void onFailure(Throwable value) {
stderr("Disconnecting onFailure: " + value);
stderr(value);
done.countDown();
}
});
}
});
}
});
*/

connection.listener(new org.fusesource.mqtt.client.Listener() {

public void onConnected() {
stdout("Listener onConnected");
}

public void onDisconnected() {
stdout("Listener onDisconnected");
}

public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
stdout(topic + " --> " + body.toString());
ack.run();
}

public void onFailure(Throwable value) {
stdout("Listener onFailure: " + value);
stderr(value);
done.countDown();
}
});

connection.resume();

connection.connect(new Callback() {
public void onFailure(Throwable value) {
stderr("Connect onFailure...: " + value);
stderr(value);
done.countDown();
}

public void onSuccess(Void value) {
final Topic[] ta = topics.toArray(new Topic[topics.size()]);
connection.subscribe(ta, new Callback() {
public void onSuccess(byte[] value) {
for (int i = 0; i < value.length; i++) {
stdout("Subscribed to Topic: " + ta[i].name() + " with QoS: " + QoS.values()[value[i]]);
}
subscriptionSuccessful();
}
public void onFailure(Throwable value) {
stderr("Subscribe failed: " + value);
stderr(value);
done.countDown();
}
});
}
});

try {
done.await();
} catch (Exception e) {
stderr(e);
}
}

@Override
public void run() {
while (tryToListen()) {
initMQTT();
listen();

try {
sleepBeforeReAttempt();
} catch (InterruptedException e) {
stderr("Sleep failed:" + e);
stderr(e);
}
}

listenerReAttemptsOver();
}

}