@events/server/build/subscribe.js

'use strict';

Object.defineProperty(exports, "__esModule", {
  value: true
});

var _extends = Object.assign || function (target) { for (var i = 1; i < arguments.length; i++) { var source = arguments[i]; for (var key in source) { if (Object.prototype.hasOwnProperty.call(source, key)) { target[key] = source[key]; } } } return target; };

var _kefir = require('kefir');

var _kefir2 = _interopRequireDefault(_kefir);

var _once = require('lodash/fp/once');

var _once2 = _interopRequireDefault(_once);

var _range = require('lodash/fp/range');

var _range2 = _interopRequireDefault(_range);

var _redisClient = require('./redis-client');

var _query = require('./query');

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

function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }

var tryParse = function tryParse(raw) {
  try {
    return JSON.parse(raw);
  } catch (ex) {
    return { type: '@@RAW', payload: raw };
  }
};

function flush(response) {
  if (response.flush && response.flush.name !== 'deprecated') {
    response.flush();
  }
}

var toOutput = function toOutput(events) {
  return 'id: ' + events[events.length - 1].id + '\nevent: INCMSG\ndata: ' + JSON.stringify(events) + '\n\n';
};

exports.default = function (_ref) {
  var redis = _ref.redis,
      history = _ref.history,
      debug = _ref.debug,
      namespc = _ref.namespc,
      burst = _ref.burst;

  var subClient = (0, _redisClient.createClient)(redis);
  var queryClient = (0, _redisClient.createClient)(redis);

  // prepare cache
  var MAX_SIZE = history.size;
  var list = [];
  var cache = {};

  var addToCache = function addToCache(message) {
    debug && console.log('adding to cache', message.id);

    var newSize = list.unshift(message.id);
    cache[message.id] = message;

    if (newSize > MAX_SIZE) {
      var expired = list.splice(MAX_SIZE);
      expired.forEach(function (removingId) {
        delete cache[removingId];
      });
    }

    debug && console.log('HISTORY', list);
    debug && console.log('CACHE\n', cache);
  };

  var message$ = _kefir2.default.fromEvents(subClient, 'message', function (channel, message) {
    return message;
  }).map(function (message) {
    var rawId = message.split(':')[0];
    var id = Number(rawId);
    var rawEvent = message.slice(rawId.length + 1);

    return _extends({
      id: id
    }, tryParse(rawEvent));
  }).onValue(addToCache);

  var addClient = function addClient(src, opts, match, req, res) {
    req.socket.setTimeout(0);
    req.socket.setNoDelay(true);
    req.socket.setKeepAlive(true);

    res.writeHead(200, {
      'Content-Type': 'text/event-stream;charset=UTF-8',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive'
    });

    res.write(':ok\n\n');

    // cast to integer
    var retry = opts.retry | 0;

    if (retry) {
      res.write('retry: ' + retry + '\n');
    }

    flush(res);

    var subscription = src.filter(function (x) {
      return x;
    }).bufferWithTimeOrCount(burst.time, burst.count).filter(function (b) {
      return b.length;
    }).map(toOutput).observe(function (block) {
      debug && console.log('send to %s %s', req.url, block);
      res.write(block);
    });

    var removeClient = (0, _once2.default)(function () {
      debug && console.log('removing', req.url);
      subscription.unsubscribe();
      res.end();
    });

    req.on('end', removeClient);
    req.on('close', removeClient);
    res.on('finish', removeClient);
  };

  var getInitialValues = function getInitialValues(lastEventId) {
    if (!lastEventId) {
      return _kefir2.default.never();
    }

    // history[0] is the latest id
    var current = list[0];
    var oldestInCache = list[list.length - 1];

    debug && console.log('current', current);
    if (oldestInCache <= lastEventId + 1) {
      return _kefir2.default.constant((0, _range2.default)(lastEventId + 1, current + 1).map(function (id) {
        return cache[id];
      })).flatten();
    }

    debug && console.error('too old, getting more from redis');
    var fromCache$ = _kefir2.default.constant((0, _range2.default)(oldestInCache, current + 1).map(function (id) {
      return cache[id];
    }));

    var fromRedis$ = _kefir2.default.fromPromise((0, _query.query)(queryClient, namespc, lastEventId + 1, oldestInCache - 1));

    return _kefir2.default.concat([fromRedis$, fromCache$]).flatten();
  };

  var service = function () {
    var _ref2 = _asyncToGenerator(function* (req, res) {
      try {
        debug && console.log('connected', req.url);
        var lastEventId = Number(req.headers['last-event-id']);

        var initialValues = getInitialValues(lastEventId);

        debug && console.log('lastEventId', lastEventId);

        // const url = req.url;
        // const query = parse(url.split('?')[1]);

        addClient(_kefir2.default.concat([initialValues, message$]), {}, {}, req, res);
      } catch (ex) {
        console.log(ex);
        throw ex;
      }
    });

    return function service(_x, _x2) {
      return _ref2.apply(this, arguments);
    };
  }();

  subClient.subscribe(namespc + '::events');

  return {
    service: service,
    unsubscribe: function unsubscribe() {
      console.log('unsubscribing from redis');
      subClient.unsubscribe(namespc + '::events');
    }
  };
};