I'm coding a browser notification using rabbitMQ and socket.io. My configuration is working fine except for one case.
When i login to my system with a user it creates a notification-UID-userid queue (For now the queueName is sent by query oaraeter, i will implement more sofisticated method as soon as i will solve the problem)
If i login with another user on another browser it creates another queue notification-UID-seconduserid.
If i logout one of the user the queue will disappear (as it's not durable).
The problem is that when i refresh or load another page on the other session it recreates the second queue even if the paramater queuename isn't sent.
server.js
var amqp = require('amqp');
var app = require('express')();
var http = require('http').Server(app);
var io = require('socket.io')(http);
var rabbitMqConnection = null;
var _queue = null;
var _consumerTag = null;
io.use(function (socket, next) {
var handshakeData = socket.handshake;
// Here i will implement token verification
console.log(socket.handshake.query.queueName);
next();
});
// Gets the connection event form client
io.sockets.on('connection', function (socket) {
var queueName = socket.handshake.query.queueName;
console.log("Socket Connected");
// Connects to rabbiMq
rabbitMqConnection = amqp.createConnection({host: 'localhost', reconnect: false});
// Update our stored tag when it changes
rabbitMqConnection.on('tag.change', function (event) {
if (_consumerTag === event.oldConsumerTag) {
_consumerTag = event.consumerTag;
// Consider unsubscribing from the old tag just in case it lingers
_queue.unsubscribe(event.oldConsumerTag);
}
});
// Listen for ready event
rabbitMqConnection.on('ready', function () {
console.log('Connected to rabbitMQ');
// Listen to the queue
rabbitMqConnection.queue(queueName, {
closeChannelOnUnsubscribe: true,
durable: false,
autoClose: true
},
function (queue) {
console.log('Connected to ' + queueName);
_queue = queue;
// Bind to the exchange
queue.bind('users.direct', queueName);
queue.subscribe({ack: false, prefetchCount: 1}, function (message, headers, deliveryInfo, ack) {
console.log("Received a message from route " + deliveryInfo.routingKey);
socket.emit('notification', message);
//ack.acknowledge();
}).addCallback(function (res) {
// Hold on to the consumer tag so we can unsubscribe later
_consumerTag = res.consumerTag;
});
});
});
// Listen for disconnection
socket.on('disconnect', function () {
_queue.unsubscribe(_consumerTag);
rabbitMqConnection.disconnect();
console.log("Socket Disconnected");
});
});
http.listen(8080);
client.js
var io = require('socket.io-client');
$(document).ready(function () {
var socket = io('http://myserver.it:8080/', {
query: { queueName: 'notification-UID-' + UID},
'sync disconnect on unload': true,
});
socket.on('notification', function (data) {
console.log(data);
});
})
Any idea?
So i solved my problem, it was a variable scope problem that messed things. Let me explain what i'm doing, maybe it could be useful to someone.
Basically i'm trying to create a browser notification system, it means that my application publish (producer side) to an exchange a notification object that contains some info such as subject, link and message.
The exchange is a fanout (users.notification.fanout) that have two binded exchanges: users.direct (direct type) and users.notification.store (fanout type).
When the producer publishes a notification, it does to the users.notification.fanout with routing key "notification-UID-userid" (where userid is the real user id.
The notification object goes to both, users.direct and users.notification.store The last one has a consumer that writes the notification to the DB in case the user is not logged, the first one publishes the notification to the browser.
So how the browser consumer works?
I used the classical combination of socket.io, node server, and amqplib.
Every time a user login, the socket.io creates a queue with name and routing key notification-UID-userid and binds it to users.direct exchange.
In the meanwhile i've added https to my server so things changed a little bit from first version.
You can read comments to know what it does.
So my server.js is
var amqp = require('amqp');
var fs = require('fs');
var app = require('express')();
// Https server, certificates and private key added
var https = require('https').Server({
key: fs.readFileSync('/home/www/site/privkey.pem'),
cert: fs.readFileSync('/home/www/site/fullchain.pem')},app);
var io = require('socket.io')(https);
// Used to verify if token is valid
// If not it will discard connection
io.use(function (socket, next) {
var handshakeData = socket.handshake;
// Here i will implement token verification
console.log("Check this token: " + handshakeData.query.token);
next();
});
// Gets the connection event from client
io.sockets.on('connection', function (socket) {
// Connection log
console.log("Socket Connected with ID: " + socket.id);
// THIS WAS THE PROBLEM
// Local variables for connections
// Former i've put these variables outside the connection so at
// every client they were "overridden".
// RabbitMq Connection (Just for current client)
var _rabbitMqConnection = null;
// Queue (just for current client)
var _queue = null;
// Consumer tag (just for current client)
var _consumerTag = null;
// Queue name and routing key for current user
var queueName = socket.handshake.query.queueName;
// Connects to rabbiMq with default data to localhost guest guest
_rabbitMqConnection = amqp.createConnection();
// Connection ready
_rabbitMqConnection.on('ready', function () {
// Connection log
console.log('#' + socket.id + ' - Connected to RabbitMQ');
// Creates the queue (default is transient and autodelete)
// https://www.npmjs.com/package/amqp#connectionqueuename-options-opencallback
_rabbitMqConnection.queue(queueName, function (queue) {
// Connection log
console.log('#' + socket.id + ' - Connected to ' + queue.name + ' queue');
// Stores local queue
_queue = queue;
// Bind to the exchange (default)
queue.bind('users.direct', queueName, function () {
// Binding log
console.log('#' + socket.id + ' - Binded to users.direct exchange');
// Consumer definition
queue.subscribe({ack: false}, function (message, headers, deliveryInfo, messageObject) {
// Message log
console.log('#' + socket.id + ' - Received a message from route ' + deliveryInfo.routingKey);
// Emit the message to the client
socket.emit('notification', message);
}).addCallback(function (res) {
// Hold on to the consumer tag so we can unsubscribe later
_consumerTag = res.consumerTag;
// Consumer tag log
console.log('#' + socket.id + ' - Consumer ' + _consumerTag + ' created');
})
});
});
});
// Update our stored tag when it changes
_rabbitMqConnection.on('tag.change', function (event) {
if (_consumerTag === event.oldConsumerTag) {
_consumerTag = event.consumerTag;
// Unsubscribe from the old tag just in case it lingers
_queue.unsubscribe(event.oldConsumerTag);
}
});
// Listen for disconnection
socket.on('disconnect', function () {
_queue.unsubscribe(_consumerTag);
_rabbitMqConnection.disconnect();
console.log('#' + socket.id + ' - Socket Disconnected');
});
});
https.listen(8080);
Then my client.js
var io = require('socket.io-client');
$(document).ready(function () {
var socket = io('https://myserver.com:8080/', {
secure: true, // for ssl connections
query: { queueName: 'notification-UID-' + UID, token: JWTToken}, // params sent to server, JWTToken for authentication
'sync disconnect on unload': true // Every time the client unload, socket disconnects
});
socket.on('notification', function (data) {
// Do what you want with your data
console.log(data);
});
})