<commit_before>(function() {
  'use strict';

  var app     = this.app
    , _       = require('lodash')
    , debug   = require('debug')('signalk-server:interfaces:ws')
    , Primus  = require('primus')
    , primus  = new Primus(app.server, { 
      transformer: 'websockets',
      pathname: '/signalk/stream' })
  ;

  app.connections = {};
  
  debug('Starting Primus/WS interface');

  primus.on('connection', function(spark) {
    debug(spark.id + " connected with params " + JSON.stringify(spark.query));

    var onChange;
    var event;
    if (spark.query.stream === 'delta') {
      event = 'change:delta';
      onChange = function(data) {
        spark.write(data);
      }
    } else {
      event = 'change';
      onChange = function(data) {
        spark.write(app.signalk.retrieve());
      };
      spark.write(app.signalk.retrieve());//output tree after connect
    }

    app.signalk.on(event, function(data) {
      onChange(data);
    });
    spark.onDisconnect = function() {
      app.signalk.removeListener(event, onChange);
    }

    app.connections[spark.id] = spark;
  });

  primus.on('disconnection', function(spark) {
    spark.onDisconnect();
    debug(spark.id + " disconnected");
  });

}).call(global);<commit_msg>Add subscription protocol Add a way to subscribe to different SignalK messages by specifying stream=subscribe in the query string and then by sending a JSON command message the consumer can specify which SignalK paths it is interested. Current implementatin works only with deltas and ignores context, just considers the paths in the update message. See also https://github.com/SignalK/maptracker/blob/5cd3ba626188563640329253d6792aeee41a46fc/index.html#L158<commit_after>(function() {
  'use strict';

  var app     = this.app
    , _       = require('lodash')
    , debug   = require('debug')('signalk-server:interfaces:ws')
    , Primus  = require('primus')
    , primus  = new Primus(app.server, { 
      transformer: 'websockets',
      pathname: '/signalk/stream' })
  ;

  app.connections = {};
  
  debug('Starting Primus/WS interface');

  primus.on('connection', function(spark) {
    debug(spark.id + " connected with params " + JSON.stringify(spark.query));

    var onChange;
    var event;
    if (spark.query.stream === 'delta') {
      event = 'change:delta';
      onChange = function(data) {
        spark.write(data);
      };
    } else if (spark.query.stream === 'subscribe') {
      debug('stream:subscribe');
      spark.paths = [];
      event = 'change:delta';
      onChange = function (delta) {
        var doSend = delta.updates.some(function (update) {
          return update.values.some(function (value) {
            return spark.paths.some(function (path) {
              return path === value.path;
            });
          });
        });
        if (doSend) {
          spark.write(delta);
        }
      };
      spark.on('data', function (msg) {
        debug("<" + JSON.stringify(msg));
        if (msg.command === 'subscribe') {
          spark.paths = spark.paths.concat(msg.paths);
        }
      });
    } else {
      event = 'change';
      onChange = function(data) {
        spark.write(app.signalk.retrieve());
      };
      spark.write(app.signalk.retrieve());//output tree after connect
    }

    app.signalk.on(event, function(data) {
      onChange(data);
    });
    spark.onDisconnect = function() {
      app.signalk.removeListener(event, onChange);
    }

    app.connections[spark.id] = spark;
  });

  primus.on('disconnection', function(spark) {
    spark.onDisconnect();
    debug(spark.id + " disconnected");
  });

}).call(global);
{'fraction_non_alphanumeric': 0.14767123287671233, 'fraction_numerical': 0.008493150684931507, 'mean_word_length': 2.6400797607178466, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5962708', 'n_tokens_mistral': 1152, 'n_tokens_neox': 1097, 'n_words': 224}