0.48.0

- Fix broken actions
- Gradle update
- Refactor simpleinvokehandler
- New action unit tests

Author: a-hansen

build.gradle

@@ -5,7 +5,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.47.0'
+    version '0.48.0'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8
@@ -47,5 +47,5 @@ if (JavaVersion.current().isJava8Compatible()) {
 }
 
 wrapper {
-    gradleVersion = '4.10.2'
+    gradleVersion = '5.0'
 }

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

@@ -157,25 +157,26 @@ public void run() {
                 DSIResponder responder = (DSIResponder) path.getTarget();
                 setPath(path.getPath());
                 result = responder.onInvoke(this);
-            }
-            DSInfo info = path.getTargetInfo();
-            if (!info.isAction()) {
-                throw new DSRequestException("Not an action " + path.getPath());
-            }
-            if (info.isAdmin()) {
-                if (!permission.isConfig()) {
-                    throw new DSPermissionException("Config permission required");
-                }
-            } else if (!info.isReadOnly()) {
-                if (DSPermission.WRITE.isGreaterThan(permission)) {
-                    throw new DSPermissionException("Write permission required");
-                }
             } else {
-                if (DSPermission.READ.isGreaterThan(permission)) {
-                    throw new DSPermissionException("Read permission required");
+                DSInfo info = path.getTargetInfo();
+                if (!info.isAction()) {
+                    throw new DSRequestException("Not an action " + path.getPath());
+                }
+                if (info.isAdmin()) {
+                    if (!permission.isConfig()) {
+                        throw new DSPermissionException("Config permission required");
+                    }
+                } else if (!info.isReadOnly()) {
+                    if (DSPermission.WRITE.isGreaterThan(permission)) {
+                        throw new DSPermissionException("Write permission required");
+                    }
+                } else {
+                    if (DSPermission.READ.isGreaterThan(permission)) {
+                        throw new DSPermissionException("Read permission required");
+                    }
                 }
+                result = path.getNode().invoke(info, path.getParentInfo(), this);
             }
-            path.getNode().invoke(info, path.getParentInfo(), this);
         } catch (Exception x) {
             error(getPath(), x);
             close(x);

dslink-v2/src/main/java/org/iot/dsa/dslink/requester/SimpleInvokeHandler.java

@@ -1,141 +1,231 @@
 package org.iot.dsa.dslink.requester;
 
-import org.iot.dsa.dslink.DSRequestException;
+import java.util.ArrayList;
 import org.iot.dsa.node.DSList;
 import org.iot.dsa.node.DSMap;
 
 /**
  * Action handler for non-tables/streams.
  * <p>
- * Call getResult(long timeout) to block until the invocation is complete. It will either return
+ * Call getUpdate(long timeout) to block until the invocation is complete. It will either return
  * the result (possibly null), or throw an exception.
  *
  * @author Aaron Hansen
  */
 public class SimpleInvokeHandler extends AbstractInvokeHandler {
 
     ///////////////////////////////////////////////////////////////////////////
-    // Fields
+    // Instance Fields
     ///////////////////////////////////////////////////////////////////////////
 
-    private boolean autoClose = true;
     private boolean closed = false;
+    private DSList columns;
     private RuntimeException error;
-    private DSList result;
+    private Mode mode;
+    private DSMap tableMeta;
+    private ArrayList<DSList> updates;
 
     ///////////////////////////////////////////////////////////////////////////
-    // Methods
+    // Public Methods
     ///////////////////////////////////////////////////////////////////////////
 
     /**
-     * Waits for the stream to close before returning, or the timeout to occur.
+     * It is possible for a values action to not return columns.
      *
-     * @param timeout Passed to Object.wait
-     * @return Null, or the first update.
+     * @return Beware of null.
+     */
+    public int getColumnCount() {
+        if (columns == null) {
+            return 0;
+        }
+        return columns.size();
+    }
+
+    /**
+     * The metadata map for the given column.
+     */
+    public DSMap getColumnMetadata(int idx) {
+        return columns.getMap(idx);
+    }
+
+    public RuntimeException getError() {
+        return error;
+    }
+
+    public Mode getMode() {
+        return mode;
+    }
+
+    public DSMap getTableMeta() {
+        return tableMeta;
+    }
+
+    /**
+     * The next available update, or null for actions return void.
+     * Will wait for an update if one isn't available.  Will return all updates before
+     * throwing any exceptions.
+     *
+     * @param timeout How long to wait for an update or the stream to close.
+     * @return Null if the action doesn't return anything.
      * @throws RuntimeException if there is a timeout, or if there are any errors.
      */
-    public DSList getResult(long timeout) {
+    public DSList getUpdate(long timeout) {
+        long end = System.currentTimeMillis() + timeout;
         synchronized (this) {
-            if (!closed) {
+            while (!closed && !hasError() && !hasUpdates()) {
                 try {
                     wait(timeout);
-                } catch (Exception x) {
+                } catch (Exception expected) {
+                }
+                if (System.currentTimeMillis() > end) {
+                    break;
                 }
             }
-        }
-        if (error != null) {
-            throw error;
-        }
-        if (!closed) {
+            if (hasUpdates()) {
+                return updates.remove(0);
+            }
+            if (hasError()) {
+                throw error;
+            }
+            if (closed) {
+                return null;
+            }
             throw new IllegalStateException("Action timed out");
         }
-        return result;
     }
 
     /**
-     * True by default, whether or not to close the stream upon receiving the first result.
+     * Takes the updates such that subsequent calls will never return the same updates, except when
+     * there are no updates in which case this returns null.
+     *
+     * @return Possibly null.
      */
-    public boolean isAutoClose() {
-        return autoClose;
+    public synchronized ArrayList<DSList> getUpdates() {
+        ArrayList<DSList> ret = updates;
+        updates = null;
+        return ret;
     }
 
-    /**
-     * Causes getResult to return.
-     */
-    public void onClose() {
-        synchronized (this) {
-            closed = true;
-            notifyAll();
-        }
+    public boolean hasError() {
+        return error != null;
     }
 
-    /**
-     * Will create an exception to be thrown by getResult.
-     */
-    public void onError(ErrorType type, String msg) {
-        synchronized (this) {
-            error = ErrorType.makeException(type, msg);
-            getStream().closeStream();
-            notifyAll();
+    public synchronized boolean hasUpdates() {
+        if (updates == null) {
+            return false;
         }
+        return !updates.isEmpty();
     }
 
-    /**
-     * Does nothing.
-     */
-    public void onColumns(DSList list) {
+    public boolean isClosed() {
+        return closed;
+    }
+
+    @Override
+    public synchronized void onClose() {
+        closed = true;
+        notifyAll();
+    }
+
+    @Override
+    public synchronized void onColumns(DSList list) {
+        this.columns = list;
+        notifyAll();
     }
 
     /**
-     * Will result in an error since tables and streams are not supported.
+     * Creates an exception that will be thrown by getUpdate.
      */
-    public void onInsert(int index, DSList rows) {
-        synchronized (this) {
-            error = new DSRequestException("Tables and streams not supported");
-            getStream().closeStream();
-            notifyAll();
-        }
+    @Override
+    public synchronized void onError(ErrorType type, String msg) {
+        error = ErrorType.makeException(type, msg);
+        notifyAll();
     }
 
     /**
-     * Does nothing.
+     * Does nothing other than notify and threads waiting on this instance.
      */
-    public void onMode(Mode mode) {
+    @Override
+    public synchronized void onInsert(int index, DSList rows) {
+        notifyAll();
+    }
+
+    @Override
+    public synchronized void onMode(Mode mode) {
+        this.mode = mode;
+        notifyAll();
     }
 
     /**
-     * Will result in an error since tables and streams are not supported.
+     * Does nothing other than notify and threads waiting on this instance.
      */
-    public void onReplace(int start, int end, DSList rows) {
-        synchronized (this) {
-            error = new DSRequestException("Tables and streams not supported");
-            getStream().closeStream();
-            notifyAll();
-        }
+    @Override
+    public synchronized void onReplace(int start, int end, DSList rows) {
+        notifyAll();
     }
 
-    public void onTableMeta(DSMap map) {
+    @Override
+    public synchronized void onTableMeta(DSMap map) {
+        this.tableMeta = map;
+        notifyAll();
     }
 
     /**
      * Captures the result and if auto-close is true, closes the stream.
      */
-    public void onUpdate(DSList row) {
+    @Override
+    public synchronized void onUpdate(DSList row) {
+        if (updates == null) {
+            updates = new ArrayList<>();
+        }
+        updates.add(row);
+        notifyAll();
+    }
+
+    /**
+     * Waits for any callback from the responder.  Will return immediately if already closed.
+     *
+     * @param timeout Passed to Object.wait
+     * @throws RuntimeException      if there is an error with the invocation.
+     * @throws IllegalStateException if there is a timeout, or if there are any errors.
+     */
+    public void waitForCallback(long timeout) {
         synchronized (this) {
-            result = row;
-            if (autoClose) {
-                getStream().closeStream();
+            if (!closed) {
+                long end = System.currentTimeMillis() + timeout;
+                try {
+                    wait(timeout);
+                } catch (Exception x) {
+                }
+                if (System.currentTimeMillis() > end) {
+                    throw new IllegalStateException("Action timed out");
+                }
             }
         }
     }
 
     /**
-     * Whether or not to auto close the stream on the first update.  True by default, this
-     * only needs to be called to disable.
+     * Waits for the stream to close or the timeout to occur.
+     *
+     * @param timeout Passed to Object.wait
+     * @throws IllegalStateException if there is a timeout, or if there are any errors.
      */
-    public SimpleInvokeHandler setAutoClose(boolean arg) {
-        autoClose = arg;
-        return this;
+    public void waitForClose(long timeout) {
+        long end = System.currentTimeMillis() + timeout;
+        synchronized (this) {
+            while (!closed) {
+                try {
+                    wait(timeout);
+                } catch (Exception x) {
+                }
+                if (System.currentTimeMillis() > end) {
+                    break;
+                }
+            }
+        }
+        if (!closed) {
+            throw new IllegalStateException("Action timed out");
+        }
     }
 
 }