0.56.0

- Eliminate dependencies between specific link connections and DSSession.
- Fix blasting of broker with empty messages.  Better tracking whether ack required.
- Add DSISession and DSITransport - in public api.
- Eliminate dstranport dependencies (apis) that won't be used in non-blocking IO (future).
- Fix msgpack bug.
- Fix handshake with tokens.
- Adds DSSession stats.

Author: a-hansen

build.gradle

@@ -5,7 +5,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.55.0'
+    version '0.56.0'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8

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

@@ -1,6 +1,5 @@
 package com.acuity.iot.dsa.dslink.io;
 
-import com.acuity.iot.dsa.dslink.transport.DSTransport;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.PrintStream;
@@ -559,15 +558,6 @@ public int sendTo(ByteBuffer buf) {
         return len;
     }
 
-    /**
-     * Push bytes from the internal buffer to the transport.
-     */
-    public void sendTo(DSTransport transport, boolean isLast) {
-        transport.write(buffer, offset, length, isLast);
-        offset = 0;
-        length = 0;
-    }
-
     /**
      * Push bytes from the internal buffer to the given.
      */

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

@@ -1,7 +1,6 @@
 package com.acuity.iot.dsa.dslink.io.msgpack;
 
 import com.acuity.iot.dsa.dslink.io.DSByteBuffer;
