added kafka ascoltatore

.travis.yml

@@ -13,6 +13,12 @@ before_install:
   - if [ "$MONGODB_VERSION" = "3.0*" ]; then echo "deb http://repo.mongodb.org/apt/ubuntu "$(lsb_release -sc)"/mongodb-org/3.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-3.0.list; fi
   - if [ "$MONGODB_VERSION" = "3.2*" ]; then echo "deb http://repo.mongodb.org/apt/ubuntu "$(lsb_release -sc)"/mongodb-org/testing multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-3.1.list; fi
   - sudo apt-get update -qq
+  - wget http://www.us.apache.org/dist/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz -O kafka.tgz
+  - mkdir -p kafka && tar xzf kafka.tgz -C kafka --strip-components 1
+  - nohup bash -c "cd kafka && bin/zookeeper-server-start.sh config/zookeeper.properties &"
+  - sleep 10
+  - nohup bash -c "cd kafka && bin/kafka-server-start.sh config/server.properties &"
+  - sleep 10
 
 install:
   - sudo apt-get install libzmq3-dev

README.md

@@ -12,6 +12,7 @@ Ascoltatori is a simple publish/subscribe library supporting the following broke
 * [RabbitMQ](http://www.rabbitmq.com/) and all implementations of the [AMQP](http://www.amqp.org/) protocol.
 * [ZeroMQ](http://www.zeromq.org/) to use Ascoltatori in a P2P fashion.
 * [QlobberFSQ](https://github.com/davedoesdev/qlobber-fsq), a shared file system queue.
+* [Apache Kafka](http://kafka.apache.org), a high-throughput distributed messaging system.
 * Memory-only routing, using [Qlobber](https://github.com/davedoesdev/qlobber).
 
 Find out more about Ascoltatori reading the
@@ -289,6 +290,32 @@ require('ascoltatori').build({ json: false }, function(err, a) {
 });
 ```
 
+### Apache Kafka
+
+```javascript
+var ascoltatori = require('ascoltatori');
+var settings = {
+  type: 'kafka',
+  json: false,
+  kafka: require("kafka-node"),
+  connectString: "localhost:2181",
+  clientId: "ascoltatori",
+  groupId: "ascoltatori",
+  defaultEncoding: "utf8",
+  encodings: {
+    image: "buffer"
+  }
+};
+
+ascoltatori.build(settings, function (err, ascoltatore) {
+  // ...
+});
+```
+
+If you publish to a kafka topic that doesn't exist, that topic will be created using the default settings.
+
+If you subscribe to a kafka topic that doesn't exist, that subscription will take affect only when something is published to the kafka topic through this ascoltatori.
+
 ## Domain support
 
 Ascoltatori supports the [node.js domain API](http://nodejs.org/api/domain.html).
@@ -336,6 +363,7 @@ The following debug flags are supported:
 * `ascoltatori:zmq`
 * `ascoltatori:ee2`
 * `ascoltatori:filesystem`
+* `ascoltatori:kafka`
 
 
 ## Reliability
@@ -362,6 +390,7 @@ Use the [issue tracker](http://github.com/mcollina/ascoltatori/issues) for bugs.
 * [Mosquitto](http://mosquitto.org/)
 * [RabbitMQ](http://www.rabbitmq.com/)
 * [ZeroMQ](http://www.zeromq.org/)
+* [Apache Kafka](http://kafka.apache.org/)
 
 
 ## Authors

lib/ascoltatori.js

@@ -20,6 +20,7 @@ module.exports.MongoAscoltatore = require('./mongo_ascoltatore');
 module.exports.DecoratorAscoltatore = require("./decorator_ascoltatore");
 module.exports.JSONAscoltatore = require("./json_ascoltatore");
 module.exports.FileSystemAscoltatore = require("./filesystem_ascoltatore");
+module.exports.KafkaAscoltatore = require("./kafka_ascoltatore");
 
 /**
  *
@@ -33,6 +34,7 @@ var classes = {
   "redis": module.exports.RedisAscoltatore,
   "zmq": module.exports.ZeromqAscoltatore,
   "mongo": module.exports.MongoAscoltatore,
+  "kafka": module.exports.KafkaAscoltatore,
   "filesystem": module.exports.FileSystemAscoltatore
 };