CODEWITHSUNDEEP
node.jssocketssails.jsrxjs
TheUnreal
TheUnreal

Reputation: 24472

NodeJS - Response stream

I built a simple API endpoint with NodeJS using Sails.js.

When someone access my API endpoint, the server starts to wait for data and whenever a new data appears, he broadcasts it using sockets. Each client should receive his own stream of data based on his user input.

var Cap = require('cap').Cap;

collect: function (req, res) {

var iface = req.param("ip");

var c = new Cap(),
device = Cap.findDevice(ip);

    c.on('data', function(myData) {
        sails.sockets.blast('message', {"host": myData});
    });
});

The response do not complete (I never send a res.json() - what actually happens is that the browser keep loading - but the above functionality works).

2 Problems:

How I can create a stream/service-like API endpoint with Sails.js that will emit new data to each user based on his input?

My goal is to be able to subscribe / unsubscribe to this API endpoint from each client.

Upvotes: 2

Views: 434

Answers (1)

ContinuousLoad
ContinuousLoad

Reputation: 4912

Revised Answer

Let's assume your API endpoint is defined in config/routes.js like this:

...
'get     /collect': 'SomeController.collectSubscribe',
'delete  /collect': 'SomeController.collectUnsubscribe',

Since each Cap instance is tied to one device, we need one instance for each subscription. Instead of using the sails join/leave methods, we keep track of Cap instances in memory and just broadcast to the request socket's id. This works because Sails sockets are subscribed to their own ids by default.

In api/controllers/SomeController.js:

// In order for the `Cap` instances to persist after `collectSubscribe` finishes, we store them all in an Object, associated with which socket the were created for.
var caps = {/* req.socket.id: <instance of Cap>, */};

module.exports = {

...

  collectSubscribe: function(req, res) {
    if (!res.isSocket) return res.badRequest("I need a websocket! Help!");
    if (!!caps[req.socket.id]) return res.badRequest("Dude, you are already subscribed.");

    caps[req.socket.id] = new Cap();
    var c = caps[req.socket.id]; // remember that `c` is a reference to our new `Cap`, not a copy.
    var device = c.findDevice(req.param('ip'));

    c.open(device, ...);
    c.on('data', function(myData) {
      sails.sockets.broadcast(req.socket.id, 'message', {host: myData});
    });

    return res.ok();
  },

  collectUnsubscribe: function(req, res) {
    if (!res.isSocket) return res.badRequest("I need a websocket! Help!");
    if (!caps[req.socket.id]) return res.badRequest("I can't unsubscribe you unless you actually subscribe first.");

    caps[req.socket.id].removeAllListeners('data');
    delete caps[req.socket.id]; 

    return res.ok();
  }
}

Basically, it goes like this: when a browser request triggers collectSubscribe, a new Cap instance listens to the provided IP. When the browser triggers collectUnsubscribe, the server retreives that Cap instance, tells it to stop listening, and then deletes it.

Production Considerations: please be aware that the list of Caps is NOT PERSISTENT (since it is stored in memory and not a DB)! So if your server is turned off and rebooted (due to lightning storm, etc), the list will be cleared, but considering that all websocket connections will be dropped anyway, I don't see any need to worry about this.

Old Answer, Kept for Reference

You can use sails.sockets.join(req, room) and sails.sockets.leave(req, room) to manage socket rooms. Essentially you have a room called "collect", and only sockets joined in that room will receive a sails.sockets.broadcast(room, eventName, data).

More info on how to user sails.sockets here.

In api/controllers/SomeController.js:

collectSubscribe: function(req, res) {
  if (!res.isSocket) return res.badRequest();

  sails.sockets.join(req, 'collect');
  return res.ok();

},

collectUnsubscribe: function(req, res) {
  if (!res.isSocket) return res.badRequest();

  sails.sockets.leave(req, 'collect');
  return res.ok();
}

Finally, we need to tell the server to broadcast messages to our 'collect' room. Note that this only need to happen once, so you can do this in a file under the config/ directory.

For this example, I'll put this in config/sockets.js

module.exports = {
  // ...
};


c.on('data', function(myData) {
  var eventName = 'message';
  var data = {host: myData};
  sails.sockets.broadcast('collect', eventName, data);
});

I am assuming that c is accessible here; If not, you could define it as sails.c = ... to make it globally accessible.

Upvotes: 1

Related Questions