diff --git a/client.js b/client.js new file mode 100644 index 000000000..7d6110adc --- /dev/null +++ b/client.js @@ -0,0 +1,43 @@ +var Container = machinetalk.protobuf.message.Container; +var ContainerType = machinetalk.protobuf.message.ContainerType; + +function get_uuid () { + return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) { + var r = crypto.getRandomValues(new Uint8Array(1))[0]%16|0, v = c == 'x' ? r : (r&0x3|0x8); + return v.toString(16); + }); +} +var uuidDealer = get_uuid(); +var uuidSub = get_uuid(); +var socket = io(); + +$('form').submit(function(){ + var msg = {type: ContainerType.MT_PING, name: $('#m').val()}; + var encoded = Container.encode(msg).toArrayBuffer(); + socket.emit(uuidDealer + 'msg', encoded); + $('#m').val(''); + return false; +}); + +socket.on('connect', function() { + socket.emit('connect socket', {uri: 'tcp://127.0.0.1:12345', type: 'dealer', uuid: uuidDealer}); + socket.emit('connect socket', {uri: 'tcp://127.0.0.1:12346', type: 'sub', uuid: uuidSub}); +}); + +socket.on(uuidDealer + 'msg', function(msg) { + var bier = Container.decode(msg[0]); + $('#messages').append($('

').text("me: " + bier.name)); +}); + +socket.on(uuidSub + 'msg', function(msg) { + var bier = Container.decode(msg[1]); + $('#messages').append($('

').text("other: " + bier.name)); +}); + +socket.on(uuidSub + 'connected', function() { + socket.emit(uuidSub + 'subscribe', 'msg'); +}); + +$(window).on('beforeunload', function() { + socket.close(); +}); diff --git a/index.html b/index.html index 5f2767e98..945579755 100644 --- a/index.html +++ b/index.html @@ -16,20 +16,14 @@

- + + + + diff --git a/index.js b/index.js index a322dca0f..b98a3b8eb 100644 --- a/index.js +++ b/index.js @@ -1,17 +1,38 @@ var app = require('express')(); var http = require('http').Server(app); var io = require('socket.io')(http); +var _ = require('underscore'); +var SocketManager = require('./socketmanager.js'); app.get('/', function(req, res){ res.sendFile(__dirname + '/index.html'); }); - -io.on('connection', function(socket){ - socket.on('chat message', function(msg){ - io.emit('chat message', msg); - }); +app.get('/client.js', function(req, res){ + res.sendFile(__dirname + '/client.js'); +}); +app.get('/machinetalk-protobuf.min.js', function(req, res){ + res.sendFile(__dirname + '/machinetalk-protobuf.min.js'); }); + +var socketManager = new SocketManager(io); +// var retVal = socketManager.connectSocket({type: 'dealer', uri: 'tcp://127.0.0.1:12345'}); +// var subscriber = socketManager.connectSocket({type: 'sub', uri: 'tcp://127.0.0.1:12346'}); +// var subscriber2 = socketManager.connectSocket({type: 'sub', uri: 'tcp://127.0.0.1:12346'}); +// var sendMessage = function() { +// retVal.socket.send('bla'); +// } +// setInterval(sendMessage, 500); +// subscriber.socket.subscribe('topic1'); +// subscriber2.socket.subscribe('topic1'); +// subscriber.socket.on('message', function (topic, msg) { +// console.log('got', msg.toString()); +// }); + +// retVal.socket.on('message', function (msg) { +// console.log('got reply', msg.toString()); +// }); + http.listen(3000, function(){ console.log('listening on *:3000'); }); diff --git a/package.json b/package.json index 46f75814b..d5ab87e08 100644 --- a/package.json +++ b/package.json @@ -4,6 +4,10 @@ "description": "my first socket.io app", "dependencies": { "express": "4.10.2", - "socket.io": "1.2.0" + "socket.io": "1.2.0", + "protobufjs": "^4.0.0", + "underscore": "^1.8.3", + "zmq": "^2.13.0", + "uuid": "2.0.1" } } diff --git a/responder.js b/responder.js new file mode 100644 index 000000000..441877bfe --- /dev/null +++ b/responder.js @@ -0,0 +1,62 @@ +var zmq = require('zmq'); +var util = require('util'); +var EventEmitter = require('events').EventEmitter; +var protobufMessage = require('machinetalk-protobuf').message; +var Container = protobufMessage.Container; +var ContainerType = protobufMessage.ContainerType; + +function Responder(routerUri, pubUri) { + this.routerUri = routerUri; + this.pubUri = pubUri; + this.socket = zmq.socket('router'); + this.socket.bindSync(this.routerUri); + this.pubSocket = zmq.socket('pub'); + this.pubSocket.bindSync(this.pubUri); + + this.socket.on('message', this.respond.bind(this)); +} +util.inherits(Responder, EventEmitter); + +Responder.prototype.respond = function() { + var args = Array.apply(null, arguments); + console.log('incoming', args[args.length-1].toString()); + this.pubSocket.send(['msg', args[args.length-1]]); + this.socket.send(args); +}; + +function Publisher(uri) { + this.uri = uri; + this.socket = zmq.socket('pub'); + this.socket.bindSync(this.uri); + + setInterval(this.publish.bind(this), 500); +} +util.inherits(Publisher, EventEmitter); + +Publisher.prototype.publish = function() { + var msg = {type: ContainerType.MT_PING, name: "123456789"}; + var encoded = Container.encode(msg).toBuffer(); + this.socket.send(['topic1', encoded]); + this.socket.send(['topic2', 'test2']); +}; + +function Sender(uri) { + this.uri = uri; + this.socket = zmq.socket('dealer'); + this.socket.connect(this.uri); + + this.socket.on('message', function(msg) { + console.log('received', msg.toString()); + }); + + setInterval(this.sendMessage.bind(this), 500); +} +util.inherits(Sender, EventEmitter); + +Sender.prototype.sendMessage = function() { + this.socket.send('test'); +}; + +var responder = new Responder('tcp://127.0.0.1:12345', 'tcp://127.0.0.1:12346'); +//var sender = new Sender('tcp://127.0.0.1:12345'); +var publisher = new Publisher('tcp://127.0.0.1:12347'); diff --git a/socketmanager.js b/socketmanager.js new file mode 100644 index 000000000..18945f7a3 --- /dev/null +++ b/socketmanager.js @@ -0,0 +1,196 @@ +var zmq = require('zmq'); +var util = require('util'); +var uuid = require('uuid'); +var EventEmitter = require('events').EventEmitter; +var protobufMessage = require('machinetalk-protobuf').message; +var Container = protobufMessage.Container; +var ContainerType = protobufMessage.ContainerType; + +function ZmqConnection(uri, type, uuid) { + this.uri = uri; + this.type = type; + this.socket = zmq.socket(type); + this.timeout = 2000; + + this.socket.connect(this.uri); + this.socket.on('message', this.emit.bind(this, 'message')); + //this.socket.on('message', this.refreshTimeout.bind(this)); + + this.timer = undefined; + //this.refreshTimeout(); + // function (msg) { + // this.emit('message', msg); + // //console.log('test', msg.toString()); + // }); +} +util.inherits(ZmqConnection, EventEmitter); + +// refresh a timeout +ZmqConnection.prototype.refreshTimeout = function() { + clearTimeout(this.timer); + this.timer = setTimeout(this.close.bind(this), this.timeout); +}; + +ZmqConnection.prototype.close = function() { + clearTimeout(this.timer); + this.socket.close(); + this.emit('closed', this); + delete this; +}; + +function ZmqBroker(uri, type) { + this.uri = uri; + var transport = (type === 'dealer' ? 'inproc://' : 'ipc://ipc/'); + this.backendUri = transport + uuid.v4(); + this.type = type; + this.frontend = zmq.socket(type === 'sub' ? 'xsub' : type); + this.backend = zmq.socket(type === 'dealer' ? 'router' : 'xpub'); + this.connections = new Set(); + + this.frontend.connect(this.uri); + this.backend.bindSync(this.backendUri); + this.createBroker(this.frontend, this.backend); +} +util.inherits(ZmqBroker, EventEmitter); + +ZmqBroker.prototype.createConnection = function () { + var connection = new ZmqConnection(this.backendUri, this.type); + connection.on('closed', this.connectionClosed.bind(this)); + this.connections.add(connection); + return connection; +}; + +ZmqBroker.prototype.connectionClosed = function(connection) { + this.connections.delete(connection); + if (this.connections.size === 0) { + this.close(); + } +}; + +ZmqBroker.prototype.createBroker = function(frontend, backend) { + frontend.on('message', function() { + // Note that separate message parts come as function arguments. + var args = Array.apply(null, arguments); + // Pass array of strings/buffers to send multipart messages. + backend.send(args); + }); + + backend.on('message', function() { + var args = Array.apply(null, arguments); + frontend.send(args); + }); +}; + +ZmqBroker.prototype.close = function() { + this.backend.close(); + this.frontend.close(); + this.emit('closed', this.type, this.uri); +}; + +function SocketManager(io) { + this.brokers = {sub: {}, dealer: {}}; // map of brokers + this.io = io; + this.connections = {}; + + io.on('connection', this._handleConnection.bind(this)); +} +util.inherits(SocketManager, EventEmitter); + +SocketManager.prototype._handleConnection = function(socket) { + socket.on('chat message', function(msg) { + socket.broadcast.emit('chat message', msg); + }); + + socket.on('connect socket', this.connectSocket.bind(this, socket)); + socket.on('disconnect socket', this.disconnectSocket.bind(this)); +}; + +SocketManager.prototype.connectSocket = function(socket, msg) { + console.log(msg); + var connection = this.createSocket(msg); + if (connection !== undefined) { + var uuid = msg.uuid; + var type = msg.type; + this.connections[uuid] = connection; + + socket.on('disconnect', this.websocketDisconnected.bind(this, uuid)); + connection.on('message', this.socketMessageReceived.bind(this, socket, type, uuid)); + if (type === 'sub') { + socket.on(uuid + 'subscribe', this.websocketSubscribeReceived.bind(this, connection)); + socket.on(uuid + 'unsubscribe', this.websocketUnsubscribeReceived.bind(this, connection)); + } + else { + socket.on(uuid + 'msg', this.websocketMessageReceived.bind(this, connection)); + } + socket.emit(uuid + 'connected'); + } + else { + socket.emit('error', 'something went wrong'); + } +}; + +SocketManager.prototype.disconnectSocket = function(msg) { + if (msg.uuid !== undefined) { + this.closeSocket(msg.uuid); + } +}; + +SocketManager.prototype.socketMessageReceived = function (socket, type, uuid) { + var args = Array.apply(null, arguments); + args.splice(0, 3); // remove normal params + if (type === 'sub') { + args[0] = args[0].toString(); + } + socket.emit(uuid + 'msg', args); +}; + +SocketManager.prototype.websocketMessageReceived = function(connection, msg) { + connection.socket.send(msg); +}; + +SocketManager.prototype.websocketSubscribeReceived = function(connection, msg) { + connection.socket.subscribe(msg); +}; + +SocketManager.prototype.websocketUnsubscribeReceived = function(connection, msg) { + connection.socket.unsubscribe(msg); +}; + +SocketManager.prototype.websocketDisconnected = function(uuid) { + this.closeSocket(uuid); +}; + +SocketManager.prototype.createSocket = function(msg) { + if ((msg.uri === undefined) + || (msg.type === undefined) + || (msg.uuid == undefined)) { + return undefined; + } + + if ((msg.type !== 'sub') && (msg.type !== 'dealer')) { + return undefined; + } + + var broker = this.brokers[msg.type][msg.uri]; + if (broker === undefined) { + broker = new ZmqBroker(msg.uri, msg.type); + this.brokers[msg.type][msg.uri] = broker; + broker.on('closed', this.brokerClosed.bind(this)); + } + return broker.createConnection(); // TODO use connection +}; + +SocketManager.prototype.closeSocket = function(uuid) { + var connection = this.connections[uuid]; + if (connection !== undefined) { + connection.close(); + delete this.connections[uuid]; + console.log('socket closed'); + } +}; + +SocketManager.prototype.brokerClosed = function(type, uri) { + delete this.brokers[type][uri]; +}; + +module.exports = SocketManager;