-import com.acuity.iot.dsa.dslink.transport.DSTransport;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
@@ -83,10 +82,10 @@ public void writeTo(ByteBuffer out) {
 
     /**
      * Writes the internal buffer to the parameter.  The internal buffer will be cleared.
+     public void writeTo(DSITransport out) {
+     byteBuffer.sendTo(out, (frame == null));
+     }
      */
-    public void writeTo(DSTransport out) {
-        byteBuffer.sendTo(out, (frame == null));
-    }
 
     /**
      * Writes the internal buffer to the parameter.  The internal buffer will be cleared.

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

@@ -1,31 +1,45 @@
 package com.acuity.iot.dsa.dslink.protocol;
 
+import org.iot.dsa.dslink.DSLinkConnection;
 import org.iot.dsa.node.DSInfo;
 import org.iot.dsa.node.DSString;
 import org.iot.dsa.node.action.ActionInvocation;
 import org.iot.dsa.node.action.ActionResult;
 import org.iot.dsa.node.action.DSAction;
+import org.iot.dsa.util.DSException;
 
 /**
  * Represents an upstream connection to a broker.
  *
  * @author Aaron Hansen
  */
-public abstract class DSBrokerConnection extends DSTransportConnection {
+public abstract class DSBrokerConnection extends DSLinkConnection {
 
     ///////////////////////////////////////////////////////////////////////////
     // Class Fields
     ///////////////////////////////////////////////////////////////////////////
 
-    protected static final String BROKER_URI = "Broker URI";
+
+    protected static final String BROKER_ID = "Broker ID";
+    protected static final String BROKER_FORMAT = "Broker Format";
     protected static final String BROKER_PATH = "Path In Broker";
+    protected static final String BROKER_PUB_KEY = "Broker Public Key";
+    protected static final String BROKER_SALT = "Broker Salt";
+    protected static final String BROKER_URI = "Broker URI";
+    protected static final String BROKER_VERSION = "Broker Version";
     protected static final String RECONNECT = "Reconnect";
 
     ///////////////////////////////////////////////////////////////////////////
     // Instance Fields
     ///////////////////////////////////////////////////////////////////////////
 
+    private DSInfo brokerDsId = getInfo(BROKER_ID);
+    private DSInfo brokerFormat = getInfo(BROKER_FORMAT);
     private DSInfo brokerPath = getInfo(BROKER_PATH);
+    private DSInfo brokerPubKey = getInfo(BROKER_PUB_KEY);
+    private DSInfo brokerSalt = getInfo(BROKER_SALT);
+    private DSInfo brokerUri = getInfo(BROKER_URI);
+    private DSInfo brokerVersion = getInfo(BROKER_VERSION);
 
     ///////////////////////////////////////////////////////////////////////////
     // Public Methods
@@ -38,15 +52,27 @@ public String getPathInBroker() {
         return brokerPath.getElement().toString();
     }
 
+    /**
+     * For the sessions to update
+     */
+    public void setBrokerSalt(String arg) {
+        put(brokerSalt, DSString.valueOf(arg));
+    }
+
     ///////////////////////////////////////////////////////////////////////////
     // Protected Methods
     ///////////////////////////////////////////////////////////////////////////
 
     @Override
     protected void declareDefaults() {
         super.declareDefaults();
-        declareDefault(BROKER_URI, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_ID, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_FORMAT, DSString.NULL).setTransient(true).setReadOnly(true);
         declareDefault(BROKER_PATH, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_PUB_KEY, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_SALT, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_URI, DSString.NULL).setTransient(true).setReadOnly(true);
+        declareDefault(BROKER_VERSION, DSString.NULL).setTransient(true).setReadOnly(true);
         declareDefault(RECONNECT, new DSAction.Parameterless() {
             @Override
             public ActionResult invoke(DSInfo target, ActionInvocation invocation) {
@@ -56,6 +82,43 @@ public ActionResult invoke(DSInfo target, ActionInvocation invocation) {
         });
     }
 
+    @Override
+    protected void doConnect() {
+        try {
+            initializeConnection();
+            getTransport().open();
+            connOk();
+        } catch (Exception x) {
+            connDown(DSException.makeMessage(x));
+        }
+    }
+
+    protected String getBrokerFormat() {
+        return brokerFormat.get().toString();
+    }
+
+    protected String getBrokerId() {
+        return brokerDsId.get().toString();
+    }
+
+    protected String getBrokerKey() {
+        return brokerPubKey.get().toString();
+    }
+
+    protected String getBrokerSalt() {
+        return brokerSalt.get().toString();
+    }
+
+    protected String getBrokerUri() {
+        return brokerUri.get().toString();
+    }
+
+    protected String getBrokerVersion() {
+        return brokerVersion.get().toString();
+    }
+
+    protected abstract void initializeConnection();
+
     /**
      * Creates and starts a thread for running the connection lifecycle.
      */
@@ -67,6 +130,26 @@ protected void onStable() {
         t.start();
     }
 
+    protected void setBrokerFormat(String arg) {
+        put(brokerFormat, DSString.valueOf(arg));
+    }
+
+    protected void setBrokerId(String arg) {
+        put(brokerDsId, DSString.valueOf(arg));
+    }
+
+    protected void setBrokerKey(String arg) {
+        put(brokerPubKey, DSString.valueOf(arg));
+    }
+
+    protected void setBrokerUri(String arg) {
+        put(brokerUri, DSString.valueOf(arg));
+    }
+
+    protected void setBrokerVersion(String arg) {
+        put(brokerVersion, DSString.valueOf(arg));
+    }
+
     protected void setPathInBroker(String path) {
         put(brokerPath, DSString.valueOf(path));
     }

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

@@ -12,7 +12,7 @@
 import org.iot.dsa.util.DSUtil;
 
 /**
- * Link implementation for standalone v1 and v2 links.
+ * Links that also the root of the node tree.  These links have sys and upstream children.
  *
  * @author Aaron Hansen
  */
@@ -26,7 +26,6 @@ public class DSRootLink extends DSLink {
     // Instance Fields
     ///////////////////////////////////////////////////////////////////////////
 
-    private DSKeys keys;
     private DSInfo main;
     private DSInfo sys = getInfo(SYS);
     private DSInfo upstream;
@@ -39,13 +38,6 @@ public class DSRootLink extends DSLink {
     // Public Methods
     ///////////////////////////////////////////////////////////////////////////
 
-    /**
-     * Public / private keys of the link, used to prove identity with brokers.
-     */
-    public DSKeys getKeys() {
-        return keys;
-    }
-
     @Override
     public DSMainNode getMain() {
         return (DSMainNode) main.getNode();
@@ -67,13 +59,13 @@ public DSLinkConnection getUpstream() {
     @Override
     protected void declareDefaults() {
         super.declareDefaults();
-        declareDefault(SYS, new DSSysNode(), "Services common to all links.").setAdmin(true);
+        declareDefault(SYS, new DSSysNode(),
+                       "Services common to all links in the process.").setAdmin(true);
     }
 
     @Override
     protected DSLink init(DSLinkOptions config) {
         super.init(config);
-        keys = config.getKeys();
         main = getInfo(MAIN);
         if (main == null) {
             String type = getOptions().getMainType();