artemis/lib/nmts.js at master · ENRG/artemis · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
var EventEmitter  = require('events').EventEmitter;
var db            = require('db');
var utils         = require('utils');
var NmtStream     = require('p4441-stream').LeqStream;

module.exports = Object.create( utils.extend( {},

EventEmitter.prototype, {

  streams:  {}

, isRunning: false

, init: function( options ){
    this.options = options ? utils.clone( options ) : {};

    utils.defaults( this.options, {
      loopInterval: 1000
    });

    return this;
  }

, start: function( callback ){
    this.isRunning = true;
    this.loop( callback );

    return this;
  }

, stop: function(){
    this.isRunning = false;
    return this;
  }

, loop: function( callback ){
    if ( !this.isRunning ) return;

    var this_ = this;

    db.nmts.find( {}, function( error, results ){
      if ( callback && error ) return callback( error );
      if ( error ) throw error;

      results.forEach( function( nmt ){
        if ( nmt.is_active ){
          if ( !(nmt.id in this_.streams) ) this_.register( nmt );
          else {
            // Must have had a pending stream request
            this_.streams[ nmt.id ].setNmt( nmt );
          }
        } else {
          if ( nmt.id in this_.streams ) this_.unregister( nmt.id );
        }
      });
    });

    return setTimeout( function(){
      if ( callback ) callback();
      this_.loop();
    }, this.options.loopInterval );
  }

, register: function( nmt, callback ){
    callback = callback || utils.noop;

    var this_ = this;
    var stream = this.streams[ nmt.id ] = new NmtStream( nmt );

    stream.connect();

    stream.on( 'close', function(){
      delete this_.streams[ nmt.id ];
    });

    return stream;
  }

, unregister: function( id ){
    if ( !this.streams[ id ] ) return this;

    this.streams[ id ].destroy();

    return this;
  }

, getStreamById: function( id ){
    if ( !(id in this.streams) ){
      this.streams[ id ] = new NmtStream();
    }

    return this.streams[ id ];
  }
}));