0.66.0

- Fix raw invoke responses.
- Add $disconnectedTs to list responses on errors.

Author: a-hansen

build.gradle

@@ -5,14 +5,17 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.65.1'
+    version '0.66.0'
 
     targetCompatibility = JavaVersion.VERSION_1_8
     sourceCompatibility = JavaVersion.VERSION_1_8
 
     repositories {
-        mavenLocal()
         jcenter()
+        mavenLocal()
+        maven {
+            url "https://plugins.gradle.org/m2/"
+        }
     }
 
     jar {

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

@@ -39,7 +39,6 @@ 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
@@ -93,8 +92,7 @@ public void close(Exception reason) {
             return;
         }
         closeReason = reason;
-        state = STATE_CLOSE_PENDING;
-        enqueueResponse();
+        close();
     }
 
     /**
@@ -155,7 +153,8 @@ public void run() {
                 setPath(path.getPath());
                 result = responder.onInvoke(this);
                 if (result instanceof RawActionResult) {
-                    state = STATE_RAW;
+                    state = STATE_UPDATES;
+                    return;
                 }
             } else {
                 DSInfo info = path.getTargetInfo();
@@ -194,7 +193,11 @@ public void send(DSList row) {
         enqueueUpdate(new Update(row));
     }
 
+    /**
+     * For the broker to use as a pass-thru mechanism.
+     */
     public void sendRaw(DSMap raw) {
+        state = STATE_UPDATES;
         enqueueUpdate(new Update(raw));
     }
 
@@ -219,9 +222,6 @@ public boolean write(DSSession session, MessageWriter writer) {
         }
         writeBegin(writer);
         switch (state) {
-            case STATE_RAW:
-                writeRaw(writer);
-                break;
             case STATE_INIT:
                 writeColumns(writer);
                 writeInitialResults(writer);
@@ -306,14 +306,11 @@ private void doClose() {
         if (result == null) {
             return;
         }
-        DSRuntime.run(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    result.onClose();
-                } catch (Exception x) {
-                    error(getPath(), x);
-                }
+        DSRuntime.run(() -> {
+            try {
+                result.onClose();
+            } catch (Exception x) {
+                error(getPath(), x);
             }
         });
     }
@@ -437,29 +434,17 @@ private void writeInitialResults(MessageWriter writer) {
         }
     }
 
-    private void writeRaw(MessageWriter writer) {
+    /**
+     * Only called by writeUpdates.
+     */
+    private void writeRaw(Update update, 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;
-            }
+        map = update.raw;
+        map.remove("rid");
+        for (DSMap.Entry e : map) {
+            out.key(e.getKey());
+            out.value(e.getValue());
         }
         String stream = map.get("stream", "");
         if (stream.equals("closed")) {
@@ -468,11 +453,15 @@ private void writeRaw(MessageWriter writer) {
     }
 
     private void writeUpdates(MessageWriter writer) {
-        DSIWriter out = writer.getWriter();
         Update update = updateHead; //peak ahead
         if (update == null) {
             return;
         }
+        if (update.isRaw()) {
+            writeRaw(update, writer);
+            return;
+        }
+        DSIWriter out = writer.getWriter();
         if (update.type != null) {
             out.key("meta")
                .beginMap()
@@ -494,7 +483,8 @@ private void writeUpdates(MessageWriter writer) {
             if ((updateHead == null) || (updateHead.type != null)) {
                 break;
             }
-            if (responder.shouldEndMessage()) {
+            //raw messaging won't be mixed with normal messaging, but best to be safe
+            if (updateHead.isRaw() || responder.shouldEndMessage()) {
                 enqueueResponse();
                 break;
             }
@@ -569,6 +559,10 @@ protected static class Update {
             this.type = type;
         }
 
+        boolean isRaw() {
+            return type == UpdateType.RAW;
+        }
+
         String typeKey() {
             if ((type == UpdateType.INSERT) || (type == UpdateType.REPLACE)) {
                 return "modify";

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

@@ -31,6 +31,7 @@
 import org.iot.dsa.node.event.DSISubscriber;
 import org.iot.dsa.node.event.DSISubscription;
 import org.iot.dsa.security.DSPermission;
+import org.iot.dsa.time.DSDateTime;
 import org.iot.dsa.util.DSException;
 
 /**
@@ -95,9 +96,8 @@ public void close(Exception reason) {
         if (!isOpen()) {
             return;
         }
-        state = STATE_CLOSE_PENDING;
         closeReason = reason;
-        enqueueResponse();
+        close();
     }
 
     @Override
@@ -183,6 +183,7 @@ public void run() {
                 response = this;
             }
         } catch (Exception x) {
+            update("$disconnectedTs", DSDateTime.now().toElement());
             error(getPath(), x);
             close(x);
             return;
@@ -489,10 +490,6 @@ private void encodeTarget(ApiObject object, MessageWriter writer) {
      * Called by encodeTarget for actions.
      */
     private void encodeTargetAction(ApiObject object, MessageWriter writer) {
-        DSInfo info = null;
-        if (object instanceof DSInfo) {
-            info = (DSInfo) object;
-        }
         ActionSpec action = object.getAction();
         DSAction dsAction = null;
         if (action instanceof DSAction) {
@@ -651,9 +648,7 @@ private void fixRangeTypes(DSMap arg) {
             String type = arg.getString(DSMetadata.TYPE);
             if ("bool".equals(type)) {
                 DSList range = (DSList) arg.remove(DSMetadata.BOOLEAN_RANGE);
-                if ((range == null) || (range.size() != 2)) {
-                    return;
-                } else {
+                if ((range != null) || (range.size() == 2)) {
                     String utf8 = DSString.UTF8.toString();
                     cacheBuf.setLength(0);
                     cacheBuf.append(type);
@@ -784,6 +779,7 @@ protected static class RemoveUpdate extends Update {
         RemoveUpdate(String name) {
             this.name = name;
         }
+
     }
 
     private static class RootInfo extends DSInfo {