path: root/testing/xpcshell/node-http2/test/stream.js

var expect = require('chai').expect;
var util = require('./util');

var stream = require('../lib/protocol/stream');
var Stream = stream.Stream;

function createStream() {
  var stream = new Stream(util.log, null);
  stream.upstream._window = Infinity;
  return stream;
}

// Execute a list of commands and assertions
var recorded_events = ['state', 'error', 'window_update', 'headers', 'promise'];
function execute_sequence(stream, sequence, done) {
  if (!done) {
    done = sequence;
    sequence = stream;
    stream = createStream();
  }

  var outgoing_frames = [];

  var emit = stream.emit, events = [];
  stream.emit = function(name) {
    if (recorded_events.indexOf(name) !== -1) {
      events.push({ name: name, data: Array.prototype.slice.call(arguments, 1) });
    }
    return emit.apply(this, arguments);
  };

  var commands = [], checks = [];
  sequence.forEach(function(step) {
    if ('method' in step || 'incoming' in step || 'outgoing' in step || 'wait' in step || 'set_state' in step) {
      commands.push(step);
    }

    if ('outgoing' in step || 'event' in step || 'active' in step) {
      checks.push(step);
    }
  });

  var activeCount = 0;
  function count_change(change) {
    activeCount += change;
  }

  function execute(callback) {
    var command = commands.shift();
    if (command) {
      if ('method' in command) {
        var value = stream[command.method.name].apply(stream, command.method.arguments);
        if (command.method.ret) {
          command.method.ret(value);
        }
        execute(callback);
      } else if ('incoming' in command) {
        command.incoming.count_change = count_change;
        stream.upstream.write(command.incoming);
        execute(callback);
      } else if ('outgoing' in command) {
        outgoing_frames.push(stream.upstream.read());
        execute(callback);
      } else if ('set_state' in command) {
        stream.state = command.set_state;
        execute(callback);
      } else if ('wait' in command) {
        setTimeout(execute.bind(null, callback), command.wait);
      } else {
        throw new Error('Invalid command', command);
      }
    } else {
      setTimeout(callback, 5);
    }
  }

  function check() {
    checks.forEach(function(check) {
      if ('outgoing' in check) {
        var frame = outgoing_frames.shift();
        for (var key in check.outgoing) {
          expect(frame).to.have.property(key).that.deep.equals(check.outgoing[key]);
        }
        count_change(frame.count_change);
      } else if ('event' in check) {
        var event = events.shift();
        expect(event.name).to.be.equal(check.event.name);
        check.event.data.forEach(function(data, index) {
          expect(event.data[index]).to.deep.equal(data);
        });
      } else if ('active' in check) {
        expect(activeCount).to.be.equal(check.active);
      } else {
        throw new Error('Invalid check', check);
      }
    });
    done();
  }

  setImmediate(execute.bind(null, check));
}

var example_frames = [
  { type: 'PRIORITY', flags: {}, priority: 1 },
  { type: 'WINDOW_UPDATE', flags: {}, settings: {} },
  { type: 'RST_STREAM', flags: {}, error: 'CANCEL' },
  { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
  { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
  { type: 'PUSH_PROMISE', flags: {}, headers: {}, promised_stream: new Stream(util.log, null) }
];

var invalid_incoming_frames = {
  IDLE: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
    { type: 'RST_STREAM', flags: {}, error: 'CANCEL' }
  ],
  RESERVED_LOCAL: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
  ],
  RESERVED_REMOTE: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
  ],
  OPEN: [
  ],
  HALF_CLOSED_LOCAL: [
  ],
  HALF_CLOSED_REMOTE: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} }
  ]
};

var invalid_outgoing_frames = {
  IDLE: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} }
  ],
  RESERVED_LOCAL: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
  ],
  RESERVED_REMOTE: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} },
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} }
  ],
  OPEN: [
  ],
  HALF_CLOSED_LOCAL: [
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
    { type: 'PUSH_PROMISE', flags: {}, headers: {} }
  ],
  HALF_CLOSED_REMOTE: [
  ],
  CLOSED: [
    { type: 'WINDOW_UPDATE', flags: {}, settings: {} },
    { type: 'HEADERS', flags: {}, headers: {}, priority: undefined },
    { type: 'DATA', flags: {}, data: Buffer.alloc(5) },
    { type: 'PUSH_PROMISE', flags: {}, headers: {}, promised_stream: new Stream(util.log, null) }
  ]
};

