'use strict';

var _getPrototypeOf = require('babel-runtime/core-js/object/get-prototype-of');

var _getPrototypeOf2 = _interopRequireDefault(_getPrototypeOf);

var _classCallCheck2 = require('babel-runtime/helpers/classCallCheck');

var _classCallCheck3 = _interopRequireDefault(_classCallCheck2);

var _createClass2 = require('babel-runtime/helpers/createClass');

var _createClass3 = _interopRequireDefault(_createClass2);

var _possibleConstructorReturn2 = require('babel-runtime/helpers/possibleConstructorReturn');

var _possibleConstructorReturn3 = _interopRequireDefault(_possibleConstructorReturn2);

var _inherits2 = require('babel-runtime/helpers/inherits');

var _inherits3 = _interopRequireDefault(_inherits2);

function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }

var BufferList = require('bl');

var _require = require('readable-stream'),
    Transform = _require.Transform;

var Message = require('./message');

var _require2 = require('./packet'),
    Packet = _require2.Packet,
    HEADER_LENGTH = _require2.HEADER_LENGTH;

/**
  IncomingMessageStream
  Transform received TDS data into individual IncomingMessage streams.
*/
var IncomingMessageStream = function (_Transform) {
  (0, _inherits3.default)(IncomingMessageStream, _Transform);

  function IncomingMessageStream(debug) {
    (0, _classCallCheck3.default)(this, IncomingMessageStream);

    var _this = (0, _possibleConstructorReturn3.default)(this, (IncomingMessageStream.__proto__ || (0, _getPrototypeOf2.default)(IncomingMessageStream)).call(this, { readableObjectMode: true }));

    _this.debug = debug;

    _this.currentMessage = undefined;
    _this.bl = new BufferList();
    return _this;
  }

  (0, _createClass3.default)(IncomingMessageStream, [{
    key: 'processBufferedData',
    value: function processBufferedData(callback) {
      var _this2 = this;

      // The packet header is always 8 bytes of length.
      while (this.bl.length >= HEADER_LENGTH) {
        // Get the full packet length
        var length = this.bl.readUInt16BE(2);

        if (this.bl.length >= length) {
          var data = this.bl.slice(0, length);
          this.bl.consume(length);

          // TODO: Get rid of creating `Packet` instances here.
          var packet = new Packet(data);
          this.debug.packet('Received', packet);
          this.debug.data(packet);

          var message = this.currentMessage;
          if (message === undefined) {
            message = new Message({ type: packet.type(), resetConnection: false });
            this.push(message);
          }

          if (packet.isLast()) {
            this.currentMessage = undefined;
            // Wait until the current message was fully processed before we
            // continue processing any remaining messages.
            message.once('end', function () {
              _this2.processBufferedData(callback);
            });
            message.end(packet.data());
            return;
          } else {
            this.currentMessage = message;
            // If too much data is buffering up in the
            // current message, wait for it to drain.
            if (!message.write(packet.data())) {
              message.once('drain', function () {
                _this2.processBufferedData(callback);
              });
              return;
            }
          }
        } else {
          break;
        }
      }

      // Not enough data to read the next packet. Stop here and wait for
      // the next call to `_transform`.
      callback();
    }
  }, {
    key: '_transform',
    value: function _transform(chunk, encoding, callback) {
      this.bl.append(chunk);
      this.processBufferedData(callback);
    }
  }]);
  return IncomingMessageStream;
}(Transform);

module.exports = IncomingMessageStream;