Hopiu/postal.js
1
0
Fork 0
mirror of https://github.com/Hopiu/postal.js.git synced 2026-04-21 07:11:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Refactored the API yet again - splitting out the ChannelDefinition concerns from the SubscriptionDefinition concerns

Browse source
This commit is contained in:
Jim Cowart 2011-09-12 17:49:21 -04:00
parent d742595f2e
commit f33a06dd77
12 changed files with 402 additions and 554 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

228
build/output/browser/postal.js
View file

@ -7,6 +7,10 @@
(function(global, undefined) {
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };
var DistinctPredicate = function() {
var previous;
return function(data) {
@ -22,48 +26,56 @@ var DistinctPredicate = function() {
return !eq;
};
};
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };
var defaultConfiguration = {
exchange: DEFAULT_EXCHANGE,
topic: "",
callback: NO_OP,
priority: DEFAULT_PRIORITY,
constraints: [],
disposeAfter: DEFAULT_DISPOSEAFTER,
onHandled: NO_OP,
context: null,
modifiers: []
var ChannelDefinition = function(exchange, topic) {
this.exchange = exchange;
this.topic = topic;
};
var ChannelDefinition = function(exchange, topic) {
this.configuration = _.extend(defaultConfiguration, { exchange: exchange, topic: topic });
} ;
ChannelDefinition.prototype = {
exchange: function(exchange) {
this.configuration.exchange = exchange;
return this;
subscribe: function(callback) {
var subscription = new SubscriptionDefinition(this.exchange, this.topic, callback);
postal.configuration.bus.subscribe(subscription);
return subscription;
},
topic: function(topic) {
this.configuration.topic = topic;
return this;
publish: function(data) {
postal.configuration.bus.publish({
exchange: this.exchange,
topic: this.topic,
data: data,
timeStamp: new Date()
})
}
};
var SubscriptionDefinition = function(exchange, topic, callback) {
this.exchange = exchange;
this.topic = topic;
this.callback = callback;
this.priority = DEFAULT_PRIORITY;
this.constraints = [];
this.maxCalls = DEFAULT_DISPOSEAFTER;
this.onHandled = NO_OP;
this.context = null
};
SubscriptionDefinition.prototype = {
unsubscribe: function() {
postal.configuration.bus.unsubscribe(this);
},
defer: function() {
this.configuration.modifiers.push({type: "defer"});
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn,0,data);
};
return this;
},
disposeAfter: function(receiveCount) {
if(_.isNaN(receiveCount)) {
throw "The value provided to disposeAfter (receiveCount) must be a number";
disposeAfter: function(maxCalls) {
if(_.isNaN(maxCalls)) {
throw "The value provided to disposeAfter (maxCalls) must be a number";
}
this.configuration.disposeAfter = receiveCount;
this.maxCalls = maxCalls;
return this;
},
@ -76,7 +88,7 @@ ChannelDefinition.prototype = {
if(! _.isFunction(callback)) {
throw "Value provided to 'whenHandledThenExecute' must be a function";
}
this.configuration.onHandled = callback;
this.onHandled = callback;
return this;
},
@ -84,7 +96,7 @@ ChannelDefinition.prototype = {
if(! _.isFunction(predicate)) {
throw "Predicate constraint must be a function";
}
this.configuration.constraints.push(predicate);
this.constraints.push(predicate);
return this;
},
@ -97,7 +109,7 @@ ChannelDefinition.prototype = {
},
withContext: function(context) {
this.configuration.context = context;
this.context = context;
return this;
},
@ -105,7 +117,8 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "debounce", milliseconds: milliseconds});
var fn = this.callback;
this.callback = _.debounce(fn, milliseconds);
return this;
},
@ -113,7 +126,10 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "delay", milliseconds: milliseconds});
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn, milliseconds, data);
};
return this;
},
@ -121,7 +137,7 @@ ChannelDefinition.prototype = {
if(_.isNaN(priority)) {
throw "Priority must be a number";
}
this.configuration.priority = priority;
this.priority = priority;
return this;
},
@ -129,26 +145,14 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "throttle", milliseconds: milliseconds});
var fn = this.callback;
this.callback = _.throttle(fn, milliseconds);
return this;
},
subscribe: function(callback) {
this.configuration.callback = callback || NO_OP;
return postal.subscribe(this.configuration);
},
publish: function(data) {
postal.publish({
exchange: this.configuration.exchange,
data: data,
topic: this.configuration.topic
});
}
};
var bindingsResolver = {
cache: { },
compare: function(binding, topic) {
if(this.cache[topic] && this.cache[topic][binding]) {
return true;
@ -163,38 +167,20 @@ var bindingsResolver = {
}
return result;
},
regexify: function(binding) {
return binding.replace(/\./g,"\\.") // escape actual periods
.replace(/\*/g, ".*") // asterisks match any value
.replace(/#/g, "[A-Z,a-z,0-9]*"); // hash matches any alpha-numeric 'word'
}
};
var wrapWithDelay = function(callback, config) {
return function(data) {
setTimeout(callback, config.milliseconds, data);
};
},
wrapWithDefer = function(callback) {
return function(data) {
setTimeout(callback,0,data);
}
},
wrapWithThrottle = function(callback, config) {
return _.throttle(callback, config.milliseconds);
},
wrapWithDebounce = function(callback, config) {
return _.debounce(callback, config.milliseconds);
};
var localBus = {
subscriptions: {},
wireTaps: [],
publish: function(config) {
var envelope = _.extend(defaultConfiguration, { timeStamp: new Date() }, config);
publish: function(envelope) {
_.each(this.wireTaps,function(tap) {
tap({
exchange: envelope.exchange,
@ -218,63 +204,35 @@ var localBus = {
});
},
subscribe: function(channelDef) {
var idx, found, fn, config = _.extend(defaultConfiguration, channelDef);
if(config.disposeAfter && config.disposeAfter > 0) {
fn = config.onHandled,
dispose = _.after(config.disposeAfter, _.bind(function() {
this.unsubscribe(config);
subscribe: function(subDef) {
var idx, found, fn;
if(subDef.maxCalls) {
fn = subDef.onHandled;
var dispose = _.after(subDef.maxCalls, _.bind(function() {
this.unsubscribe(subDef);
}, this));
config.onHandled = function() {
fn.apply(config.context, arguments);
subDef.onHandled = function() {
fn.apply(subDef.context, arguments);
dispose();
}
}
_.each(config.modifiers, function(modifier) {
fn = config.callback;
switch(modifier.type) {
case 'delay':
config.callback = wrapWithDelay(fn, modifier);
break;
case 'defer':
config.callback = wrapWithDefer(fn);
break;
case 'throttle':
config.callback = wrapWithThrottle(fn,modifier);
break;
case 'debounce':
config.callback = wrapWithDebounce(fn, modifier);
break;
idx = this.subscriptions[subDef.exchange][subDef.topic].length - 1;
if(!_.any(this.subscriptions[subDef.exchange][subDef.topic], function(cfg) { return cfg === subDef; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[subDef.exchange][subDef.topic][idx].priority <= subDef.priority) {
this.subscriptions[subDef.exchange][subDef.topic].splice(idx + 1, 0, subDef);
found = true;
break;
}
}
});
if(!this.subscriptions[config.exchange]) {
this.subscriptions[config.exchange] = {};
}
if(!this.subscriptions[config.exchange][config.topic]) {
this.subscriptions[config.exchange][config.topic] = [config];
}
else {
idx = this.subscriptions[config.exchange][config.topic].length - 1;
if(!_.any(this.subscriptions[config.exchange][config.topic], function(cfg) { return cfg === config; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[config.exchange][config.topic][idx].priority <= config.priority) {
this.subscriptions[config.exchange][config.topic].splice(idx + 1, 0, config);
found = true;
break;
}
}
if(!found) {
this.subscriptions[config.exchange][config.topic].unshift(config);
}
console.log("SUBSCRIBE: " + JSON.stringify(config));
if(!found) {
this.subscriptions[subDef.exchange][subDef.topic].unshift(subDef);
}
}
return _.bind(function() { this.unsubscribe(config); }, this);
return _.bind(function() { this.unsubscribe(subDef); }, this);
},
unsubscribe: function(config) {
@ -284,7 +242,6 @@ var localBus = {
for ( ; idx < len; idx++ ) {
if (this.subscriptions[config.exchange][config.topic][idx] === config) {
this.subscriptions[config.exchange][config.topic].splice( idx, 1 );
console.log("UNSUBSCRIBE: " + JSON.stringify(config));
break;
}
}
@ -302,30 +259,21 @@ var localBus = {
}
};
var postal = {
configuration: {
bus: localBus,
resolver: bindingsResolver
},
exchange: function(exchange) {
return new ChannelDefinition(exchange);
},
topic: function(topic) {
return new ChannelDefinition(undefined, topic);
},
publish: function(config) {
this.configuration.bus.publish(config);
},
subscribe: function(config) {
return this.configuration.bus.subscribe(config);
},
unsubscribe: function(config) {
this.configuration.bus.unsubscribe(config);
createChannel: function(exchange, topic) {
var exch = arguments.length === 2 ? exchange : DEFAULT_EXCHANGE,
tpc = arguments.length === 2 ? topic : exchange;
if(!this.configuration.bus.subscriptions[exch]) {
this.configuration.bus.subscriptions[exch] = {};
}
if(!this.configuration.bus.subscriptions[exch][tpc]) {
this.configuration.bus.subscriptions[exch][tpc] = [];
}
return new ChannelDefinition(exch, tpc);
},
addWireTap: function(callback) {

228
build/output/nodejs/postal.js
View file

@ -5,6 +5,10 @@
Version 0.1.0
*/
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };
var DistinctPredicate = function() {
var previous;
return function(data) {
@ -20,48 +24,56 @@ var DistinctPredicate = function() {
return !eq;
};
};
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };
var defaultConfiguration = {
exchange: DEFAULT_EXCHANGE,
topic: "",
callback: NO_OP,
priority: DEFAULT_PRIORITY,
constraints: [],
disposeAfter: DEFAULT_DISPOSEAFTER,
onHandled: NO_OP,
context: null,
modifiers: []
var ChannelDefinition = function(exchange, topic) {
this.exchange = exchange;
this.topic = topic;
};
var ChannelDefinition = function(exchange, topic) {
this.configuration = _.extend(defaultConfiguration, { exchange: exchange, topic: topic });
} ;
ChannelDefinition.prototype = {
exchange: function(exchange) {
this.configuration.exchange = exchange;
return this;
subscribe: function(callback) {
var subscription = new SubscriptionDefinition(this.exchange, this.topic, callback);
postal.configuration.bus.subscribe(subscription);
return subscription;
},
topic: function(topic) {
this.configuration.topic = topic;
return this;
publish: function(data) {
postal.configuration.bus.publish({
exchange: this.exchange,
topic: this.topic,
data: data,
timeStamp: new Date()
})
}
};
var SubscriptionDefinition = function(exchange, topic, callback) {
this.exchange = exchange;
this.topic = topic;
this.callback = callback;
this.priority = DEFAULT_PRIORITY;
this.constraints = [];
this.maxCalls = DEFAULT_DISPOSEAFTER;
this.onHandled = NO_OP;
this.context = null
};
SubscriptionDefinition.prototype = {
unsubscribe: function() {
postal.configuration.bus.unsubscribe(this);
},
defer: function() {
this.configuration.modifiers.push({type: "defer"});
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn,0,data);
};
return this;
},
disposeAfter: function(receiveCount) {
if(_.isNaN(receiveCount)) {
throw "The value provided to disposeAfter (receiveCount) must be a number";
disposeAfter: function(maxCalls) {
if(_.isNaN(maxCalls)) {
throw "The value provided to disposeAfter (maxCalls) must be a number";
}
this.configuration.disposeAfter = receiveCount;
this.maxCalls = maxCalls;
return this;
},
@ -74,7 +86,7 @@ ChannelDefinition.prototype = {
if(! _.isFunction(callback)) {
throw "Value provided to 'whenHandledThenExecute' must be a function";
}
this.configuration.onHandled = callback;
this.onHandled = callback;
return this;
},
@ -82,7 +94,7 @@ ChannelDefinition.prototype = {
if(! _.isFunction(predicate)) {
throw "Predicate constraint must be a function";
}
this.configuration.constraints.push(predicate);
this.constraints.push(predicate);
return this;
},
@ -95,7 +107,7 @@ ChannelDefinition.prototype = {
},
withContext: function(context) {
this.configuration.context = context;
this.context = context;
return this;
},
@ -103,7 +115,8 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "debounce", milliseconds: milliseconds});
var fn = this.callback;
this.callback = _.debounce(fn, milliseconds);
return this;
},
@ -111,7 +124,10 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "delay", milliseconds: milliseconds});
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn, milliseconds, data);
};
return this;
},
@ -119,7 +135,7 @@ ChannelDefinition.prototype = {
if(_.isNaN(priority)) {
throw "Priority must be a number";
}
this.configuration.priority = priority;
this.priority = priority;
return this;
},
@ -127,26 +143,14 @@ ChannelDefinition.prototype = {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "throttle", milliseconds: milliseconds});
var fn = this.callback;
this.callback = _.throttle(fn, milliseconds);
return this;
},
subscribe: function(callback) {
this.configuration.callback = callback || NO_OP;
return postal.subscribe(this.configuration);
},
publish: function(data) {
postal.publish({
exchange: this.configuration.exchange,
data: data,
topic: this.configuration.topic
});
}
};
var bindingsResolver = {
cache: { },
compare: function(binding, topic) {
if(this.cache[topic] && this.cache[topic][binding]) {
return true;
@ -161,38 +165,20 @@ var bindingsResolver = {
}
return result;
},
regexify: function(binding) {
return binding.replace(/\./g,"\\.") // escape actual periods
.replace(/\*/g, ".*") // asterisks match any value
.replace(/#/g, "[A-Z,a-z,0-9]*"); // hash matches any alpha-numeric 'word'
}
};
var wrapWithDelay = function(callback, config) {
return function(data) {
setTimeout(callback, config.milliseconds, data);
};
},
wrapWithDefer = function(callback) {
return function(data) {
setTimeout(callback,0,data);
}
},
wrapWithThrottle = function(callback, config) {
return _.throttle(callback, config.milliseconds);
},
wrapWithDebounce = function(callback, config) {
return _.debounce(callback, config.milliseconds);
};
var localBus = {
subscriptions: {},
wireTaps: [],
publish: function(config) {
var envelope = _.extend(defaultConfiguration, { timeStamp: new Date() }, config);
publish: function(envelope) {
_.each(this.wireTaps,function(tap) {
tap({
exchange: envelope.exchange,
@ -216,63 +202,35 @@ var localBus = {
});
},
subscribe: function(channelDef) {
var idx, found, fn, config = _.extend(defaultConfiguration, channelDef);
if(config.disposeAfter && config.disposeAfter > 0) {
fn = config.onHandled,
dispose = _.after(config.disposeAfter, _.bind(function() {
this.unsubscribe(config);
subscribe: function(subDef) {
var idx, found, fn;
if(subDef.maxCalls) {
fn = subDef.onHandled;
var dispose = _.after(subDef.maxCalls, _.bind(function() {
this.unsubscribe(subDef);
}, this));
config.onHandled = function() {
fn.apply(config.context, arguments);
subDef.onHandled = function() {
fn.apply(subDef.context, arguments);
dispose();
}
}
_.each(config.modifiers, function(modifier) {
fn = config.callback;
switch(modifier.type) {
case 'delay':
config.callback = wrapWithDelay(fn, modifier);
break;
case 'defer':
config.callback = wrapWithDefer(fn);
break;
case 'throttle':
config.callback = wrapWithThrottle(fn,modifier);
break;
case 'debounce':
config.callback = wrapWithDebounce(fn, modifier);
break;
idx = this.subscriptions[subDef.exchange][subDef.topic].length - 1;
if(!_.any(this.subscriptions[subDef.exchange][subDef.topic], function(cfg) { return cfg === subDef; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[subDef.exchange][subDef.topic][idx].priority <= subDef.priority) {
this.subscriptions[subDef.exchange][subDef.topic].splice(idx + 1, 0, subDef);
found = true;
break;
}
}
});
if(!this.subscriptions[config.exchange]) {
this.subscriptions[config.exchange] = {};
}
if(!this.subscriptions[config.exchange][config.topic]) {
this.subscriptions[config.exchange][config.topic] = [config];
}
else {
idx = this.subscriptions[config.exchange][config.topic].length - 1;
if(!_.any(this.subscriptions[config.exchange][config.topic], function(cfg) { return cfg === config; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[config.exchange][config.topic][idx].priority <= config.priority) {
this.subscriptions[config.exchange][config.topic].splice(idx + 1, 0, config);
found = true;
break;
}
}
if(!found) {
this.subscriptions[config.exchange][config.topic].unshift(config);
}
console.log("SUBSCRIBE: " + JSON.stringify(config));
if(!found) {
this.subscriptions[subDef.exchange][subDef.topic].unshift(subDef);
}
}
return _.bind(function() { this.unsubscribe(config); }, this);
return _.bind(function() { this.unsubscribe(subDef); }, this);
},
unsubscribe: function(config) {
@ -282,7 +240,6 @@ var localBus = {
for ( ; idx < len; idx++ ) {
if (this.subscriptions[config.exchange][config.topic][idx] === config) {
this.subscriptions[config.exchange][config.topic].splice( idx, 1 );
console.log("UNSUBSCRIBE: " + JSON.stringify(config));
break;
}
}
@ -300,30 +257,21 @@ var localBus = {
}
};
var postal = {
configuration: {
bus: localBus,
resolver: bindingsResolver
},
exchange: function(exchange) {
return new ChannelDefinition(exchange);
},
topic: function(topic) {
return new ChannelDefinition(undefined, topic);
},
publish: function(config) {
this.configuration.bus.publish(config);
},
subscribe: function(config) {
return this.configuration.bus.subscribe(config);
},
unsubscribe: function(config) {
this.configuration.bus.unsubscribe(config);
createChannel: function(exchange, topic) {
var exch = arguments.length === 2 ? exchange : DEFAULT_EXCHANGE,
tpc = arguments.length === 2 ? topic : exchange;
if(!this.configuration.bus.subscriptions[exch]) {
this.configuration.bus.subscriptions[exch] = {};
}
if(!this.configuration.bus.subscriptions[exch][tpc]) {
this.configuration.bus.subscriptions[exch][tpc] = [];
}
return new ChannelDefinition(exch, tpc);
},
addWireTap: function(callback) {

4
build/scripts/SourceManifest-node.txt
View file

@ -1,5 +1,7 @@
src/Constants.js
src/DistinctPredicate.js
src/ChannelDefinition.js
src/SubscriptionDefinition.js
src/BindingsResolver.js
src/Bus.js
src/LocalBus.js
src/Postal.js

4
build/scripts/source-browser-postal.txt
View file

@ -1,5 +1,7 @@
src/Constants.js
src/DistinctPredicate.js
src/ChannelDefinition.js
src/SubscriptionDefinition.js
src/BindingsResolver.js
src/Bus.js
src/LocalBus.js
src/Postal.js

10
spec/runner.html
View file

@ -7,11 +7,11 @@
<script type="text/javascript" src="../lib/amplify.core.js"></script>
<script type="text/javascript" src="../lib/amplify.store.js"></script>
<script type="text/javascript" src="../lib/underscore.js"></script>
<script type="text/javascript" src="../src/BindingsResolver.js"></script>
<script type="text/javascript" src="../src/DistinctPredicate.js"></script>
<script type="text/javascript" src="../src/ChannelDefinition.js"></script>
<script type="text/javascript" src="../src/Bus.js"></script>
<script type="text/javascript" src="../src/Postal.js"></script>
<script type="text/javascript" src="../src/old/BindingsResolver.js"></script>
<script type="text/javascript" src="../src/old/DistinctPredicate.js"></script>
<script type="text/javascript" src="../src/old/ChannelDefinition.js"></script>
<script type="text/javascript" src="../src/old/Bus.js"></script>
<script type="text/javascript" src="../src/old/Postal.js"></script>
<script type="text/javascript" src="BindingsResolver.spec.js"></script>
<script type="text/javascript" src="DistinctPredicate.spec.js"></script>
<script type="text/javascript" src="ChannelDefinition.spec.js"></script>

4
src/BindingsResolver.js
View file

@ -1,6 +1,6 @@
var bindingsResolver = {
cache: { },
compare: function(binding, topic) {
if(this.cache[topic] && this.cache[topic][binding]) {
return true;
@ -15,7 +15,7 @@ var bindingsResolver = {
}
return result;
},
regexify: function(binding) {
return binding.replace(/\./g,"\\.") // escape actual periods
.replace(/\*/g, ".*") // asterisks match any value

131
src/Bus.js
View file

@ -1,131 +0,0 @@
var wrapWithDelay = function(callback, config) {
return function(data) {
setTimeout(callback, config.milliseconds, data);
};
},
wrapWithDefer = function(callback) {
return function(data) {
setTimeout(callback,0,data);
}
},
wrapWithThrottle = function(callback, config) {
return _.throttle(callback, config.milliseconds);
},
wrapWithDebounce = function(callback, config) {
return _.debounce(callback, config.milliseconds);
};
var localBus = {
subscriptions: {},
wireTaps: [],
publish: function(config) {
var envelope = _.extend(defaultConfiguration, { timeStamp: new Date() }, config);
_.each(this.wireTaps,function(tap) {
tap({
exchange: envelope.exchange,
topic: envelope.topic,
data: envelope.data,
timeStamp: envelope.timeStamp
});
});
_.each(this.subscriptions[envelope.exchange], function(topic) {
_.each(topic, function(binding){
if(postal.configuration.resolver.compare(binding.topic, envelope.topic)) {
if(_.all(binding.constraints, function(constraint) { return constraint(envelope.data); })) {
if(typeof binding.callback === 'function') {
binding.callback.apply(binding.context, [envelope.data]);
binding.onHandled();
}
}
}
});
});
},
subscribe: function(channelDef) {
var idx, found, fn, config = _.extend(defaultConfiguration, channelDef);
if(config.disposeAfter && config.disposeAfter > 0) {
fn = config.onHandled,
dispose = _.after(config.disposeAfter, _.bind(function() {
this.unsubscribe(config);
}, this));
config.onHandled = function() {
fn.apply(config.context, arguments);
dispose();
}
}
_.each(config.modifiers, function(modifier) {
fn = config.callback;
switch(modifier.type) {
case 'delay':
config.callback = wrapWithDelay(fn, modifier);
break;
case 'defer':
config.callback = wrapWithDefer(fn);
break;
case 'throttle':
config.callback = wrapWithThrottle(fn,modifier);
break;
case 'debounce':
config.callback = wrapWithDebounce(fn, modifier);
break;
}
});
if(!this.subscriptions[config.exchange]) {
this.subscriptions[config.exchange] = {};
}
if(!this.subscriptions[config.exchange][config.topic]) {
this.subscriptions[config.exchange][config.topic] = [config];
}
else {
idx = this.subscriptions[config.exchange][config.topic].length - 1;
if(!_.any(this.subscriptions[config.exchange][config.topic], function(cfg) { return cfg === config; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[config.exchange][config.topic][idx].priority <= config.priority) {
this.subscriptions[config.exchange][config.topic].splice(idx + 1, 0, config);
found = true;
break;
}
}
if(!found) {
this.subscriptions[config.exchange][config.topic].unshift(config);
}
console.log("SUBSCRIBE: " + JSON.stringify(config));
}
}
return _.bind(function() { this.unsubscribe(config); }, this);
},
unsubscribe: function(config) {
if(this.subscriptions[config.exchange][config.topic]) {
var len = this.subscriptions[config.exchange][config.topic].length,
idx = 0;
for ( ; idx < len; idx++ ) {
if (this.subscriptions[config.exchange][config.topic][idx] === config) {
this.subscriptions[config.exchange][config.topic].splice( idx, 1 );
console.log("UNSUBSCRIBE: " + JSON.stringify(config));
break;
}
}
}
},
addWireTap: function(callback) {
this.wireTaps.push(callback);
return function() {
var idx = this.wireTaps.indexOf(callback);
if(idx !== -1) {
this.wireTaps.splice(idx,1);
}
};
}
};

127
src/ChannelDefinition.js
View file

@ -1,124 +1,21 @@
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };
var defaultConfiguration = {
exchange: DEFAULT_EXCHANGE,
topic: "",
callback: NO_OP,
priority: DEFAULT_PRIORITY,
constraints: [],
disposeAfter: DEFAULT_DISPOSEAFTER,
onHandled: NO_OP,
context: null,
modifiers: []
var ChannelDefinition = function(exchange, topic) {
this.exchange = exchange;
this.topic = topic;
};
var ChannelDefinition = function(exchange, topic) {
this.configuration = _.extend(defaultConfiguration, { exchange: exchange, topic: topic });
} ;
ChannelDefinition.prototype = {
exchange: function(exchange) {
this.configuration.exchange = exchange;
return this;
},
topic: function(topic) {
this.configuration.topic = topic;
return this;
},
defer: function() {
this.configuration.modifiers.push({type: "defer"});
return this;
},
disposeAfter: function(receiveCount) {
if(_.isNaN(receiveCount)) {
throw "The value provided to disposeAfter (receiveCount) must be a number";
}
this.configuration.disposeAfter = receiveCount;
return this;
},
ignoreDuplicates: function() {
this.withConstraint(new DistinctPredicate());
return this;
},
whenHandledThenExecute: function(callback) {
if(! _.isFunction(callback)) {
throw "Value provided to 'whenHandledThenExecute' must be a function";
}
this.configuration.onHandled = callback;
return this;
},
withConstraint: function(predicate) {
if(! _.isFunction(predicate)) {
throw "Predicate constraint must be a function";
}
this.configuration.constraints.push(predicate);
return this;
},
withConstraints: function(predicates) {
var self = this;
if(_.isArray(predicates)) {
_.each(predicates, function(predicate) { self.withConstraint(predicate); } );
}
return self;
},
withContext: function(context) {
this.configuration.context = context;
return this;
},
withDebounce: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "debounce", milliseconds: milliseconds});
return this;
},
withDelay: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "delay", milliseconds: milliseconds});
return this;
},
withPriority: function(priority) {
if(_.isNaN(priority)) {
throw "Priority must be a number";
}
this.configuration.priority = priority;
return this;
},
withThrottle: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
this.configuration.modifiers.push({type: "throttle", milliseconds: milliseconds});
return this;
},
subscribe: function(callback) {
this.configuration.callback = callback || NO_OP;
return postal.subscribe(this.configuration);
var subscription = new SubscriptionDefinition(this.exchange, this.topic, callback);
postal.configuration.bus.subscribe(subscription);
return subscription;
},
publish: function(data) {
postal.publish({
exchange: this.configuration.exchange,
data: data,
topic: this.configuration.topic
});
postal.configuration.bus.publish({
exchange: this.exchange,
topic: this.topic,
data: data,
timeStamp: new Date()
})
}
};

4
src/Constants.js Normal file
View file

@ -0,0 +1,4 @@
var DEFAULT_EXCHANGE = "/",
DEFAULT_PRIORITY = 50,
DEFAULT_DISPOSEAFTER = 0,
NO_OP = function() { };

84
src/LocalBus.js Normal file
View file

@ -0,0 +1,84 @@
var localBus = {
subscriptions: {},
wireTaps: [],
publish: function(envelope) {
_.each(this.wireTaps,function(tap) {
tap({
exchange: envelope.exchange,
topic: envelope.topic,
data: envelope.data,
timeStamp: envelope.timeStamp
});
});
_.each(this.subscriptions[envelope.exchange], function(topic) {
_.each(topic, function(binding){
if(postal.configuration.resolver.compare(binding.topic, envelope.topic)) {
if(_.all(binding.constraints, function(constraint) { return constraint(envelope.data); })) {
if(typeof binding.callback === 'function') {
binding.callback.apply(binding.context, [envelope.data]);
binding.onHandled();
}
}
}
});
});
},
subscribe: function(subDef) {
var idx, found, fn;
if(subDef.maxCalls) {
fn = subDef.onHandled;
var dispose = _.after(subDef.maxCalls, _.bind(function() {
this.unsubscribe(subDef);
}, this));
subDef.onHandled = function() {
fn.apply(subDef.context, arguments);
dispose();
}
}
idx = this.subscriptions[subDef.exchange][subDef.topic].length - 1;
if(!_.any(this.subscriptions[subDef.exchange][subDef.topic], function(cfg) { return cfg === subDef; })) {
for(; idx >= 0; idx--) {
if(this.subscriptions[subDef.exchange][subDef.topic][idx].priority <= subDef.priority) {
this.subscriptions[subDef.exchange][subDef.topic].splice(idx + 1, 0, subDef);
found = true;
break;
}
}
if(!found) {
this.subscriptions[subDef.exchange][subDef.topic].unshift(subDef);
}
}
return _.bind(function() { this.unsubscribe(subDef); }, this);
},
unsubscribe: function(config) {
if(this.subscriptions[config.exchange][config.topic]) {
var len = this.subscriptions[config.exchange][config.topic].length,
idx = 0;
for ( ; idx < len; idx++ ) {
if (this.subscriptions[config.exchange][config.topic][idx] === config) {
this.subscriptions[config.exchange][config.topic].splice( idx, 1 );
break;
}
}
}
},
addWireTap: function(callback) {
this.wireTaps.push(callback);
return function() {
var idx = this.wireTaps.indexOf(callback);
if(idx !== -1) {
this.wireTaps.splice(idx,1);
}
};
}
};

29
src/Postal.js
View file

@ -1,28 +1,19 @@
var postal = {
configuration: {
bus: localBus,
resolver: bindingsResolver
},
exchange: function(exchange) {
return new ChannelDefinition(exchange);
},
topic: function(topic) {
return new ChannelDefinition(undefined, topic);
},
publish: function(config) {
this.configuration.bus.publish(config);
},
subscribe: function(config) {
return this.configuration.bus.subscribe(config);
},
unsubscribe: function(config) {
this.configuration.bus.unsubscribe(config);
createChannel: function(exchange, topic) {
var exch = arguments.length === 2 ? exchange : DEFAULT_EXCHANGE,
tpc = arguments.length === 2 ? topic : exchange;
if(!this.configuration.bus.subscriptions[exch]) {
this.configuration.bus.subscriptions[exch] = {};
}
if(!this.configuration.bus.subscriptions[exch][tpc]) {
this.configuration.bus.subscriptions[exch][tpc] = [];
}
return new ChannelDefinition(exch, tpc);
},
addWireTap: function(callback) {

103
src/SubscriptionDefinition.js Normal file
View file

@ -0,0 +1,103 @@
var SubscriptionDefinition = function(exchange, topic, callback) {
this.exchange = exchange;
this.topic = topic;
this.callback = callback;
this.priority = DEFAULT_PRIORITY;
this.constraints = [];
this.maxCalls = DEFAULT_DISPOSEAFTER;
this.onHandled = NO_OP;
this.context = null
};
SubscriptionDefinition.prototype = {
unsubscribe: function() {
postal.configuration.bus.unsubscribe(this);
},
defer: function() {
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn,0,data);
};
return this;
},
disposeAfter: function(maxCalls) {
if(_.isNaN(maxCalls)) {
throw "The value provided to disposeAfter (maxCalls) must be a number";
}
this.maxCalls = maxCalls;
return this;
},
ignoreDuplicates: function() {
this.withConstraint(new DistinctPredicate());
return this;
},
whenHandledThenExecute: function(callback) {
if(! _.isFunction(callback)) {
throw "Value provided to 'whenHandledThenExecute' must be a function";
}
this.onHandled = callback;
return this;
},
withConstraint: function(predicate) {
if(! _.isFunction(predicate)) {
throw "Predicate constraint must be a function";
}
this.constraints.push(predicate);
return this;
},
withConstraints: function(predicates) {
var self = this;
if(_.isArray(predicates)) {
_.each(predicates, function(predicate) { self.withConstraint(predicate); } );
}
return self;
},
withContext: function(context) {
this.context = context;
return this;
},
withDebounce: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
var fn = this.callback;
this.callback = _.debounce(fn, milliseconds);
return this;
},
withDelay: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
var fn = this.callback;
this.callback = function(data) {
setTimeout(fn, milliseconds, data);
};
return this;
},
withPriority: function(priority) {
if(_.isNaN(priority)) {
throw "Priority must be a number";
}
this.priority = priority;
return this;
},
withThrottle: function(milliseconds) {
if(_.isNaN(milliseconds)) {
throw "Milliseconds must be a number";
}
var fn = this.callback;
this.callback = _.throttle(fn, milliseconds);
return this;
}
};