- Another topic refactor.

- DSNode subscription refactor.
- DynamicActions -> VirtualActions

Author: Aaron

dslink-v2-poc/src/main/java/org/iot/dsa/dslink/poc/MainNode.java

@@ -53,7 +53,7 @@ public class MainNode extends DSMainNode implements Runnable {
     ///////////////////////////////////////////////////////////////////////////
 
     @Override
-    public DSInfo getDynamicAction(DSInfo target, String name) {
+    public DSInfo getVirtualAction(DSInfo target, String name) {
         if (target == incrementingInt) {
             if (name.equals("Reset")) {
                 return actionInfo(name, new DSAction.Parameterless() {
@@ -75,17 +75,17 @@ public ActionResult invoke(DSInfo target, ActionInvocation invocation) {
                 });
             }
         }
-        return super.getDynamicAction(target, name);
+        return super.getVirtualAction(target, name);
     }
 
     @Override
-    public void getDynamicActions(DSInfo target, Collection<String> bucket) {
+    public void getVirtualActions(DSInfo target, Collection<String> bucket) {
         if (target == incrementingInt) {
             bucket.add("Reset");
         } else if (target.get() == this) {
             bucket.add("Foobar");
         }
-        super.getDynamicActions(target, bucket);
+        super.getVirtualActions(target, bucket);
     }
 
     @Override

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

@@ -156,7 +156,7 @@ private void advance(int nameIndex) {
             node = (DSNode) target;
             targetInfo = node.getInfo(names[nameIndex]);
         } else {
-            targetInfo = node.getDynamicAction(targetInfo, names[nameIndex]);
+            targetInfo = node.getVirtualAction(targetInfo, names[nameIndex]);
         }
         if (targetInfo == null) {
             //TODO - need to undo this

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

@@ -26,7 +26,7 @@
 import org.iot.dsa.node.action.DSAction;
 import org.iot.dsa.node.event.DSIEvent;
 import org.iot.dsa.node.event.DSISubscriber;
-import org.iot.dsa.node.event.DSITopic;
+import org.iot.dsa.node.event.DSISubscription;
 import org.iot.dsa.node.event.DSNodeTopic;
 import org.iot.dsa.security.DSPermission;
 
@@ -64,6 +64,7 @@ public class DSInboundList extends DSInboundRequest
     private OutboundListResponse response;
     private int state = STATE_INIT;
     private boolean stream = true;
+    private DSISubscription subscription;
     private DSTarget target;
     private Update updateHead;
     private Update updateTail;
@@ -127,8 +128,9 @@ public boolean isOpen() {
 
     @Override
     public void onClose() {
-        if (node != null) {
-            node.unsubscribe(null, null, this);
+        if (subscription != null) {
+            subscription.close();
+            subscription = null;
         }
     }
 
@@ -145,6 +147,12 @@ public void onClose(Integer requestId) {
         doClose();
     }
 
+    @Override
+    public void onClosed(DSISubscription subscription) {
+        subscription = null;
+        close();
+    }
+
     @Override
     public void onEvent(DSNode node, DSInfo child, DSIEvent event) {
         switch ((DSNodeTopic) event.getTopic()) {
@@ -165,11 +173,6 @@ public void onEvent(DSNode node, DSInfo child, DSIEvent event) {
         }
     }
 
-    @Override
-    public void onUnsubscribed(DSITopic topic, DSNode node, DSInfo child) {
-        close();
-    }
-
     @Override
     public void run() {
         try {
@@ -185,7 +188,7 @@ public void run() {
                 }
                 if (info.isNode()) {
                     node = info.getNode();
-                    node.subscribe(null, null, null, this);
+                    this.subscription = node.subscribe(null, null, null, this);
                 }
                 response = this;
             }
@@ -366,10 +369,10 @@ protected String encodeName(String name, StringBuilder buf) {
      */
     protected void encodeUpdate(Update update, MessageWriter writer, StringBuilder buf) {
         if (update instanceof AddUpdate) {
-            encodeChild(((AddUpdate)update).child, writer);
+            encodeChild(((AddUpdate) update).child, writer);
         } else {
             writer.getWriter().beginMap()
-                  .key("name").value(encodeName(((RemoveUpdate)update).name, buf))
+                  .key("name").value(encodeName(((RemoveUpdate) update).name, buf))
                   .key("change").value("remove")
                   .endMap();
         }

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

@@ -15,6 +15,7 @@
 import org.iot.dsa.node.DSStatus;
 import org.iot.dsa.node.event.DSIEvent;
 import org.iot.dsa.node.event.DSISubscriber;
+import org.iot.dsa.node.event.DSISubscription;
 import org.iot.dsa.node.event.DSITopic;
 import org.iot.dsa.time.DSTime;
 
@@ -40,6 +41,7 @@ public class DSInboundSubscription extends DSInboundRequest
     private boolean open = true;
     private int qos = 0;
     private Integer sid;
+    private DSISubscription subscription;
     private Update updateHead;
     private Update updateTail;
 
@@ -117,7 +119,8 @@ public void onEvent(DSNode node, DSInfo child, DSIEvent event) {
     }
 
     @Override
-    public void onUnsubscribed(DSITopic topic, DSNode node, DSInfo info) {
+    public void onClosed(DSISubscription subscription) {
+        this.subscription = null;
         close();
     }
 
@@ -218,13 +221,13 @@ protected void init() {
             DSIObject obj = path.getTarget();
             if (obj instanceof DSNode) {
                 node = (DSNode) obj;
-                node.subscribe(DSNode.VALUE_CHANGED, null, null, this);
+                this.subscription = node.subscribe(DSNode.VALUE_CHANGED, null, null, this);
                 onEvent(node, null, DSNode.VALUE_CHANGED);
             } else {
                 DSInfo info = path.getTargetInfo();
                 node = path.getNode();
                 child = info;
-                node.subscribe(DSNode.VALUE_CHANGED, info, null, this);
+                this.subscription = node.subscribe(DSNode.VALUE_CHANGED, info, null, this);
                 onEvent(node, info, DSNode.VALUE_CHANGED);
             }
         }
@@ -308,15 +311,16 @@ void onClose() {
             open = false;
         }
         try {
-            if (closeHandler != null) {
-                closeHandler.onClose(getSubscriptionId());
+            if (subscription != null) {
+                subscription.close();
+                subscription = null;
             }
         } catch (Exception x) {
             manager.debug(manager.getPath(), x);
         }
         try {
-            if (node != null) {
-                node.unsubscribe(DSNode.VALUE_CHANGED, child, this);
+            if (closeHandler != null) {
+                closeHandler.onClose(getSubscriptionId());
             }
         } catch (Exception x) {
             manager.debug(manager.getPath(), x);

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

@@ -149,13 +149,13 @@ public Iterator<ApiObject> getChildren() {
         DSNode node;
         if (isNode()) {
             node = getNode();
-            node.getDynamicActions(this, actions);
+            node.getVirtualActions(this, actions);
         } else {
             node = getParent();
-            node.getDynamicActions(this, actions);
+            node.getVirtualActions(this, actions);
         }
         for (String s : actions) {
-            info = node.getDynamicAction(this, s);
+            info = node.getVirtualAction(this, s);
             if (info != null) {
                 ret.add(info);
             }
@@ -538,7 +538,7 @@ int getFlags() {
     /**
      * Whether or not the info is for a dynamic action.
      */
-    boolean isDynamic() {
+    boolean isVirtual() {
         return false;
     }