0.75.2

* Fix sending first message on connect.
* Increase buffer sizes in small increments to decrease changes of OOM.
* Lock Tyrus dependencies at 1.16 for now.

Author: a-hansen

build.gradle

@@ -5,7 +5,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.75.1'
+    version '0.75.2'
 
     targetCompatibility = JavaVersion.VERSION_1_8
     sourceCompatibility = JavaVersion.VERSION_1_8

dslink-v2-websocket/build.gradle

@@ -1,6 +1,6 @@
 dependencies {
     api project(':dslink-v2')
-    implementation 'org.glassfish.tyrus.bundles:tyrus-standalone-client:[1.15,)'
+    implementation 'org.glassfish.tyrus.bundles:tyrus-standalone-client:1.16'
 }
 
 artifacts {

dslink-v2/build.gradle

@@ -1,7 +1,7 @@
 dependencies {
     api project(':dslink-v2-api')
-    compileOnly 'org.glassfish.tyrus.bundles:tyrus-standalone-client:[1.15,)'
-    testImplementation 'org.glassfish.tyrus.bundles:tyrus-standalone-client:[1.15,)'
+    compileOnly 'org.glassfish.tyrus.bundles:tyrus-standalone-client:1.16'
+    testImplementation 'org.glassfish.tyrus.bundles:tyrus-standalone-client:1.16'
     testImplementation 'org.testng:testng:+'
     testImplementation 'org.openjdk.jmh:jmh-core:+'
     testImplementation 'org.openjdk.jmh:jmh-generator-annprocess:+'

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

@@ -618,7 +618,11 @@ public byte[] toByteArray() {
     private void growBuffer(int minSize) {
         int size = buffer.length;
         while (size < minSize) {
-            size *= 2;
+            if (size < 1000000) {
+                size *= 2;
+            } else {
+                size += 1000000;
+            }
         }
         byte[] tmp = new byte[size];
         System.arraycopy(buffer, offset, tmp, 0, length);

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

@@ -153,7 +153,11 @@ public int read(char[] buf, int off, int len) {
     private void growBuffer(int minSize) {
         int size = buffer.length;
         while (size < minSize) {
-            size *= 2;
+            if (size < 1000000) {
+                size *= 2;
+            } else {
+                size += 1000000;
+            }
         }
         char[] tmp = new char[size];
         System.arraycopy(buffer, offset, tmp, 0, length);

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

@@ -95,7 +95,7 @@ public void close() {
     @Override
     public void close(Exception reason) {
         if (isOpen()) {
-            getResponder().sendError(this, reason);
+            getResponder().sendError(this, reason, false);
         }
         close();
     }

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

@@ -97,7 +97,11 @@ public DSStream removeRequest(Integer rid) {
 
     public abstract void sendClose(int rid);
 
-    public abstract void sendError(DSInboundRequest req, Throwable reason);
+    public void sendError(DSInboundRequest req, Throwable reason) {
+        sendError(req, reason, true);
+    }
+
+    public abstract void sendError(DSInboundRequest req, Throwable reason, boolean close);
 
     public void sendResponse(OutboundMessage res) {
         session.enqueueOutgoingResponse(res);

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

@@ -125,9 +125,11 @@ protected boolean hasPingToSend() {
     @Override
     protected void onConnected() {
         super.onConnected();
+        lastMessageSent = 0;
         setRequesterAllowed(true); //Rick says Dart broker doesn't send this
         requester.onConnected();
         responder.onConnected();
+        sendMessage();
     }
 
     @Override

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

@@ -128,8 +128,8 @@ public void sendClose(int rid) {
     }
 
     @Override
-    public void sendError(DSInboundRequest req, Throwable reason) {
-        sendResponse(new ErrorMessage(req.getRequestId(), reason));
+    public void sendError(DSInboundRequest req, Throwable reason, boolean close) {
+        sendResponse(new ErrorMessage(req.getRequestId(), reason).setClose(close));
     }
 
     public void sendError(int rid, Throwable reason) {

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

@@ -18,6 +18,8 @@
 class ErrorMessage implements OutboundMessage {
 
     private static String SERVER_ERROR = "serverError";
+
+    private boolean close = true;
     private String message;
     private Integer rid;
     private String type = SERVER_ERROR;
@@ -46,6 +48,14 @@ public boolean canWrite(DSSession session) {
         return true;
     }
 
+    /**
+     * Whether or not to close the stream.
+     */
+    public ErrorMessage setClose(boolean close) {
+        this.close = close;
+        return this;
+    }
+
     public ErrorMessage setType(String type) {
         this.type = type;
         return this;
@@ -54,9 +64,10 @@ public ErrorMessage setType(String type) {
     @Override
     public boolean write(DSSession session, MessageWriter writer) {
         DSIWriter out = writer.getWriter();
-        out.beginMap()
-           .key("rid").value(rid)
-           .key("stream").value("closed");
+        out.beginMap().key("rid").value(rid);
+        if (close) {
+           out.key("stream").value("closed");
+        }
         out.key("error").beginMap()
            .key("type").value(type)
            .key("msg").value(message)

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

@@ -83,7 +83,7 @@ public void sendClose(int rid) {
     }
 
     @Override
-    public void sendError(DSInboundRequest req, Throwable reason) {
+    public void sendError(DSInboundRequest req, Throwable reason, boolean close) {
         sendResponse(new ErrorMessage(req, reason));
     }