0.59.0

- Changes to support invoke method on broker.
- Fix bug in streaming logs.
- InboundRequest.getParams() -> getParameters()
- OutboundRequestHandler.getParams() -> getParameters()
- InboundListRequest added()->update(), changed()->update(), removed()->remove()

Author: a-hansen

build.gradle

@@ -5,14 +5,14 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.58.0'
+    version '0.59.0'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8
 
     repositories {
         mavenLocal()
-        mavenCentral()
+        jcenter()
     }
 
     task sourcesJar(group: 'build', type: Jar, dependsOn: classes) {
@@ -45,5 +45,5 @@ allprojects {
 }
 
 wrapper {
-    gradleVersion = '5.2'
+    gradleVersion = '5.4'
 }

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

@@ -40,6 +40,10 @@ public OutboundInvokeHandler getHandler() {
     @Override
     public void handleResponse(DSMap response) {
         try {
+            if (handler instanceof RawHandler) {
+                ((RawHandler) handler).handleRaw(response);
+                return;
+            }
             DSList updates = response.getList("updates");
             DSMap meta = response.getMap("meta");
             if (meta != null) {
@@ -112,4 +116,10 @@ protected DSMap getParams() {
         return params;
     }
 
+    public interface RawHandler {
+
+        public void handleRaw(DSMap response);
+
+    }
+
 }

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

@@ -159,7 +159,7 @@ public boolean isStreamOpen() {
         }
 
         @Override
-        public DSIValue getParams() {
+        public DSIValue getParameters() {
             return null;
         }
 

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

@@ -39,6 +39,7 @@ public class DSInboundInvoke extends DSInboundRequest
     private static final int STATE_UPDATES = 2;
     private static final int STATE_CLOSE_PENDING = 3;
     private static final int STATE_CLOSED = 4;
+    private static final int STATE_RAW = 10;
 
     ///////////////////////////////////////////////////////////////////////////
     // Instance Fields
@@ -156,6 +157,9 @@ public void run() {
                 DSIResponder responder = (DSIResponder) path.getTarget();
                 setPath(path.getPath());
                 result = responder.onInvoke(this);
+                if (result instanceof RawActionResult) {
+                    state = STATE_RAW;
+                }
             } else {
                 DSInfo info = path.getTargetInfo();
                 if (!info.isAction()) {
@@ -193,6 +197,10 @@ public void send(DSList row) {
         enqueueUpdate(new Update(row));
     }
 
+    public void sendRaw(DSMap raw) {
+        enqueueUpdate(new Update(raw));
+    }
+
     /**
      * For v2 only, set to false to auto close the stream after sending the initial state.
      */
@@ -214,6 +222,9 @@ public boolean write(DSSession session, MessageWriter writer) {
         }
         writeBegin(writer);
         switch (state) {
+            case STATE_RAW:
+                writeRaw(writer);
+                break;
             case STATE_INIT:
                 writeColumns(writer);
                 writeInitialResults(writer);
@@ -429,6 +440,36 @@ private void writeInitialResults(MessageWriter writer) {
         }
     }
 
+    private void writeRaw(MessageWriter writer) {
+        DSIWriter out = writer.getWriter();
+        Update update = updateHead; //peak ahead
+        if (update == null) {
+            return;
+        }
+        DSMap map;
+        DSResponder responder = getResponder();
+        while (true) {
+            update = dequeueUpdate();
+            map = update.raw;
+            map.remove("rid");
+            for (DSMap.Entry e : map) {
+                out.key(e.getKey());
+                out.value(e.getValue());
+            }
+            if (updateHead == null) {
+                break;
+            }
+            if (responder.shouldEndMessage()) {
+                enqueueResponse();
+                break;
+            }
+        }
+        String stream = map.get("stream", "");
+        if (stream.equals("closed")) {
+            doClose();
+        }
+    }
+
     private void writeUpdates(MessageWriter writer) {
         DSIWriter out = writer.getWriter();
         Update update = updateHead; //peak ahead
@@ -474,7 +515,8 @@ private void writeUpdates(MessageWriter writer) {
     protected enum UpdateType {
         INSERT("insert"),
         REFRESH("refresh"),
-        REPLACE("replace");
+        REPLACE("replace"),
+        RAW("raw");
 
         private String display;
 
@@ -487,6 +529,13 @@ public String toString() {
         }
     }
 
+    /**
+     * Used as a pass through on the broker.
+     */
+    public interface RawActionResult extends ActionResult {
+
+    }
+
     /**
      * Describes an update to be sent to the requester.
      */
@@ -495,10 +544,16 @@ protected static class Update {
         int beginIndex = -1;
         int endIndex = -1;
         Update next;
+        DSMap raw;
         DSList row;
         DSList[] rows;
         UpdateType type;
 
+        Update(DSMap raw) {
+            this.raw = raw;
+            this.type = UpdateType.RAW;
+        }
+
         Update(DSList row) {
             this.row = row;
         }

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

@@ -77,7 +77,7 @@ public class DSInboundList extends DSInboundRequest
     ///////////////////////////////////////////////////////////////////////////
 
     @Override
-    public void added(String name, ApiObject child) {
+    public void update(String name, ApiObject child) {
         if (!isClosed()) {
             enqueue(new AddUpdate(name, child));
         }
@@ -89,7 +89,7 @@ public boolean canWrite(DSSession session) {
     }
 
     @Override
-    public void changed(String name, DSElement value) {
+    public void update(String name, DSElement value) {
         if (!isClosed()) {
             enqueue(new ChangeUpdate(name, value));
         }
@@ -160,20 +160,20 @@ public void onClosed(DSISubscription subscription) {
     public void onEvent(DSEvent event, DSNode node, DSInfo child, DSIValue data) {
         switch (event.getEventId()) {
             case DSNode.CHILD_ADDED:
-                added(child.getName(), child);
+                update(child.getName(), child);
                 break;
             case DSNode.CHILD_RENAMED:
-                removed(data.toString());
-                added(child.getName(), child);
+                remove(data.toString());
+                update(child.getName(), child);
                 break;
             case DSNode.CHILD_REMOVED:
-                removed(child.getName());
+                remove(child.getName());
                 break;
         }
     }
 
     @Override
-    public void removed(String name) {
+    public void remove(String name) {
         if (!isClosed()) {
             enqueue(new RemoveUpdate(name));
         }