describe('stream.js', function() {
  describe('Stream class', function() {
    describe('._transition(sending, frame) method', function() {
      it('should emit error, and answer RST_STREAM for invalid incoming frames', function() {
        Object.keys(invalid_incoming_frames).forEach(function(state) {
          invalid_incoming_frames[state].forEach(function(invalid_frame) {
            var stream = createStream();
            var connectionErrorHappened = false;
            stream.state = state;
            stream.once('connectionError', function() { connectionErrorHappened = true; });
            stream._transition(false, invalid_frame);
            expect(connectionErrorHappened);
          });
        });

        // CLOSED state as a result of incoming END_STREAM (or RST_STREAM)
        var stream = createStream();
        stream.headers({});
        stream.end();
        stream.upstream.write({ type: 'HEADERS', headers:{}, flags: { END_STREAM: true }, count_change: util.noop });
        example_frames.slice(2).forEach(function(invalid_frame) {
          invalid_frame.count_change = util.noop;
          expect(stream._transition.bind(stream, false, invalid_frame)).to.throw('Uncaught, unspecified "error" event.');
        });

        // CLOSED state as a result of outgoing END_STREAM
        stream = createStream();
        stream.upstream.write({ type: 'HEADERS', headers:{}, flags: { END_STREAM: true }, count_change: util.noop });
        stream.headers({});
        stream.end();
        example_frames.slice(3).forEach(function(invalid_frame) {
          invalid_frame.count_change = util.noop;
          expect(stream._transition.bind(stream, false, invalid_frame)).to.throw('Uncaught, unspecified "error" event.');
        });
      });
      it('should throw exception for invalid outgoing frames', function() {
        Object.keys(invalid_outgoing_frames).forEach(function(state) {
          invalid_outgoing_frames[state].forEach(function(invalid_frame) {
            var stream = createStream();
            stream.state = state;
            expect(stream._transition.bind(stream, true, invalid_frame)).to.throw(Error);
          });
        });
      });
      it('should close the stream when there\'s an incoming or outgoing RST_STREAM', function() {
        [
          'RESERVED_LOCAL',
          'RESERVED_REMOTE',
          'OPEN',
          'HALF_CLOSED_LOCAL',
          'HALF_CLOSED_REMOTE'
        ].forEach(function(state) {
            [true, false].forEach(function(sending) {
              var stream = createStream();
              stream.state = state;
              stream._transition(sending, { type: 'RST_STREAM', flags: {} });
              expect(stream.state).to.be.equal('CLOSED');
            });
          });
      });
      it('should ignore any incoming frame after sending reset', function() {
        var stream = createStream();
        stream.reset();
        example_frames.forEach(stream._transition.bind(stream, false));
      });
      it('should ignore certain incoming frames after closing the stream with END_STREAM', function() {
        var stream = createStream();
        stream.upstream.write({ type: 'HEADERS', flags: { END_STREAM: true }, headers:{} });
        stream.headers({});
        stream.end();
        example_frames.slice(0,3).forEach(function(frame) {
          frame.count_change = util.noop;
          stream._transition(false, frame);
        });
      });
    });
  });
  describe('test scenario', function() {
    describe('sending request', function() {
      it('should trigger the appropriate state transitions and outgoing frames', function(done) {
        execute_sequence([
          { method  : { name: 'headers', arguments: [{ ':path': '/' }] } },
          { outgoing: { type: 'HEADERS', flags: { }, headers: { ':path': '/' } } },
          { event   : { name: 'state', data: ['OPEN'] } },

          { wait    : 5 },
          { method  : { name: 'end', arguments: [] } },
          { event   : { name: 'state', data: ['HALF_CLOSED_LOCAL'] } },
          { outgoing: { type: 'DATA', flags: { END_STREAM: true  }, data: Buffer.alloc(0) } },

          { wait    : 10 },
          { incoming: { type: 'HEADERS', flags: { }, headers: { ':status': 200 } } },
          { incoming: { type: 'DATA'   , flags: { END_STREAM: true  }, data: Buffer.alloc(5) } },
          { event   : { name: 'headers', data: [{ ':status': 200 }] } },
          { event   : { name: 'state', data: ['CLOSED'] } },

          { active  : 0 }
        ], done);
      });
    });
    describe('answering request', function() {
      it('should trigger the appropriate state transitions and outgoing frames', function(done) {
        var payload = Buffer.alloc(5);
        execute_sequence([
          { incoming: { type: 'HEADERS', flags: { }, headers: { ':path': '/' } } },
          { event   : { name: 'state', data: ['OPEN'] } },
          { event   : { name: 'headers', data: [{ ':path': '/' }] } },

          { wait    : 5 },
          { incoming: { type: 'DATA', flags: { }, data: Buffer.alloc(5) } },
          { incoming: { type: 'DATA', flags: { END_STREAM: true  }, data: Buffer.alloc(10) } },
          { event   : { name: 'state', data: ['HALF_CLOSED_REMOTE'] } },

          { wait    : 5 },
          { method  : { name: 'headers', arguments: [{ ':status': 200 }] } },
          { outgoing: { type: 'HEADERS', flags: { }, headers: { ':status': 200 } } },

          { wait    : 5 },
          { method  : { name: 'end', arguments: [payload] } },
          { outgoing: { type: 'DATA', flags: { END_STREAM: true  }, data: payload } },
          { event   : { name: 'state', data: ['CLOSED'] } },

          { active  : 0 }
        ], done);
      });
    });
    describe('sending push stream', function() {
      it('should trigger the appropriate state transitions and outgoing frames', function(done) {
        var payload = Buffer.alloc(5);
        var pushStream;

        execute_sequence([
          // receiving request
          { incoming: { type: 'HEADERS', flags: { END_STREAM: true }, headers: { ':path': '/' } } },
          { event   : { name: 'state', data: ['OPEN'] } },
          { event   : { name: 'state', data: ['HALF_CLOSED_REMOTE'] } },
          { event   : { name: 'headers', data: [{ ':path': '/' }] } },

          // sending response headers
          { wait    : 5 },
          { method  : { name: 'headers', arguments: [{ ':status': '200' }] } },
          { outgoing: { type: 'HEADERS', flags: {  }, headers: { ':status': '200' } } },

          // sending push promise
          { method  : { name: 'promise', arguments: [{ ':path': '/' }], ret: function(str) { pushStream = str; } } },
          { outgoing: { type: 'PUSH_PROMISE', flags: { }, headers: { ':path': '/' } } },

          // sending response data
          { method  : { name: 'end', arguments: [payload] } },
          { outgoing: { type: 'DATA', flags: { END_STREAM: true  }, data: payload } },
          { event   : { name: 'state', data: ['CLOSED'] } },

          { active  : 0 }
        ], function() {
          // initial state of the promised stream
          expect(pushStream.state).to.equal('RESERVED_LOCAL');

          execute_sequence(pushStream, [
            // push headers
            { wait    : 5 },
            { method  : { name: 'headers', arguments: [{ ':status': '200' }] } },
            { outgoing: { type: 'HEADERS', flags: { }, headers: { ':status': '200' } } },
            { event   : { name: 'state', data: ['HALF_CLOSED_REMOTE'] } },

            // push data
            { method  : { name: 'end', arguments: [payload] } },
            { outgoing: { type: 'DATA', flags: { END_STREAM: true  }, data: payload } },
            { event   : { name: 'state', data: ['CLOSED'] } },

            { active  : 1 }
          ], done);
        });
      });
    });
    describe('receiving push stream', function() {
      it('should trigger the appropriate state transitions and outgoing frames', function(done) {
        var payload = Buffer.alloc(5);
        var original_stream = createStream();
        var promised_stream = createStream();

        done = util.callNTimes(2, done);

        execute_sequence(original_stream, [
          // sending request headers
          { method  : { name: 'headers', arguments: [{ ':path': '/' }] } },
          { method  : { name: 'end', arguments: [] } },
          { outgoing: { type: 'HEADERS', flags: { END_STREAM: true  }, headers: { ':path': '/' } } },
          { event   : { name: 'state', data: ['OPEN'] } },
          { event   : { name: 'state', data: ['HALF_CLOSED_LOCAL'] } },

          // receiving response headers
          { wait    : 10 },
          { incoming: { type: 'HEADERS', flags: { }, headers: { ':status': 200 } } },
          { event   : { name: 'headers', data: [{ ':status': 200 }] } },

          // receiving push promise
          { incoming: { type: 'PUSH_PROMISE', flags: { }, headers: { ':path': '/2.html' }, promised_stream: promised_stream } },
          { event   : { name: 'promise', data: [promised_stream, { ':path': '/2.html' }] } },

          // receiving response data
          { incoming: { type: 'DATA'   , flags: { END_STREAM: true  }, data: payload } },
          { event   : { name: 'state', data: ['CLOSED'] } },

          { active  : 0 }
        ], done);

        execute_sequence(promised_stream, [
          // initial state of the promised stream
          { event   : { name: 'state', data: ['RESERVED_REMOTE'] } },

          // push headers
          { wait    : 10 },
          { incoming: { type: 'HEADERS', flags: { END_STREAM: false }, headers: { ':status': 200 } } },
          { event   : { name: 'state', data: ['HALF_CLOSED_LOCAL'] } },
          { event   : { name: 'headers', data: [{ ':status': 200 }] } },

          // push data
          { incoming: { type: 'DATA', flags: { END_STREAM: true  }, data: payload } },
          { event   : { name: 'state', data: ['CLOSED'] } },

          { active  : 0 }
        ], done);
      });
    });
  });

  describe('bunyan formatter', function() {
    describe('`s`', function() {
      var format = stream.serializers.s;
      it('should assign a unique ID to each frame', function() {
        var stream1 = createStream();
        var stream2 = createStream();
        expect(format(stream1)).to.be.equal(format(stream1));
        expect(format(stream2)).to.be.equal(format(stream2));
        expect(format(stream1)).to.not.be.equal(format(stream2));
      });
    });
  });
});