1

I have a Java MQTT subscription client that uses the subscribed message as input to update 2 tables in a SQLite database. The tables, data and queries are light-weight and simple UPDATE and INSERT operations. Here is what I see:

  • Message received on the subscription topic.
  • The messageArrived callback kicks off a database operation.
  • Database updated successfully.
  • Control comes to connectionLost callback.
  • Client (i.e. the Java process) terminates.

When the database operation is replaced with a System.out.println the behavior is normal. I tried fiddling with the keepAlive value in the connection options thinking that, the database operation may impact the PINGREQ and PINGRESP (even though, the client is in the 'background').

Can you please suggest, why is the client losing connection?

Java MQTT client:

public class wSync implements MqttCallback {
    MqttClient w;

    private String BROKER_URL = "";
    private String PROTOCOL = "tcp://";
    private String PORT = "1883";
    private String TOPIC_ROOT_UNCNF = "/P/uncnf/";
    private String TOPIC_ROOT_CNF = "/P/cnf/";
    private int KEEP_ALIVE = 180;

    private we WE = new WE();

    public static void main(String[] args) {
        wSync wsy = new wSync();
        processTxns(args[0]);
    }

    public void setBrokerUrl() {
        this.BROKER_URL = PROTOCOL + System.getenv("BROKER_DNS") + ":" + PORT;
    }

    public String getBrokerUrl() {
        return this.BROKER_URL;
    }

    public void processTxns(String wName) {
        this.setBrokerUrl();

        String broker = this.getBrokerUrl();
        String clientId = wName + "-SUB";
        String topic = TOPIC_ROOT_CNF + wName + "/CR";
        MemoryPersistence persistence = new MemoryPersistence();   

        try {
            MqttConnectOptions c = new MqttConnectOptions();
            c.setKeepAliveInterval(KEEP_ALIVE);
            c.setCleanSession(false);

            w = new MqttClient(broker, clientId, persistence);
            w.connect(c);
            w.setCallback(this);
            w.subscribe(topic, 2);
            System.out.println(w.getServerURI() + " " + w.getClientId() + " " + w.isConnected());

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void connectionLost(Throwable arg0) {
        System.out.println("Connection lost at : " + new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss").format(new java.util.Date()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken arg0) {
        // TODO Auto-generated method stub

    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String s = new String(message.getPayload());
        we.addSeed(s);
    }
}

UPDATE

Since I am not sure, why the disconnection happens, I decided to handle it. After all, network errors in field deployment is not unheard of. So, I re-wrote the connectionLost callback thusly:

@Override
public void connectionLost(Throwable arg0) {
    System.out.println("A " + w.isConnected() + " " + new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss").format(new java.util.Date()));

    try {
        w.connect(mqttConn);
        w.setCallback(this);
        w.subscribe("/P/cnf/wA/CR", 2);
    } catch (MqttSecurityException e) {
        e.printStackTrace();
    } catch (MqttException e) {
        e.printStackTrace();
    }

    System.out.println("B " + w.isConnected() + " " + new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss").format(new java.util.Date()));
}

This seems to handle disconnection and receipt of messages published on to the topic. However, when left to itself (meaning no publications received), I see these error messages:

champ@champ:~/bin$ java -jar w-sync.jar pull wA
* true 2017-02-19.23:21:00
{
  "meta" : {"tStamp" : "current_Date"},
  "data" : 
    [
      {"txn" : "thingy,thingamajig,whatchamacallit"}
    ]
}

A false 2017-02-19.23:21:29
B true 2017-02-19.23:21:29
A false 2017-02-19.23:22:29
A false 2017-02-19.23:22:29
B true 2017-02-19.23:22:29
MqttException (0) - java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:171)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at java.util.Vector.removeElementAt(Vector.java:558)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:137)
    ... 1 more
B true 2017-02-19.23:22:29
A false 2017-02-19.23:24:29
MqttException (0) - java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:171)
    at java.lang.Thread.run(Thread.java:745)
A false 2017-02-19.23:24:29
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at java.util.Vector.removeElementAt(Vector.java:558)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:137)
    ... 1 more
B false 2017-02-19.23:24:29
MqttException (0) - java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:171)
    at java.lang.Thread.run(Thread.java:745)
A false 2017-02-19.23:24:29
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0 >= 0
    at java.util.Vector.removeElementAt(Vector.java:558)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:137)
    ... 1 more
B false 2017-02-19.23:24:29
Client is not connected (32104)
    at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:27)
    at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:132)
    at org.eclipse.paho.client.mqttv3.MqttAsyncClient.subscribe(MqttAsyncClient.java:657)
    at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(MqttClient.java:282)
    at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(MqttClient.java:275)A false 2017-02-19.23:24:29

    at com.rpk.demo.sync.wSync.connectionLost(wSync.java:109)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.connectionLost(CommsCallback.java:229)
    at org.eclipse.paho.client.mqttv3.internal.ClientComms.shutdownConnection(ClientComms.java:339)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:171)
    at java.lang.Thread.run(Thread.java:745)
B false 2017-02-19.23:24:29
B true 2017-02-19.23:24:29
6
  • You can try catch Exception(Throwable) wrapper arround the database operation to check whether any exception is thrown. BTW, there is no while..loop or something like that in your code to "listen" from the broker. Likely, the Main method will be terminated very early?
    – NangSaigon
    Feb 19, 2017 at 17:04
  • Actually, the database operation goes through fine; I can see the updates. And, I don't need any loop; the client keeps running in background. See this example: stackoverflow.com/a/22716162/919480 Feb 19, 2017 at 17:08
  • Are you calling the processTxns method multiple times with the same wName?
    – hardillb
    Feb 19, 2017 at 17:10
  • No; while I am currently invoking via Eclipse, the idea is to do a java -jar $RELEVANT_JAR args0 args1 and then let each subscribed message do a database update on its own. Are you referring to the case where there is a situation of two clients trying to connect with same client ID? Feb 19, 2017 at 17:17
  • Further update. When disconnected, the ((MqttException)arg0).getReasonCode() and ((MqttException) arg0).getCause() are respectively 0 and java.lang.ArrayIndexOutOfBoundsException: 3. Subsequently, the cause becomes, java.lang.ArrayIndexOutOfBoundsException: 0 >=0 . Does this mean any thing? Feb 19, 2017 at 18:49

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.