mutual - npm Package Compare versions

Comparing version 0.4.15 to 0.4.16

lib/durable-channel.js

@@ -74,15 +74,14 @@ // Generated by CoffeeScript 1.7.1
         return this.store;
-      } else {
-        return this.events.serially((function(_this) {
-          return function(go) {
-            go(function() {
-              return _this.adapter.collection("" + _this.name + ".messages");
-            });
-            return go(function(store) {
-              _this.store = store;
-              return _this.store;
-            });
-          };
-        })(this))();
       }
+      return this.events.serially((function(_this) {
+        return function(go) {
+          go(function() {
+            return _this.adapter.collection("" + _this.name + ".messages");
+          });
+          return go(function(store) {
+            _this.store = store;
+            return _this.store;
+          });
+        };
+      })(this))();
     };
@@ -93,15 +92,13 @@
         return this.destinationStores[name];
-      } else {
-        return this.events.serially((function(_this) {
-          return function(go) {
-            go(function() {
-              return _this.adapter.collection("" + name + ".messages");
-            });
-            return go(function(store) {
-              _this.destinationStores[name] = store;
-              return store;
-            });
-          };
-        })(this))();
       }
+      return this.events.serially((function(_this) {
+        return function(go) {
+          go(function() {
+            return _this.adapter.collection("" + name + ".messages");
+          });
+          return go(function(store) {
+            return _this.destinationStores[name] = store;
+          });
+        };
+      })(this))();
     };
@@ -111,9 +108,6 @@
       var _base;
-      if ((_base = this.destinationQueues)[name] == null) {
-        _base[name] = new RemoteQueue({
-          name: "" + name + ".queue",
-          transport: this.transport
-        });
-      }
-      return this.destinationQueues[name];
+      return (_base = this.destinationQueues)[name] != null ? _base[name] : _base[name] = new RemoteQueue({
+        name: "" + name + ".queue",
+        transport: this.transport
+      });
     };
@@ -141,2 +135,12 @@
 
+    DurableChannel.prototype.getMessageTimeout = function(name, channel, id) {
+      if (id != null) {
+        return this.events.source((function(_this) {
+          return function(events) {
+            return _this.adapter.client.zscore(["" + name + ".pending", "" + channel + "::" + id], events.callback);
+          };
+        })(this));
+      }
+    };
+
     DurableChannel.prototype.monitorTimeouts = function() {
@@ -200,20 +204,32 @@ var loopToMonitor;
           go(function(_message) {
-            message = _message;
-            if (message != null) {
-              return store["delete"](id);
-            }
+            return message = _message;
           });
           go(function() {
-            return _this.clearMessageTimeout(_this.name, channel, id);
+            return _this.getMessageTimeout(_this.name, channel, id);
           });
-          return go(function() {
-            if (message != null) {
-              return _this.fire({
-                event: "timeout",
-                content: {
-                  content: message.content,
-                  requestId: message.requestId
+          return go(function(timeout) {
+            if (timeout == null) {
+              return;
+            }
+            return _this.events.serially(function(go) {
+              go(function() {
+                if (message != null) {
+                  return store["delete"](id);
                 }
               });
-            }
+              go(function() {
+                return _this.clearMessageTimeout(_this.name, channel, id);
+              });
+              return go(function() {
+                if (message != null) {
+                  return _this.fire({
+                    event: "timeout",
+                    content: {
+                      content: message.content,
+                      requestId: message.requestId
+                    }
+                  });
+                }
+              });
+            })();
           });
@@ -237,4 +253,4 @@ };
           });
-          go(function(store) {
-            return store.put(message.id, message);
+          go(function(destStore) {
+            return destStore.put(message.id, message);
           });
@@ -264,16 +280,19 @@ go(function() {
             if (request == null) {
-              return;
+              return null;
             }
+            message = _this["package"]({
+              content: response,
+              to: request.from,
+              requestId: message.requestId,
+              timeout: timeout
+            });
             return _this.events.serially(function(go) {
               go(function() {
-                message = _this["package"]({
-                  content: response,
-                  to: request.from,
-                  requestId: message.requestId,
-                  timeout: timeout
-                });
+                return _this.clearMessageTimeout(request.from, _this.name, message.requestId);
+              });
+              go(function() {
                 return _this.getDestinationStore(request.from);
               });
-              go(function(store) {
-                return store.put(message.id, message);
+              go(function(destStore) {
+                return destStore.put(message.id, message);
               });
@@ -293,4 +312,2 @@ go(function() {
     DurableChannel.prototype.close = function(message) {
-      var request;
-      request = null;
       return this.events.serially((function(_this) {
@@ -317,2 +334,4 @@ return function(go) {
             messageHandler = function(messageId) {
+              var store;
+              store = null;
               return _this.events.serially(function(go) {
@@ -322,42 +341,58 @@ go(function() {
                 });
-                go(function(store) {
+                go(function(_store) {
+                  store = _store;
                   return store.get(messageId);
                 });
                 go(function(message) {
+                  var destStore;
                   if (message == null) {
-                    return;
+                    return null;
                   }
-                  return _this.events.serially(function(go) {
-                    go(function() {
-                      if (message.requestId != null) {
-                        return _this.clearMessageTimeout(_this.name, message.from, message.requestId);
-                      }
-                    });
-                    go(function() {
-                      if (message.requestId != null) {
+                  if (message.requestId == null) {
+                    return message;
+                  }
+                  destStore = null;
+                  return _this.events.source(function(events) {
+                    return _this.events.serially(function(go) {
+                      go(function() {
                         return _this.getDestinationStore(message.from);
-                      }
-                    });
-                    go(function(store) {
-                      if (message.requestId != null) {
-                        return store["delete"](message.requestId);
-                      }
-                    });
-                    return go(function() {
-                      var _message;
-                      _message = {
-                        content: message.content
-                      };
-                      _message.from = message.requestId != null ? message.to : message.from;
-                      _message.to = message.requestId != null ? message.from : message.to;
-                      _message.requestId = message.requestId != null ? message.requestId : message.id;
-                      if (message.requestId != null) {
-                        _message.responseId = message.id;
-                      }
-                      return _this.fire({
-                        event: "message",
-                        content: _message
                       });
+                      go(function(_destStore) {
+                        destStore = _destStore;
+                        return destStore.get(message.requestId);
+                      });
+                      return go(function(request) {
+                        return _this.events.serially(function(go) {
+                          go(function() {
+                            if (request != null) {
+                              return destStore["delete"](message.requestId);
+                            } else {
+                              return store["delete"](messageId);
+                            }
+                          });
+                          return go(function() {
+                            return events.emit("success", (request != null ? message : null));
+                          });
+                        })();
+                      });
+                    })();
+                  });
+                });
+                go(function(message) {
+                  var _message;
+                  if (message != null) {
+                    _message = {
+                      content: message.content
+                    };
+                    _message.from = message.requestId != null ? message.to : message.from;
+                    _message.to = message.requestId != null ? message.from : message.to;
+                    _message.requestId = message.requestId != null ? message.requestId : message.id;
+                    if (message.requestId != null) {
+                      _message.responseId = message.id;
+                    }
+                    return _this.fire({
+                      event: "message",
+                      content: _message
                     });
-                  })();
+                  }
                 });
@@ -364,0 +399,0 @@ return go(function() {

package.json

 {
   "name": "mutual",
-  "version": "0.4.15",
+  "version": "0.4.16",
   "description": "Scala-inspired Actors that use Redis as a message transport",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

49960

increased by3.03%

Lines of code

1221

increased by2.86%

No dependency changes