Blame view

node_modules/websocket-extensions/lib/pipeline/cell.js 1.44 KB
aaac7fed   liuqimichale   add
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
  'use strict';
  
  var Functor = require('./functor'),
      Pledge  = require('./pledge');
  
  var Cell = function(tuple) {
    this._ext     = tuple[0];
    this._session = tuple[1];
  
    this._functors = {
      incoming: new Functor(this._session, 'processIncomingMessage'),
      outgoing: new Functor(this._session, 'processOutgoingMessage')
    };
  };
  
  Cell.prototype.pending = function(direction) {
    var functor = this._functors[direction];
    if (!functor._stopped) functor.pending += 1;
  };
  
  Cell.prototype.incoming = function(error, message, callback, context) {
    this._exec('incoming', error, message, callback, context);
  };
  
  Cell.prototype.outgoing = function(error, message, callback, context) {
    this._exec('outgoing', error, message, callback, context);
  };
  
  Cell.prototype.close = function() {
    this._closed = this._closed || new Pledge();
    this._doClose();
    return this._closed;
  };
  
  Cell.prototype._exec = function(direction, error, message, callback, context) {
    this._functors[direction].call(error, message, function(err, msg) {
      if (err) err.message = this._ext.name + ': ' + err.message;
      callback.call(context, err, msg);
      this._doClose();
    }, this);
  };
  
  Cell.prototype._doClose = function() {
    var fin  = this._functors.incoming,
        fout = this._functors.outgoing;
  
    if (!this._closed || fin.pending + fout.pending !== 0) return;
    if (this._session) this._session.close();
    this._session = null;
    this._closed.done();
  };
  
  module.exports = Cell;