chore(): initial commit

Author: marcghorayeb

.gitignore

@@ -0,0 +1,2 @@
+.DS_Store
+node_modules/

.jscsrc

@@ -0,0 +1,50 @@
+{
+	"validateIndentation": "\t",
+
+	"requireSpaceAfterKeywords": [
+		"if",
+		"else",
+		"for",
+		"while",
+		"do",
+		"switch",
+		"case",
+		"return",
+		"try",
+		"catch",
+		"typeof"
+	],
+
+	"requireSpacesInNamedFunctionExpression": {
+		"beforeOpeningCurlyBrace": true
+	},
+	"disallowSpacesInNamedFunctionExpression": {
+		"beforeOpeningRoundBrace": true
+	},
+	"requireSpacesInAnonymousFunctionExpression": {
+		"beforeOpeningRoundBrace": true,
+		"beforeOpeningCurlyBrace": true
+	},
+	"validateParameterSeparator": ", ",
+
+	"disallowMultipleVarDecl": true,
+
+	"requireSpacesInConditionalExpression": true,
+	"disallowYodaConditions": true,
+
+	"disallowSpacesInsideParentheses": true,
+	"requireSpacesInsideObjectBrackets": "all",
+	"disallowSpaceAfterObjectKeys": true,
+	"requireSpaceBeforeObjectValues": true,
+	"requireSpacesInsideArrayBrackets": "all",
+	"disallowTrailingComma": true,
+
+	"requireSpaceAfterLineComment": true,
+
+	"requireSpaceBeforeBlockStatements": true,
+	"disallowNewlineBeforeBlockStatements": true,
+	"disallowPaddingNewlinesInBlocks": true,
+
+	"disallowTrailingWhitespace": true,
+	"validateLineBreaks": "LF"
+}

.jshintrc

@@ -0,0 +1,20 @@
+{
+	"camelcase": true,
+	"devel": true,
+	"immed": true,
+	"latedef": true,
+	"maxlen": 120,
+	"node": true,
+	"nonbsp": true,
+	"trailing": true,
+	"undef": true,
+	"unused": true,
+	"white": true,
+	"globals": {
+		"angular": true,
+		"describe": true,
+		"it": true,
+		"expect": true,
+		"beforeEach": true
+	}
+}

amqp.js

@@ -0,0 +1,32 @@
+var _ = require('lodash');
+var amqplib = require('amqplib');
+var format = require('util').format;
+
+function AMQP(config) {
+	var uri = config;
+
+	if (_.isObject(uri)) {
+		_.defaults(uri, {
+			user: '',
+			password: '',
+			host: '',
+			vhost: ''
+		});
+
+		uri = format(
+			'amqp://%s:%s@%s/%s',
+			uri.user,
+			uri.password,
+			uri.host,
+			uri.vhost
+		);
+	}
+
+	if (!_.isString(uri) || !uri) throw new Error('Malformed config');
+
+	return amqplib.connect(uri).then(function (conn) {
+		return conn.createChannel();
+	});
+}
+
+module.exports = AMQP;

consumer.js

@@ -0,0 +1,50 @@
+function Consumer(amqp) {
+	if (!this.domain) throw new Error('Missing domain property');
+	if (!amqp) throw new Error('Missing amqp connection');
+
+	this.amqp = amqp;
+
+	this.consume(this.domain, this.handleEvent.bind(this));
+}
+
+Consumer.prototype.connect = function (callback) {
+	var self = this;
+	this.amqp.then(function (channel) {
+		self.channel = channel;
+		callback(channel);
+	});
+};
+
+Consumer.prototype.consume = function (queue, callback) {
+	var self = this;
+
+	this.channel.assertQueue(queue);
+
+	this.channel.consume(queue, function (msg) {
+		self.handleMessage(msg, callback);
+	});
+};
+
+Consumer.prototype.handleMessage = function (msg, callback) {
+	var data = JSON.parse(msg.content.toString());
+	callback(data, this.onProcessed.bind(this));
+};
+
+Consumer.prototype.handleEvent = function (event, callback) {
+	if (!event.type) throw new Error('Missing event type');
+	if (!this.projections[event.type]) throw new Error('No projections for ' + event.type);
+	this.projections[event.type].bind(this)(event.data || event.args || {}, callback);
+};
+
+Consumer.prototype.onProcessed = function (err, results) {
+	this.channel.ack(msg);
+
+	if (!msg.properties.replyTo) return;
+
+	var reply = { err: err, results: results };
+	var rpcConfig = { correlationId: msg.properties.correlationId };
+
+	this.channel.sendToQueue(msg.properties.replyTo, new Buffer(JSON.stringify(reply)), rpcConfig);
+};
+
+module.exports = Consumer;

