0.31.0

Ack Tracking

Author: a-hansen

README.md

@@ -1,10 +1,9 @@
-# sdk-dslink-java-v2 (PRE-ALPHA)
+# sdk-dslink-java-v2
+[![](https://jitpack.io/v/iot-dsa-v2/sdk-dslink-java-v2.svg)](https://jitpack.io/#iot-dsa-v2/sdk-dslink-java-v2)
 
-* Version: 0.0.0
-* JDK 1.6 Compatibility is required.
 * [Developer Guide](https://iot-dsa-v2.github.io/sdk-dslink-java-v2/)
 * [Javadoc](https://iot-dsa-v2.github.io/sdk-dslink-java-v2/javadoc/)
-* [Apache License 2.0](http://www.apache.org/licenses/LICENSE-2.0)
+* JDK 1.6+
 
 
 ## Overview
@@ -38,8 +37,3 @@ _Silk Icons_
 This software uses icons from Silk Icons 1.3 created by 
 [Mark James](http://www.famfamfam.com/lab/icons/silk/) and licensed 
 under a [Creative Commons Attribute 2.5 License](http://creativecommons.org/licenses/by/2.5/).
-
-## History
-
-* 1.0.0
-  - Hello World.

build.gradle

@@ -4,7 +4,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.30.0'
+    version '0.31.0'
 
     sourceCompatibility = 1.6
     targetCompatibility = 1.6

dslink-v2-poc/dslink.json

@@ -12,7 +12,7 @@
     "log": {
       "desc": "all, trace, debug, fine, warn, info, error, admin, fatal, none",
       "type": "enum",
-      "value": "all"
+      "value": "warn"
     },
     "token": {
       "desc": "Authentication token for the broker.",

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

@@ -1,6 +1,7 @@
 package com.acuity.iot.dsa.dslink.protocol;
 
 import com.acuity.iot.dsa.dslink.protocol.message.OutboundMessage;
+import com.acuity.iot.dsa.dslink.protocol.responder.DSResponder;
 import com.acuity.iot.dsa.dslink.transport.DSTransport;
 import java.io.IOException;
 import java.util.LinkedList;
@@ -21,18 +22,20 @@ public abstract class DSSession extends DSNode {
     // Class Fields
     ///////////////////////////////////////////////////////////////////////////
 
-    private static final int MAX_MSG_ID = 2147483647;
+    private static final int MAX_MSG_ID = Integer.MAX_VALUE;
     private static final long MSG_TIMEOUT = 60000;
 
     ///////////////////////////////////////////////////////////////////////////
     // Instance Fields
     ///////////////////////////////////////////////////////////////////////////
 
+    private int ackRcvd = -1;
+    private int ackToSend = -1;
     private boolean connected = false;
     private DSLinkConnection connection;
-    private long lastRecv;
-    private long lastSend;
-    private int nextAck = -1;
+    private long lastTimeRecv;
+    private long lastTimeSend;
+    private int messageId = 0;
     private int nextMessage = 1;
     private Object outgoingMutex = new Object();
     private List<OutboundMessage> outgoingRequests = new LinkedList<OutboundMessage>();
@@ -51,7 +54,7 @@ public DSSession(DSLinkConnection connection) {
     }
 
     ///////////////////////////////////////////////////////////////////////////
-    // Methods
+    // Public Methods
     ///////////////////////////////////////////////////////////////////////////
 
     /**
@@ -96,37 +99,37 @@ public void enqueueOutgoingResponse(OutboundMessage arg) {
         }
     }
 
-    public DSLinkConnection getConnection() {
-        return connection;
-    }
-
     /**
-     * Last ack received from the broker.
+     * Last ack received from the broker, or -1.
      */
-    public abstract long getLastAckRcvd();
+    public int getAckRcvd() {
+        return ackRcvd;
+    }
 
     /**
-     * The next ack id, or -1.
+     * The next ack id to send, or -1.
      */
-    public synchronized int getNextAck() {
-        int ret = nextAck;
-        nextAck = -1;
+    public synchronized int getAckToSend() {
+        int ret = ackToSend;
+        ackToSend = -1;
         return ret;
     }
 
+    public DSLinkConnection getConnection() {
+        return connection;
+    }
+
     /**
-     * Returns the next new message id.
+     * The current (last) message ID generated.
      */
-    public synchronized int getNextMessageId() {
-        int ret = nextMessage;
-        if (++nextMessage > MAX_MSG_ID) {
-            nextMessage = 1;
-        }
-        return ret;
+    public int getMessageId() {
+        return messageId;
     }
 
     public abstract DSIRequester getRequester();
 
+    public abstract DSResponder getResponder();
+
     public DSTransport getTransport() {
         return getConnection().getTransport();
     }
@@ -161,13 +164,13 @@ public void onDisconnect() {
      * implementation.  A separate thread is spun off to manage writing.
      */
     public void run() {
-        lastRecv = lastSend = System.currentTimeMillis();
+        lastTimeRecv = lastTimeSend = System.currentTimeMillis();
         new WriteThread(getConnection().getLink().getLinkName() + " Writer").start();
         while (connected) {
             try {
                 verifyLastSend();
                 doRecvMessage();
-                lastRecv = System.currentTimeMillis();
+                lastTimeRecv = System.currentTimeMillis();
             } catch (Exception x) {
                 getTransport().close();
                 if (connected) {
@@ -178,16 +181,6 @@ public void run() {
         }
     }
 
-    /**
-     * Call for each incoming message id that needs to be acked.
-     */
-    public synchronized void setNextAck(int nextAck) {
-        if (nextAck > 0) {
-            this.nextAck = nextAck;
-            notifyOutgoing();
-        }
-    }
-
     /**
      * Called when the broker signifies that requests are allowed.
      */
@@ -197,6 +190,10 @@ public void setRequesterAllowed() {
 
     public abstract boolean shouldEndMessage();
 
+    ///////////////////////////////////////////////////////////////////////////
+    // Protected Methods
+    ///////////////////////////////////////////////////////////////////////////
+
     /**
      * Can return null.
      */
@@ -238,8 +235,23 @@ protected String getLogName() {
         return "Session";
     }
 
+    protected int getMissingAcks() {
+        return messageId - ackRcvd - 1;
+    }
+
+    /**
+     * Returns the next new message id.
+     */
+    protected synchronized int getNextMessageId() {
+        messageId = nextMessage;
+        if (++nextMessage > MAX_MSG_ID) {
+            nextMessage = 1;
+        }
+        return messageId;
+    }
+
     protected boolean hasAckToSend() {
-        return nextAck > 0;
+        return ackToSend > 0;
     }
 
     protected boolean hasOutgoingRequests() {
@@ -254,9 +266,12 @@ protected boolean hasOutgoingResponses() {
      * Override point, this returns the result of hasMessagesToSend.
      */
     protected boolean hasSomethingToSend() {
-        if (nextAck > 0) {
+        if (ackToSend >= 0) {
             return true;
         }
+        if (waitingForAcks()) {
+            return false;
+        }
         if (!outgoingResponses.isEmpty()) {
             return true;
         }
@@ -279,14 +294,63 @@ protected void notifyOutgoing() {
         }
     }
 
+    protected int numOutgoingRequests() {
+        return outgoingRequests.size();
+    }
+
+    protected int numOutgoingResponses() {
+        return outgoingResponses.size();
+    }
+
+    protected void requeueOutgoingRequest(OutboundMessage arg) {
+        synchronized (outgoingMutex) {
+            outgoingRequests.add(arg);
+        }
+    }
+
+    protected void requeueOutgoingResponse(OutboundMessage arg) {
+        synchronized (outgoingMutex) {
+            outgoingResponses.add(arg);
+        }
+    }
+
+    /**
+     * Call for each incoming message id that needs to be acked.
+     */
+    protected void setAckRcvd(int ackRcvd) {
+        if (ackRcvd < this.ackRcvd) {
+            warn(warn() ? String.format("Ack rcvd %s < last %s", ackRcvd, this.ackRcvd) : null);
+        }
+        this.ackRcvd = ackRcvd;
+        notifyOutgoing();
+    }
+
+    /**
+     * Call for each incoming message id that needs to be acked.
+     */
+    protected void setAckToSend(int ackToSend) {
+        if (ackToSend > 0) {
+            this.ackToSend = ackToSend;
+            notifyOutgoing();
+        }
+    }
+
+    protected boolean waitingForAcks() {
+        return getMissingAcks() > 8;
+    }
+
+    ///////////////////////////////////////////////////////////////////////////
+    // Package / Private Methods
+    ///////////////////////////////////////////////////////////////////////////
+
     private void verifyLastRead() throws IOException {
-        if ((System.currentTimeMillis() - lastRecv) > MSG_TIMEOUT) {
+        if ((System.currentTimeMillis() - lastTimeRecv) > MSG_TIMEOUT) {
             throw new IOException("No message received in " + MSG_TIMEOUT + "ms");
         }
     }
 
     private void verifyLastSend() throws IOException {
-        if ((System.currentTimeMillis() - lastSend) > MSG_TIMEOUT) {
+        if ((System.currentTimeMillis() - lastTimeSend) > MSG_TIMEOUT) {
             throw new IOException("No message sent in " + MSG_TIMEOUT + "ms");
         }
     }
@@ -320,7 +384,7 @@ public void run() {
                         }
                     }
                     doSendMessage();
-                    lastSend = System.currentTimeMillis();
+                    lastTimeSend = System.currentTimeMillis();
                 }
             } catch (Exception x) {
                 if (connected) {