support lock/unlock.

fibjs · May 17, 2018 · bedbca6 · bedbca6
1 parent 8237bd9
commit bedbca6
Show file tree

Hide file tree

Showing 2 changed files with 82 additions and 8 deletions.
diff --git a/lib/index.js b/lib/index.js
@@ -15,11 +15,12 @@ function check_websocket_like_obj(ws) {
 }
 
 function channel(name) {
+    var self = this;
     var conns = new Link();
     var msgs = new Link();
     var idle_node;
     var start_timestamp = new Date();
-    var self = this;
+    var locked = 0;
 
     function toIdle() {
         idle_node = idles.addTail(self);
@@ -30,16 +31,20 @@ function channel(name) {
         }
     }
 
+    function active() {
+        if (idle_node) {
+            idles.remove(idle_node);
+            idle_node = undefined;
+        }
+    }
+
     toIdle();
 
     self.name = name;
     self.on = (ws, timestamp, filter) => {
         check_websocket_like_obj(ws);
 
-        if (idle_node) {
-            idles.remove(idle_node);
-            idle_node = undefined;
-        }
+        active();
 
         timestamp = new Date(timestamp);
 
@@ -73,7 +78,7 @@ function channel(name) {
     };
 
     self.off = node => {
-        if (conns.remove(node) === 0)
+        if (conns.remove(node) === 0 && locked === 0)
             toIdle();
     };
 
@@ -99,13 +104,24 @@ function channel(name) {
 
         var node = conns.head();
         while (node !== undefined) {
-            if (!node.filter || node.filter(data)) {
+            if (!node.filter || node.filter(data))
                 node.data.send(json);
-            }
             node = node.next;
         }
     };
 
+    self.lock = function () {
+        locked++;
+        if (locked == 1)
+            active();
+    }
+
+    self.unlock = function () {
+        locked--;
+        if (locked == 0 && conns.count() == 0)
+            toIdle();
+    }
+
     self.status = () => {
         return conns.toJSON();
     }
@@ -153,6 +169,22 @@ exports.post = (ch, data) => {
     cho.post(data);
 };
 
+exports.lock = (ch) => {
+    var cho = chs[ch];
+    if (cho === undefined)
+        chs[ch] = cho = new channel(ch);
+
+    cho.lock();
+};
+
+exports.unlock = (ch) => {
+    var cho = chs[ch];
+    if (cho === undefined)
+        chs[ch] = cho = new channel(ch);
+
+    cho.unlock();
+};
+
 exports.status = () => {
     var r = {};
     for (ch in chs)

diff --git a/test/index.js b/test/index.js
@@ -194,6 +194,48 @@ describe("push", () => {
         ]);
     });
 
+    it("lock/unlock", () => {
+        var ws = {
+            send: m => {}
+        };
+
+        push.post(`lock_test`, 1);
+        assert.property(push.status(), "lock_test");
+
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.notProperty(push.status(), "lock_test");
+
+        push.lock("lock_test");
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.property(push.status(), "lock_test");
+
+        push.unlock("lock_test");
+        assert.property(push.status(), "lock_test");
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.notProperty(push.status(), "lock_test");
+
+        push.lock("lock_test");
+        assert.property(push.status(), "lock_test");
+
+        push.on(`lock_test`, ws);
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.property(push.status(), "lock_test");
+
+        push.unlock("lock_test");
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.property(push.status(), "lock_test");
+
+        ws.onclose();
+        for (var i = 0; i < 100; i++)
+            push.post(`lock_${i}`, 1);
+        assert.notProperty(push.status(), "lock_test");
+    });
+
     it("double on", () => {
         var ws = {
             send: m => {}