event-sourcing.js

@@ -0,0 +1,7 @@
+var consumer = require('consumer');
+var publisher = require('publisher');
+
+module.exports = {
+	consumer: consumer,
+	publisher: publisher
+};

mongodb.js

@@ -0,0 +1,37 @@
+var _ = require('lodash');
+var format = require('util').format;
+var MongoClient = require('mongodb').MongoClient;
+
+function MongoDB(config) {
+	this.uri = config;
+
+	if (_.isObject(this.uri)) {
+		_.defaults(this.uri, {
+			host: '',
+			port: '',
+			db: ''
+		});
+
+		this.uri = format(
+			'mongodb://%s:%s/%s',
+			this.uri.host,
+			this.uri.port,
+			this.uri.db
+		);
+	}
+
+	if (!_.isString(this.uri) || !this.uri) throw new Error('Malformed config');
+}
+
+MongoDB.prototype.connect = function (callback) {
+	if (this.db) return callback(null, this.db);
+
+	var self = this;
+	MongoClient.connect(this.uri, function (err, db) {
+		if (err) throw err;
+		self.db = db;
+		callback(err, db);
+	});
+};
+
+module.exports = MongoDB;

package.json

@@ -0,0 +1,25 @@
+{
+  "name": "event-sourcing",
+  "version": "0.0.1",
+  "description": "NodeJS Event-sourcing POC",
+  "main": "event-sourcing.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/marcghorayeb/event-sourcing"
+  },
+  "author": "Marc Ghorayeb <[email protected]>",
+  "license": "ISC",
+  "bugs": {
+    "url": "https://github.com/marcghorayeb/event-sourcing/issues"
+  },
+  "homepage": "https://github.com/marcghorayeb/event-sourcing",
+  "dependencies": {
+    "amqplib": "^0.3.2",
+    "lodash": "^3.7.0",
+    "mongodb": "^2.0.28",
+    "node-uuid": "^1.4.3"
+  }
+}

publisher.js

@@ -0,0 +1,62 @@
+var amqp = require('./amqp');
+var uuid = require('node-uuid');
+
+function Publisher(amqp, db) {
+	this.amqp = amqp;
+
+	var self = this;
+	db.collection('events', function (err, col) {
+		self.col = col;
+	});
+}
+
+Publisher.prototype.connect = function (callback) {
+	var self = this;
+	this.amqp.then(function (channel) {
+		self.channel = channel;
+		callback(channel);
+	});
+};
+
+Publisher.prototype.askRPC = function (queue, event, callback) {
+	var self = this;
+	this.persistEvent(event, function (err) {
+		if (err) throw err;
+		self.emitEvent(queue, event, callback);
+	});
+};
+
+Publisher.prototype.persistEvent = function (event, callback) {
+	this.col.insertOne(event, function (err, results) {
+		if (err) throw err;
+		callback(err, results);
+	});
+};
+
+Publisher.prototype.assertReplyQueue = function () {
+	return this.channel.assertQueue('', { exclusive: true }).then(function (queueAssertion) {
+		return queueAssertion.queue;
+	});
+};
+
+Publisher.prototype.emitEvent = function (queue, event, callback) {
+	event = new Buffer(JSON.stringify(event));
+
+	if (!callback) return this.channel.sendToQueue(queue, event);
+
+	var self = this;
+	var rpcConfig = { correlationId: uuid.v4(), replyTo: null };
+	var rpcCallback = function (msg) {
+		if (msg.properties.correlationId !== rpcConfig.correlationId) return;
+		callback(msg);
+	};
+
+	this.assertReplyQueue().then(function (responseQ) {
+		self.channel.consume(responseQ, rpcCallback, { noAck: true }).then(function () {
+			rpcConfig.replyTo = responseQ;
+			self.channel.sendToQueue(queue, event, rpcConfig);
+		});
+	});
+};
+
+module.exports = Publisher;