Durable sessions

Author: Aaron

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

@@ -6,22 +6,31 @@
 import java.io.IOException;
 import java.util.LinkedList;
 import java.util.List;
+import org.iot.dsa.conn.DSConnection;
+import org.iot.dsa.conn.DSIConnected;
 import org.iot.dsa.dslink.DSIRequester;
 import org.iot.dsa.dslink.DSLinkConnection;
+import org.iot.dsa.node.DSBool;
+import org.iot.dsa.node.DSInfo;
 import org.iot.dsa.node.DSNode;
+import org.iot.dsa.util.DSException;
 
 /**
  * The state of a connection to a broker as well as a protocol implementation. Not intended for link
  * implementors.
  *
  * @author Aaron Hansen
  */
-public abstract class DSSession extends DSNode {
+public abstract class DSSession extends DSNode implements DSIConnected {
 
     ///////////////////////////////////////////////////////////////////////////
     // Class Fields
     ///////////////////////////////////////////////////////////////////////////
 
+    protected static final String REQUESTER = "Requester";
+    protected static final String REQUESTER_ALLOWED = "Requester Allowed";
+    protected static final String RESPONDER = "Responder";
+
     private static final int MAX_MSG_ID = Integer.MAX_VALUE;
     private static final long MSG_TIMEOUT = 60000;
 
@@ -37,10 +46,12 @@ public abstract class DSSession extends DSNode {
     private long lastTimeSend;
     private int messageId = 0;
     private int nextMessage = 1;
-    private Object outgoingMutex = new Object();
+    private final Object outgoingMutex = new Object();
     private List<OutboundMessage> outgoingRequests = new LinkedList<OutboundMessage>();
     private List<OutboundMessage> outgoingResponses = new LinkedList<OutboundMessage>();
-    protected boolean requesterAllowed = false;
+    private DSInfo requesterAllowed = getInfo(REQUESTER_ALLOWED);
+    private ReadThread readThread;
+    private WriteThread writeThread;
 
     ///////////////////////////////////////////////////////////////////////////
     // Constructors
@@ -57,27 +68,12 @@ public DSSession(DSLinkConnection connection) {
     // Public Methods
     ///////////////////////////////////////////////////////////////////////////
 
-    /**
-     * Can be called by the subclass to force exit the run method.
-     */
-    public void disconnect() {
-        if (!connected) {
-            return;
-        }
-        connected = false;
-        getTransport().close();
-        synchronized (outgoingMutex) {
-            notifyAll();
-        }
-        info(getPath() + " locally closed");
-    }
-
     /**
      * Add a message to the outgoing request queue.
      */
     public void enqueueOutgoingRequest(OutboundMessage arg) {
         if (connected) {
-            if (!requesterAllowed) {
+            if (!isRequesterAllowed()) {
                 throw new IllegalStateException("Requests forbidden");
             }
             synchronized (outgoingMutex) {
@@ -134,58 +130,31 @@ public DSTransport getTransport() {
         return getConnection().getTransport();
     }
 
-    /**
-     * Override point, called when the previous connection can be resumed. The the transport will
-     * have already been set.
-     */
-    public void onConnect() {
-        connected = true;
-    }
-
-    /**
-     * Override point, when a connection attempt failed.
-     */
-    public void onConnectFail() {
-        connected = false;
+    public boolean isRequesterAllowed() {
+        return requesterAllowed.getElement().toBoolean();
     }
 
-    /**
-     * Override point, called after the connection is closed.
-     */
-    public void onDisconnect() {
-        synchronized (outgoingMutex) {
-            outgoingRequests.clear();
-            outgoingResponses.clear();
-        }
-    }
-
-    /**
-     * Called by the connection, this manages the running state and calls doRun for the specific
-     * implementation.  A separate thread is spun off to manage writing.
-     */
-    public void run() {
-        lastTimeRecv = lastTimeSend = System.currentTimeMillis();
-        new WriteThread(getConnection().getLink().getLinkName() + " Writer").start();
-        while (connected) {
-            try {
-                verifyLastSend();
-                doRecvMessage();
-                lastTimeRecv = System.currentTimeMillis();
-            } catch (Exception x) {
-                getTransport().close();
-                if (connected) {
-                    connected = false;
-                    error(getPath(), x);
-                }
-            }
+    @Override
+    public void onChange(DSConnection connection) {
+        switch (connection.getConnectionState()) {
+            case CONNECTED:
+                onConnected();
+                break;
+            case DISCONNECTED:
+                onDisconnected();
+                break;
+            case DISCONNECTING:
+                onDisconnecting();
+                break;
+            //case CONNECTING:
         }
     }
 
     /**
      * Called when the broker signifies that requests are allowed.
      */
-    public void setRequesterAllowed() {
-        requesterAllowed = true;
+    public void setRequesterAllowed(boolean allowed) {
+        put(requesterAllowed, DSBool.valueOf(allowed));
     }
 
     public abstract boolean shouldEndMessage();
@@ -194,6 +163,12 @@ public void setRequesterAllowed() {
     // Protected Methods
     ///////////////////////////////////////////////////////////////////////////
 
+    @Override
+    protected void declareDefaults() {
+        super.declareDefaults();
+        declareDefault(REQUESTER_ALLOWED, DSBool.FALSE);
+    }
+
     /**
      * Can return null.
      */
@@ -232,7 +207,7 @@ protected OutboundMessage dequeueOutgoingResponse() {
 
     @Override
     protected String getLogName() {
-        return "Session";
+        return getLogName("session");
     }
 
     protected int getMissingAcks() {
@@ -254,13 +229,7 @@ protected boolean hasAckToSend() {
         return ackToSend > 0;
     }
 
-    protected boolean hasOutgoingRequests() {
-        return !outgoingRequests.isEmpty();
-    }
-
-    protected boolean hasOutgoingResponses() {
-        return !outgoingResponses.isEmpty();
-    }
+    protected abstract boolean hasPingToSend();
 
     /**
      * Override point, this returns the result of hasMessagesToSend.
@@ -269,6 +238,9 @@ protected boolean hasSomethingToSend() {
         if (ackToSend >= 0) {
             return true;
         }
+        if (hasPingToSend()) {
+            return true;
+        }
         if (waitingForAcks()) {
             return false;
         }
@@ -281,10 +253,6 @@ protected boolean hasSomethingToSend() {
         return false;
     }
 
-    protected boolean isConnected() {
-        return connected;
-    }
-
     /**
      * Can be used to waking up a sleeping writer.
      */
@@ -302,6 +270,51 @@ protected int numOutgoingResponses() {
         return outgoingResponses.size();
     }
 
+    /**
+     * Creates the starts the read and write threads.
+     */
+    protected void onConnected() {
+        connected = true;
+        lastTimeRecv = lastTimeSend = System.currentTimeMillis();
+        readThread = new ReadThread(getConnection().getLink().getLinkName() + " Reader");
+        writeThread = new WriteThread(getConnection().getLink().getLinkName() + " Writer");
+        readThread.start();
+        writeThread.start();
+    }
+
+    /**
+     * Clear the outgoing queues and waits for the the read and write threads to exit.
+     */
+    protected void onDisconnected() {
+        synchronized (outgoingMutex) {
+            outgoingRequests.clear();
+            outgoingResponses.clear();
+        }
+        try {
+            writeThread.join(); //TODO - timeout?
+        } catch (Exception x) {
+            debug(getPath(), x);
+        }
+        try {
+            readThread.join(); //TODO - timeout?
+        } catch (Exception x) {
+            debug(getPath(), x);
+        }
+        writeThread = null;
+        readThread = null;
+    }
+
+    /**
+     * Sets the connected state to false so that the read and write threads will exit cleanly.
+     */
+    protected void onDisconnecting() {
+        if (!connected) {
+            return;
+        }
+        connected = false;
+        notifyOutgoing();
+    }
+
     protected void requeueOutgoingRequest(OutboundMessage arg) {
         synchronized (outgoingMutex) {
             outgoingRequests.add(arg);
@@ -364,7 +377,36 @@ private void verifyLastSend() throws IOException {
     ///////////////////////////////////////////////////////////////////////////
 
     /**
-     * A separate thread is used for writing to the connection.
+     * Receives messages.
+     */
+    private class ReadThread extends Thread {
+
+        ReadThread(String name) {
+            super(name);
+            setDaemon(true);
+        }
+
+        public void run() {
+            DSLinkConnection conn = getConnection();
+            try {
+                while (connected) {
+                    verifyLastSend();
+                    doRecvMessage();
+                    conn.connOk();
+                    lastTimeRecv = System.currentTimeMillis();
+                }
+            } catch (Exception x) {
+                if (connected) {
+                    connected = false;
+                    error(getPath(), x);
+                }
+                conn.connDown(DSException.makeMessage(x));
+            }
+        }
+    }
+
+    /**
+     * Sends messages.
      */
     private class WriteThread extends Thread {
 
@@ -374,6 +416,7 @@ private class WriteThread extends Thread {
         }
 
         public void run() {
+            DSLinkConnection conn = getConnection();
             try {
                 while (connected) {
                     verifyLastRead();
@@ -382,20 +425,21 @@ public void run() {
                             try {
                                 outgoingMutex.wait(5000);
                             } catch (InterruptedException x) {
-                                warn(getPath(), x);
+                                debug(getPath(), x);
                             }
                             continue;
                         }
                     }
                     doSendMessage();
+                    conn.connOk();
                     lastTimeSend = System.currentTimeMillis();
                 }
             } catch (Exception x) {
                 if (connected) {
                     connected = false;
-                    getTransport().close();
                     error(getPath(), x);
                 }
+                conn.connDown(DSException.makeMessage(x));
             }
         }
     }

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

@@ -92,25 +92,6 @@ public void handleUpdate(int sid, String ts, String sts, DSElement value) {
         stub.process(timestamp, value, status);
     }
 
-    public void onConnect() {
-    }
-
-    public void onConnectFail() {
-    }
-
-    public void onDisconnect() {
-        for (DSOutboundSubscribeStubs stubs : pendingSubscribe) {
-            stubs.onDisconnect();
-        }
-        pendingSubscribe.clear();
-        pendingUnsubscribe.clear();
-        for (DSOutboundSubscribeStubs stubs : pathMap.values()) {
-            stubs.onDisconnect();
-        }
-        sidMap.clear();
-        pathMap.clear();
-    }
-
     @Override
     public void write(DSSession session, MessageWriter writer) {
         if (!pendingSubscribe.isEmpty()) {
@@ -213,6 +194,22 @@ protected void doWriteUnsubscribe(MessageWriter writer, Integer sid) {
         out.value(sid);
     }
 
+    protected void onConnected() {
+    }
+
+    protected void onDisconnected() {
+        for (DSOutboundSubscribeStubs stubs : pendingSubscribe) {
+            stubs.onDisconnect();
+        }
+        pendingSubscribe.clear();
+        pendingUnsubscribe.clear();
+        for (DSOutboundSubscribeStubs stubs : pathMap.values()) {
+            stubs.onDisconnect();
+        }
+        sidMap.clear();
+        pathMap.clear();
+    }
+
     ///////////////////////////////////////////////////////////////////////////
     // Package / Private Methods
     ///////////////////////////////////////////////////////////////////////////