Subscribing to every stream of every device of every peer via a Zetta client.

client.js

var zetta = require('zetta-client');

var server = zetta().connect('http://localhost:3003');

var allQuery = server
  .from()
  .where('where type is not missing');

server.observe(allQuery, function(device) {
  var hubName = device._data.links
    .filter(function(link) {
      return link.rel.indexOf('up') !== -1;
    })[0].title;

  device._data.links
    .filter(function(link) {
      return link.rel.indexOf('http://rels.zettajs.io/object-stream') !== -1;
    })
    .map(function(link) {
      return device.createReadStream(link.title);
    })
    .forEach(function(stream) {
      stream.on('data', function(message) {
        var datum = {
          hub: hubName,
          device: {
            id: device.id,
            type: device.type
          },
          timestamp: message.timestamp,
          value: message.data
        };

        // do something with the data
        console.log(datum);
      });
    });
});