Message queue for websocket transport implemented

Author: Josema

dist/browser.js

@@ -30,11 +30,12 @@ var dop = {
     transports: {listen:{}, connect:{}},
 
     CONS: {
-        CLOSE: 0,
-        OPEN: 1,
-        CONNECT: 2,
-        RECONNECT: 3,
-        SEND: 4
+        CLOSE: '~CLOSE',
+        OPEN: '~OPEN',
+        CONNECTING: '~CONNECTING',
+        CONNECT: '~CONNECT',
+        RECONNECT: '~RECONNECT',
+        SEND: '~SEND'
     }
 
 };
@@ -182,15 +183,15 @@ function websocket(dop, node, options) {
         :
             send_queue.push(message); 
     }
-    function sendQueue(message) {
+    function sendQueue() {
         while (send_queue.length>0)
             socket.send(send_queue.shift());
     }
 
     // Socket events
     function onopen() {
         // Reconnect
-        if (node.readyState === dop.CONS.RECONNECT)
+        if (node.readyState === dop.CONS.CONNECTING)
             socket.send(node.tokenServer);
         // Connect
         else {
@@ -200,39 +201,46 @@ function websocket(dop, node, options) {
         dop.core.emitOpen(node, socket, options.transport);
     }
     function onmessage(message) {
+        // console.log( 'C<<: `'+message.data+'`' );
         // Reconnecting
-        if (node.readyState===dop.CONS.RECONNECT && message.data===node.tokenServer) {
+        if (node.readyState===dop.CONS.CONNECTING && message.data===node.tokenServer) {
             node.readyState = dop.CONS.CONNECT;
             dop.core.setSocketToNode(node, socket);
             dop.core.emitReconnect(node, oldSocket);
+            sendQueue();
         }
         else
             dop.core.emitMessage(node, message.data, message);
     }
     function onclose() {
+        node.readyState = dop.CONS.CLOSE;
         dop.core.emitClose(node, socket);
     }
 
     // dop events
-    function onconnect() {
-        if (node.readyState === dop.CONS.RECONNECT) {
+    function onconnect(message_response) {
+        if (node.readyState === dop.CONS.CONNECTING) {
             dop.core.emitDisconnect(node);
             dop.core.setSocketToNode(node, socket);
         }
+        socket.send(message_response);
         node.readyState = dop.CONS.CONNECT;
         dop.core.emitConnect(node);
+        sendQueue();
     }
     function ondisconnect() {
         node.readyState = dop.CONS.CLOSE;
         socket.close();
     }
 
     function reconnect() {
-        oldSocket = socket;
-        socket = new api(url);
-        node.readyState = dop.CONS.RECONNECT;
-        addListeners(socket, onopen, onmessage, onclose);
-        removeListeners(oldSocket, onopen, onmessage, onclose);
+        if (node.readyState !== dop.CONS.CONNECT) {
+            oldSocket = socket;
+            socket = new api(url);
+            node.readyState = dop.CONS.CONNECTING;
+            addListeners(socket, onopen, onmessage, onclose);
+            removeListeners(oldSocket, onopen, onmessage, onclose);
+        }
     }
 
     // Setting up
@@ -2605,8 +2613,7 @@ dop.protocol.onconnect = function(node, request_id, request) {
     var tokenServer=request[1],
         response = dop.core.createResponse(request_id, 0);
     node.tokenServer = tokenServer;
-    node.emit(dop.CONS.CONNECT);
-    node.send(JSON.stringify(response));
+    node.emit(dop.CONS.CONNECT, JSON.stringify(response));
 };
 
 

dist/browser.min.js

@@ -30,11 +30,12 @@ var dop = {
     transports: {listen:{}, connect:{}},
 
     CONS: {
-        CLOSE: 0,
-        OPEN: 1,
-        CONNECT: 2,
-        RECONNECT: 3,
-        SEND: 4
+        CLOSE: '~CLOSE',
+        OPEN: '~OPEN',
+        CONNECTING: '~CONNECTING',
+        CONNECT: '~CONNECT',
+        RECONNECT: '~RECONNECT',
+        SEND: '~SEND'
     }
 
 };
@@ -2426,8 +2427,7 @@ dop.protocol.onconnect = function(node, request_id, request) {
     var tokenServer=request[1],
         response = dop.core.createResponse(request_id, 0);
     node.tokenServer = tokenServer;
-    node.emit(dop.CONS.CONNECT);
-    node.send(JSON.stringify(response));
+    node.emit(dop.CONS.CONNECT, JSON.stringify(response));
 };
 
 

dist/nodejs.js

@@ -22,11 +22,12 @@ var dop = {
     transports: {listen:{}, connect:{}},
 
     CONS: {
-        CLOSE: 0,
-        OPEN: 1,
-        CONNECT: 2,
-        RECONNECT: 3,
-        SEND: 4
+        CLOSE: '~CLOSE',
+        OPEN: '~OPEN',
+        CONNECTING: '~CONNECTING',
+        CONNECT: '~CONNECT',
+        RECONNECT: '~RECONNECT',
+        SEND: '~SEND'
     }
 
 };

src/dop.js

@@ -41,6 +41,7 @@ function websocket(dop, node, options) {
         dop.core.emitOpen(node, socket, options.transport);
     }
     function onmessage(message) {
+        // console.log( 'C<<: `'+message.data+'`' );
         // Reconnecting
         if (node.readyState===dop.CONS.CONNECTING && message.data===node.tokenServer) {
             node.readyState = dop.CONS.CONNECT;
@@ -62,8 +63,7 @@ function websocket(dop, node, options) {
             dop.core.emitDisconnect(node);
             dop.core.setSocketToNode(node, socket);
         }
-        else
-            socket.send(message_response);
+        socket.send(message_response);
         node.readyState = dop.CONS.CONNECT;
         dop.core.emitConnect(node);
         sendQueue();

src/env/browser/websocket.js

@@ -29,7 +29,7 @@ function send() {
     if (nodeServer)
         nodeServer.send(String(msg));
     if (msg++ < 25)
-        setTimeout(send, 100);
+        setTimeout(send, 200);
 }
 send();
 
@@ -43,7 +43,7 @@ test('CONNECT TEST', function(t) {
         nodeServer = node;
     });
     server.on('message', function(node, message){
-        // console.log( '❌ message `'+message+'`', node.token );
+        console.log( '❌ message `'+message+'`', node.token );
     });
 
     nodeClient.on('message', function(message){
@@ -63,4 +63,5 @@ setTimeout(function(){
         console.log( 'reconnecting...' );
         nodeClient.reconnect();
     },500);
-}, 1000)
+}, 1000);
+

test/protocol/mutations_objects.js test/mutations_objects.jstest/protocol/mutations_objects.js renamed to test/mutations_objects.js

@@ -36,21 +36,19 @@ test('RECONNECT TEST', function(t) {
         else
             t.notEqual(node.socket, socketServer, '❌ connect 2');
     });
-    // server.on('close', function(socket){
-    //     t.equal(socket, socketServer, '❌ close');
-    // });
-    server.on('reconnect', function(node, oldSocket){
-        t.equal(node===nodeServer && oldSocket===socketServer, true, '❌ reconnect');
-    });
     server.on('disconnect', function(node){
         t.equal(node, nodeServer, '❌ disconnect');
     });
+    server.on('reconnect', function(node, oldSocket){
+        t.equal(false, true, '❌ reconnect'); // this should not happen
+    });
+
+
 
 
     nodeClient.on('open', function(socket) {
         if (socketClient === undefined)
             socketClient = socket;
-        // t.equal(dopClient.getNodeBySocket(socket).socket, socket, '✅ open');
     });
     nodeClient.on('connect', function() {
         if (tokenClient === undefined) {
@@ -63,15 +61,13 @@ test('RECONNECT TEST', function(t) {
             // server.listener.close();
         }
     });
-    // nodeClient.on('close', function(socket){
-    //     t.equal(socket, socketClient, '✅ close');
-    // });
-    nodeClient.on('reconnect', function(oldSocket) {
-        t.equal(oldSocket, socketClient, '✅ reconnect');
-    });
     nodeClient.on('disconnect', function() {
         t.equal(nodeClient.socket, socketClient, '✅ disconnect');
     });
+    nodeClient.on('reconnect', function(oldSocket) {
+        t.equal(true, false, '✅ reconnect'); // this should not happen
+    });
+
 });
 
 
@@ -83,6 +79,6 @@ setTimeout(function(){
 }, 500)
 // Reconnecting
 setTimeout(function(){
-    console.log( 'reconnecting...' );
+    console.log( 'late reconnecting...' );
     nodeClient.reconnect();
 }, 3000);