RemoveListener removing all listeners initiated after 'this' listener

Question

I have a socket.io server that has a custom method 'looper.output' inside each socket.io connection. Whenever I close a socket.io connection I launch another custom method 'looper.disconnect' that is supposed to remove the 'output' listener for the client that closed. My issue is that whenever a client closes, looper.disconnect not only removes the listener from that client but it also removes the listener from every client that was initiated after the client that closed.

For example, if I have 4 clients A,B,C and D that also connected to the server in that order and client C closed; looper.disconnect will remove the listeners for both C and D but leave A and B alone.

What do I need to change to make looper.disconnect only remove the listener for the client that called it? In the example above I would only want the listener for client C removed.

var express = require('express');
var http = require('http');
var spawn = require('child_process').spawn;
var util = require('util');
var fs = require('fs');
var EventEmitter = require('events').EventEmitter;
var sys = require('sys');

var app     = express(),
    server  = http.createServer(app),
    io      = require('socket.io').listen(server);

function Looper(req)  {
    this.req = req;
    EventEmitter.call(this);
}
sys.inherits(Looper, EventEmitter);

Looper.prototype.run = function() {
    var self = this;
    var cmd = spawn('./flow',[this.req]); // <-- script that outputs req every second
    cmd.stdout.setEncoding('utf8');
    cmd.stdout.on('data', function(data) {
        self.emit('output',data);
    });
}

Looper.prototype.output = function(callback) {
    this.output.outCallback = function(mydata) {
        return callback(mydata.trim());
    }
    this.on('output', this.output.outCallback);
}

Looper.prototype.disconnect = function() {
    this.removeListener('output',this.output.outCallback);
}

var looper = new Looper('blah');
looper.run();

app.use(express.static(__dirname + '/public'));

app.get('/', function(req, res) {
    res.send(
        "<script src='/socket.io/socket.io.js'></script>\n"+
        "<script>\n"+
        "\tvar socket=io.connect('http://127.0.0.1:3000');\n"+
        "\tsocket.on('stream', function(data) {\n"+
        "\t\tconsole.log(data);\n"+
        "\t});\n"+
        "</script>\n"
    );
});
server.listen(3000);

io.sockets.on('connection', function(webSocket) {
    looper.output(function(res) {
        webSocket.emit('stream',res+":"+webSocket.id);
    });

    webSocket.on('disconnect', function() {
        looper.disconnect();
    });
});

Answer 1

I don't really understand why it fails that way but your main problem is that you are saving the reference of all the functions in the same place.

Another option is to have a callback function reference for each client like this:

var express = require('express');
var http = require('http');
var spawn = require('child_process').spawn;
var util = require('util');
var fs = require('fs');
var EventEmitter = require('events').EventEmitter;
var sys = require('sys');

var app     = express(),
    server  = http.createServer(app),
    io      = require('socket.io').listen(server);

function Looper(req)  {
    this.req = req;
    EventEmitter.call(this);
}
sys.inherits(Looper, EventEmitter);

Looper.prototype.run = function() {
    var self = this;
    var cmd = spawn('./flow',[this.req]); // <-- script that outputs req every second
    cmd.stdout.setEncoding('utf8');
    cmd.stdout.on('data', function(data) {
        self.emit('output',data);
    });
}

Looper.prototype.output = function(callback) {
    this.on('output', callback);
}

Looper.prototype.disconnect = function(callback) {
    this.removeListener('output', callback);
}

var looper = new Looper('blah');
looper.run();

app.use(express.static(__dirname + '/public'));

app.get('/', function(req, res) {
    res.send(
        "<script src='/socket.io/socket.io.js'></script>\n"+
        "<script>\n"+
        "\tvar socket=io.connect('http://127.0.0.1:3000');\n"+
        "\tsocket.on('stream', function(data) {\n"+
        "\t\tconsole.log(data);\n"+
        "\t});\n"+
        "</script>\n"
    );
});
server.listen(3000);

io.sockets.on('connection', function(webSocket) {
    var callback = function(res) {
        webSocket.emit('stream',res.trim()+":"+webSocket.id);
    }

    looper.output(callback);

    webSocket.on('disconnect', function() {
        looper.disconnect(callback);
    });
});

Answer 2

You can leave the listener always on and in the listener function send data to all callbacks added to the looper object.

The callbacks are identified for each client by the session_id.

var express = require('express');
var http = require('http');
var spawn = require('child_process').spawn;
var util = require('util');
var fs = require('fs');
var EventEmitter = require('events').EventEmitter;
var sys = require('sys');

var app     = express(),
    server  = http.createServer(app),
    io      = require('socket.io').listen(server);

function Looper(req)  {
    this.req = req;
    Looper.prototype.outCallbacks = {};
    EventEmitter.call(this);
}
sys.inherits(Looper, EventEmitter);

Looper.prototype.callAllCallbacks = function(data) {
    for (var callback in this.outCallbacks) {
    if (this.outCallbacks.hasOwnProperty(callback)) {
        this.outCallbacks[callback](data);
    }
    }
}

Looper.prototype.run = function() {
    var self = this;
    var cmd = spawn('./flow',[this.req]); // <-- script that outputs req every second
    cmd.stdout.setEncoding('utf8');
    cmd.stdout.on('data', function(data) {
    self.emit('output',data);
    });

    this.on('output', this.callAllCallbacks);
}

Looper.prototype.addCallback = function(session_id, callback) {
    this.outCallbacks[session_id] = function(mydata) {
    return callback(mydata.trim());
    }
}

Looper.prototype.disconnect = function(session_id) {
    delete this.outCallbacks[session_id];
}

var looper = new Looper('blah');
looper.run();

app.use(express.static(__dirname + '/public'));

app.get('/', function(req, res) {
    res.send(
        "<script src='/socket.io/socket.io.js'></script>\n"+
        "<script>\n"+
        "\tvar socket=io.connect('http://127.0.0.1:3000');\n"+
        "\tsocket.on('stream', function(data) {\n"+
        "\t\tconsole.log(data);\n"+
        "\t});\n"+
        "</script>\n"
    );
});
server.listen(3000);

io.sockets.on('connection', function(webSocket) {
    var session_id = webSocket.id;

    looper.addCallback(session_id, function(res) {
    webSocket.emit('stream',res+":"+webSocket.id);
    });

    webSocket.on('disconnect', function() {
    looper.disconnect(session_id);
    });
});