Fix reconnection issues

Author: Aaron

build.gradle

@@ -4,7 +4,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.32.0'
+    version '0.32.8'
 
     sourceCompatibility = 1.6
     targetCompatibility = 1.6
@@ -26,5 +26,5 @@ subprojects {
 }
 
 task wrapper(type: Wrapper) {
-    gradleVersion = '4.8.1'
+    gradleVersion = '4.9'
 }

dslink-v2-websocket/src/main/java/org/iot/dsa/dslink/websocket/WsTextTransport.java

@@ -60,7 +60,7 @@ public DSTransport close() {
             return this;
         }
         open = false;
-        debug(debug() ? "WsTextTransport.close()" : null, new Exception());
+        debug(debug() ? "WsTextTransport.close()" : null);
         try {
             if (session != null) {
                 session.close();

dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/io/DSCharBuffer.java

@@ -160,7 +160,6 @@ public synchronized void put(char[] msg, int off, int len) {
                 offset = 0;
             }
         }
-        //System.arraycopy(msg, off, buffer, length + offset, len);
         System.arraycopy(msg, off, buffer, offset, len);
         length += len;
         notifyAll();

dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/DSSession.java

@@ -240,10 +240,7 @@ protected boolean hasAckToSend() {
      * Override point, this returns the result of hasMessagesToSend.
      */
     protected boolean hasSomethingToSend() {
-        if (ackToSend >= 0) {
-            return true;
-        }
-        if (hasPingToSend()) {
+        if (hasAckToSend() || hasPingToSend()) {
             return true;
         }
         if (waitingForAcks()) {
@@ -282,8 +279,9 @@ protected void onConnected() {
         connected = true;
         lastTimeRecv = lastTimeSend = System.currentTimeMillis();
         readThread = new ReadThread(getConnection().getLink().getLinkName() + " Reader");
-        writeThread = new WriteThread(getConnection().getLink().getLinkName() + " Writer");
         readThread.start();
+        Thread.yield();
+        writeThread = new WriteThread(getConnection().getLink().getLinkName() + " Writer");
         writeThread.start();
     }
 
@@ -296,11 +294,6 @@ protected void onDisconnected() {
             outgoingResponses.clear();
             outgoingMutex.notifyAll();
         }
-        try {
-            writeThread.join();
-        } catch (Exception x) {
-            debug(getPath(), x);
-        }
         try {
             readThread.join();
         } catch (Exception x) {
@@ -319,6 +312,13 @@ protected void onDisconnecting() {
         }
         connected = false;
         notifyOutgoing();
+        try {
+            writeThread.join();
+        } catch (Exception x) {
+            debug(getPath(), x);
+        }
+        //Attempt to exit cleanly, try to get acks for sent messages.
+        waitForAcks(1000);
     }
 
     protected void requeueOutgoingRequest(OutboundMessage arg) {
@@ -385,6 +385,25 @@ private void verifyLastSend() throws IOException {
         }
     }
 
+    /* Try to exit cleanly, wait for all acks for sent messages. */
+    private void waitForAcks(long timeout) {
+        long start = System.currentTimeMillis();
+        synchronized (outgoingMutex) {
+            while (getMissingAcks() > 0) {
+                try {
+                    outgoingMutex.wait(500);
+                } catch (InterruptedException x) {
+                    warn(getPath(), x);
+                }
+                if ((System.currentTimeMillis() - start) > timeout) {
+                    debug(debug() ? String.format("witForAcks timeout (%s / %s)",ackRcvd,messageId)
+                                  : null);
+                    break;
+                }
+            }
+        }
+    }
+
     ///////////////////////////////////////////////////////////////////////////
     // Inner Classes
     ///////////////////////////////////////////////////////////////////////////
@@ -400,6 +419,7 @@ private class ReadThread extends Thread {
         }
 
         public void run() {
+            debug("Enter DSSession.ReadThread");
             DSLinkConnection conn = getConnection();
             try {
                 while (connected) {
@@ -415,6 +435,7 @@ public void run() {
                     conn.connDown(DSException.makeMessage(x));
                 }
             }
+            debug("Exit DSSession.ReadThread");
         }
     }
 
@@ -430,6 +451,7 @@ private class WriteThread extends Thread {
 
         public void run() {
             DSLinkConnection conn = getConnection();
+            debug("Enter DSSession.WriteThread");
             try {
                 while (connected) {
                     verifyLastRead();
@@ -454,6 +476,7 @@ public void run() {
                     conn.connDown(DSException.makeMessage(x));
                 }
             }
+            debug("Exit DSSession.WriteThread");
         }
     }
 

dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/requester/DSOutboundSubscriptions.java

@@ -95,6 +95,7 @@ public void handleUpdate(int sid, String ts, String sts, DSElement value) {
     @Override
     public void write(DSSession session, MessageWriter writer) {
         if (!pendingSubscribe.isEmpty()) {
+            debug(debug() ? "Sending subscribe requests" : null);
             doBeginSubscribe(writer);
             Iterator<DSOutboundSubscribeStubs> it = pendingSubscribe.iterator();
             while (it.hasNext() && !session.shouldEndMessage()) {
@@ -114,6 +115,7 @@ public void write(DSSession session, MessageWriter writer) {
             doEndMessage(writer);
         }
         if (!pendingUnsubscribe.isEmpty() && !session.shouldEndMessage()) {
+            debug(debug() ? "Sending unsubscribe requests" : null);
             doBeginUnsubscribe(writer);
             Iterator<DSOutboundSubscribeStubs> it = pendingUnsubscribe.iterator();
             while (it.hasNext() && !session.shouldEndMessage()) {
@@ -129,9 +131,7 @@ public void write(DSSession session, MessageWriter writer) {
             }
             doEndMessage(writer);
         }
-        synchronized (this) {
-            enqueued = false;
-        }
+        enqueued = false;
         if (!pendingSubscribe.isEmpty() || !pendingUnsubscribe.isEmpty()) {
             sendMessage();
         }
@@ -208,6 +208,7 @@ protected void onDisconnected() {
         }
         sidMap.clear();
         pathMap.clear();
+        enqueued = false;
     }
 
     ///////////////////////////////////////////////////////////////////////////
@@ -239,6 +240,7 @@ private void sendMessage() {
      * Create or update a subscription.
      */
     OutboundSubscribeHandler subscribe(String path, int qos, OutboundSubscribeHandler req) {
+        trace(trace() ? String.format("Subscribe (qos=%s) %s", qos, path) : null);
         DSOutboundSubscribeStub stub = new DSOutboundSubscribeStub(path, qos, req);
         DSOutboundSubscribeStubs stubs = null;
         synchronized (pathMap) {

dslink-v2/src/main/java/com/acuity/iot/dsa/dslink/protocol/v1/DS1Session.java

@@ -112,19 +112,19 @@ protected void doRecvMessage() throws IOException {
     @Override
     protected void doSendMessage() {
         try {
+            beginMessage();
             if (!waitingForAcks()) {
                 requestsNext = !requestsNext;
-                beginMessage();
                 send(requestsNext);
                 if (!shouldEndMessage()) {
                     send(!requestsNext);
                 }
-                endMessage();
                 lastMessageSent = System.currentTimeMillis();
                 if (requestsBegun || responsesBegun) {
                     setAckRequired();
                 }
             }
+            endMessage();
         } finally {
             requestsBegun = false;
             responsesBegun = false;
@@ -250,11 +250,9 @@ private DSIWriter getWriter() {
      * Decomposes and processes a complete envelope which can contain multiple requests and
      * responses.
      *
-     * @param reader lastRun() will return BEGIN_MAP
+     * @param reader last() must return BEGIN_MAP
      */
     private void processEnvelope(DSIReader reader) {
-        int msg = -1;
-        Token next;
         switch (reader.next()) {
             case END_MAP:
                 return;
@@ -263,6 +261,8 @@ private void processEnvelope(DSIReader reader) {
             default:
                 throw new IllegalStateException("Poorly formatted request");
         }
+        int msg = -1;
+        Token next;
         boolean sendAck = false;
         do {
             String key = reader.getString();

dslink-v2/src/main/java/org/iot/dsa/node/DSInfo.java

@@ -176,10 +176,13 @@ public DSIObject getDefaultObject() {
     }
 
     /**
-     * A convenience that casts getObject().
+     * A convenience that casts the object.  Will call DSIValue.toElement on values.
      */
     public DSElement getElement() {
-        return (DSElement) value;
+        if (value instanceof DSElement) {
+            return (DSElement) value;
+        }
+        return ((DSIValue)value).toElement();
     }
 
     boolean getFlag(int position) {
@@ -375,6 +378,20 @@ public DSInfo next() {
         return next;
     }
 
+    /**
+     * The next DSInfo in the parent whose is of the given type.
+     */
+    public DSInfo next(Class is) {
+        DSInfo cur = next;
+        while (cur != null) {
+            if (cur.is(is)) {
+                return cur;
+            }
+            cur = cur.next();
+        }
+        return cur;
+    }
+
     /**
      * The next DSInfo in the parent that is an action, or null.
      */

dslink-v2/src/main/java/org/iot/dsa/node/DSNode.java

@@ -360,6 +360,14 @@ public DSInfo getFirstInfo() {
         return firstChild;
     }
 
+    public DSInfo getFirstInfo(Class type) {
+        DSInfo info = getFirstInfo();
+        if (info.is(type)) {
+            return info;
+        }
+        return info.next(type);
+    }
+
     /**
      * The info for the first child node, or null.
      */

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.8.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.9-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists