0.34.0

Session verifies if any outbound messages can actually be sent.

Author: a-hansen

build.gradle

@@ -4,7 +4,7 @@ subprojects {
     apply plugin: 'maven'
 
     group 'org.iot-dsa'
-    version '0.33.0'
+    version '0.34.0'
 
     sourceCompatibility = 1.6
     targetCompatibility = 1.6

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

@@ -4,8 +4,7 @@
 import com.acuity.iot.dsa.dslink.protocol.responder.DSResponder;
 import com.acuity.iot.dsa.dslink.transport.DSTransport;
 import java.io.IOException;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import org.iot.dsa.conn.DSConnection;
 import org.iot.dsa.conn.DSIConnected;
 import org.iot.dsa.dslink.DSIRequester;
@@ -49,8 +48,8 @@ public abstract class DSSession extends DSNode implements DSIConnected {
     private int messageId = 0;
     private int nextMessage = 1;
     private final Object outgoingMutex = new Object();
-    private List<OutboundMessage> outgoingRequests = new LinkedList<OutboundMessage>();
-    private List<OutboundMessage> outgoingResponses = new LinkedList<OutboundMessage>();
+    private ConcurrentLinkedQueue<OutboundMessage> outgoingRequests = new ConcurrentLinkedQueue<OutboundMessage>();
+    private ConcurrentLinkedQueue<OutboundMessage> outgoingResponses = new ConcurrentLinkedQueue<OutboundMessage>();
     private DSInfo requesterAllowed = getInfo(REQUESTER_ALLOWED);
     private ReadThread readThread;
     private WriteThread writeThread;
@@ -78,10 +77,8 @@ public void enqueueOutgoingRequest(OutboundMessage arg) {
             if (!isRequesterAllowed()) {
                 throw new IllegalStateException("Requester not allowed");
             }
-            synchronized (outgoingMutex) {
-                outgoingRequests.add(arg);
-                outgoingMutex.notify();
-            }
+            outgoingRequests.add(arg);
+            notifyOutgoing();
         }
     }
 
@@ -90,10 +87,8 @@ public void enqueueOutgoingRequest(OutboundMessage arg) {
      */
     public void enqueueOutgoingResponse(OutboundMessage arg) {
         if (connected) {
-            synchronized (outgoingMutex) {
-                outgoingResponses.add(arg);
-                outgoingMutex.notify();
-            }
+            outgoingResponses.add(arg);
+            notifyOutgoing();
         }
     }
 
@@ -175,24 +170,14 @@ protected void declareDefaults() {
      * Can return null.
      */
     protected OutboundMessage dequeueOutgoingRequest() {
-        synchronized (outgoingMutex) {
-            if (!outgoingRequests.isEmpty()) {
-                return outgoingRequests.remove(0);
-            }
-        }
-        return null;
+        return outgoingRequests.poll();
     }
 
     /**
      * Can return null.
      */
     protected OutboundMessage dequeueOutgoingResponse() {
-        synchronized (outgoingMutex) {
-            if (!outgoingResponses.isEmpty()) {
-                return outgoingResponses.remove(0);
-            }
-        }
-        return null;
+        return outgoingResponses.poll();
     }
 
     /**
@@ -247,10 +232,18 @@ protected boolean hasSomethingToSend() {
             return false;
         }
         if (!outgoingResponses.isEmpty()) {
-            return true;
+            for (OutboundMessage msg : outgoingResponses) {
+                if (msg.canWrite(this)) {
+                    return true;
+                }
+            }
         }
         if (!outgoingRequests.isEmpty()) {
-            return true;
+            for (OutboundMessage msg : outgoingRequests) {
+                if (msg.canWrite(this)) {
+                    return true;
+                }
+            }
         }
         return false;
     }
@@ -289,11 +282,9 @@ protected void onConnected() {
      * Clear the outgoing queues and waits for the the read and write threads to exit.
      */
     protected void onDisconnected() {
-        synchronized (outgoingMutex) {
-            outgoingRequests.clear();
-            outgoingResponses.clear();
-            outgoingMutex.notifyAll();
-        }
+        outgoingRequests.clear();
+        outgoingResponses.clear();
+        notifyOutgoing();
         try {
             readThread.join();
         } catch (Exception x) {
@@ -322,15 +313,11 @@ protected void onDisconnecting() {
     }
 
     protected void requeueOutgoingRequest(OutboundMessage arg) {
-        synchronized (outgoingMutex) {
-            outgoingRequests.add(arg);
-        }
+        outgoingRequests.add(arg);
     }
 
     protected void requeueOutgoingResponse(OutboundMessage arg) {
-        synchronized (outgoingMutex) {
-            outgoingResponses.add(arg);
-        }
+        outgoingResponses.add(arg);
     }
 
     /**
@@ -396,7 +383,8 @@ private void waitForAcks(long timeout) {
                     warn(getPath(), x);
                 }
                 if ((System.currentTimeMillis() - start) > timeout) {
-                    debug(debug() ? String.format("witForAcks timeout (%s / %s)",ackRcvd,messageId)
+                    debug(debug() ? String
+                            .format("waitForAcks timeout (%s / %s)", ackRcvd, messageId)
                                   : null);
                     break;
                 }