moscajs / ascoltatori Goto Github PK
View Code? Open in Web Editor NEWThe pub/sub library for node backed by Redis, MongoDB, AMQP (RabbitMQ), ZeroMQ, MQTT (Mosquitto) or just plain node!
Home Page: http://mcollina.github.com/ascoltatori
The pub/sub library for node backed by Redis, MongoDB, AMQP (RabbitMQ), ZeroMQ, MQTT (Mosquitto) or just plain node!
Home Page: http://mcollina.github.com/ascoltatori
It seems fair to have some good code coverage built in (for version 1.0.0 #63). Instanbul seems the best tool for the job.
http://xseignard.github.io/2013/04/25/quality-analysis-on-node.js-projects-with-mocha-istanbul-and-sonar/
http://gotwarlost.github.io/istanbul/
In order to get this working, we need to add QoS 1 support into Mosca.
Publishing null or false pops out as true in the subscriber callback.
Eg.
var ascoltatori = require('ascoltatori');
var settings = {
type: 'redis',
redis: require('redis'),
db: 0,
port: 6379,
host: "127.0.0.1"
};
ascoltatori.build(settings, function (ascoltatore) {
ascoltatore.subscribe("*", function(topic, msg, done) {
console.log("RX %s => %j", topic, msg);
});
setTimeout(function() {
ascoltatore.publish("hello", null);
ascoltatore.publish("hello", false);
}, 1000);
});
An Ascoltatore is a subclass of EventEmitter, as an EventEmitter it should support unlimited subscribers to its own events.
This is not relevant to the subscribers, which are already unlimited for every Ascoltatore.
Currently at least in RedisAscoltatore calling reset does not clear the subscriptions.
An ascoltatore cannot know at the beginning all the other peers, so it will need to discover them when connecting.
The documentation should give the user a simple configuration to easily understand how to make the setup. There is only one in redis. The main problem is that looking at the test suite, and checking at common.js you can make some errors, because if you do not set the type, the system will always work on memory.
Converting wildcard topic to regexp (https://github.com/mcollina/ascoltatori/blob/master/lib/memory_ascoltatore.js#L45) means this topic, for example:
foo{2}/#
Doesn't match messages with topic 'foo{2}/bar' but does match messages with topic 'fooo/bar'.
(compare with mosquitto which doesn't interpret as regexps)
649ReferenceError: regexp is not defined
650 at MemoryAscoltatore.unsubscribe (/home/travis/builds/mcollina/qest/node_modules/ascoltatori/lib/memory_ascoltatore.js:63:18)
651 at SubsCounter.forEach (/home/travis/builds/mcollina/qest/node_modules/ascoltatori/lib/subs_counter.js:45:7)
652 at EventEmitter.unsubscribe (/home/travis/builds/mcollina/qest/node_modules/ascoltatori/lib/memory_ascoltatore.js:62:15)
653 at EventEmitter.unsubscribe (/home/travis/builds/mcollina/qest/node_modules/ascoltatori/lib/redis_ascoltatore.js:79:21)
654 at Function.module.exports.Data.unsubscribe (/home/travis/builds/mcollina/qest/app/models/data.coffee:140:23)
655 at module.exports.unsubscribeAll (/home/travis/builds/mcollina/qest/app/controllers/mqtt_api.coffee:14:30)
656 at Connection.module.exports.client.on.client.unsuback.messageId (/home/travis/builds/mcollina/qest/app/controllers/mqtt_api.coffee:86:16)
657 at Connection.EventEmitter.emit (events.js:93:17)
658 at Socket.module.exports (/home/travis/builds/mcollina/qest/node_modules/mqttjs/lib/connection.js:22:14)
659 at Socket.EventEmitter.emit (events.js:93:17)
It can be easily packaged with browserify 2
We should be able to remove newTopic everywhere now, right?
It should be possible to craft a MongoAscolatore based on:
The JSONAscolatore will serialize EVERY message to json and decompose it when they pop out.
They just hangs.
From travis:
2875․ ascoltatori:memory closed +0ms
2876 mocha:runner run suite function RedisAscoltatore(opts) {
2877 AbstractAscoltatore.call(this);
2878
2879 this._ready_sub = false;
2880 this._ready_pub = false;
2881 this._opts = opts || {};
2882 this._opts.redis = this._opts.redis || require("redis");
2883
2884 this._ascoltatore = new MemoryAscoltatore();
2885 this._startSub();
2886 this._startPub();
2887
2888 this._subs_counter = new SubsCounter();
2889 this._ascoltatore.on("newTopic", this.emit.bind(this, "newTopic"));
2890} wrapping uncaughtException +81ms
2891 ascoltatori:redis connecting to undefined:undefined +0ms
2892 ascoltatori:redis selected database undefined +1ms
2893 ascoltatori:redis connecting to undefined:undefined +0ms
2894 ascoltatori:redis selected database undefined +0ms
2895 ascoltatori:redis closed +0ms
2896 ascoltatori:redis created pub connection +0ms
2897 ascoltatori:redis created sub connection +1ms
2898 ascoltatori:memory registered domain +2ms
2899 ascoltatori:memory registered new subscriber for topic throw +1ms
2900 ascoltatori:memory new topic: throw +0ms
2901 ascoltatori:redis registered new subscriber for topic throw +1ms
2902 ascoltatori:redis new message published to throw +0ms
2903 ascoltatori:redis new message received for topic throw +0ms
2904 ascoltatori:memory new message published to throw +0ms
2905․
This is needed to fix moscajs/mosca#9
Providing a safe "reset" method is actually hard.
It's way simpler to provide an "end" method to actually drop everything.
https://github.com/mcollina/ascoltatori/blob/master/lib/memory_ascoltatore.js#L95
forEach is just a for..in.. loop on _set.
But inside that unsub is called, which will end up doing _set.remove
Modifying _set while iterating over it probably isn't a good idea.
As users might want to reuse my test to develop their ascoltatori, it will be better if the test can be included directly from the published package.
Currently leaving a wildcard subscription open may cause multiple message deliveries and a memory leak.
The JSON serialization is not working as expected.
I am currently planning to separate it from the transports, and then allow to specify a setting to enable/disable it.
This will allow much a much greater flexibility, right now I need to support too many use cases and its really impossible to adress them all.
New cool client API.
It should also simplify the Ascoltatore 🍻
Currently publishing and subscribing uses strings.
We might want to support also objects, in order to specify transport-specific options.
I am think about releasing a 1.0.0 version to stabilize each Ascoltatore API.
I think it is fairly stable right now.
What I would like to include:
I'd like to access the qos and messageId properties of the mqtt message received.
At the moment, publish() just gets the topic and payload.
Adding (or replacing) with the packet is likely to change the API.
I'm happy to do the change but should this be an extra arg but which to choose?
Now: publish(topic, message, done)
Just pass packet: publish(packet, done)
Add packet: publish(topic, message, packet, done)
Backwards compatible: publish(topic, message, done, packet)
?
It can be done using https://github.com/chrisdew/barricane-db or https://github.com/1602/jugglingdb
If the reset call is issued before "ready", no callback is emitted
Ideally a bunch on clustered process should be able to exchange via pub/sub easily without external server setup.
This might be possible using the cluster node package.
The real hassle is how to test it.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.