red-rings-couchdb/util/db.js

// Generated by CoffeeScript 2.3.1
(function() {
  // This is a minimalist CouchDB (HTTP) API
  // It provides exactly what this module needs, but no more.
  var CouchDB, EventEmitter2, EventSource, LRU, URL, agent, autoRestart, ec, fromEventSource, http, https, lru_cache, lru_dispose, most, options, sleep, static_cache;

  LRU = require('lru-cache');

  ({EventEmitter2} = require('eventemitter2'));

  http = require('http');

  https = require('https');

  lru_dispose = new EventEmitter2({
    newListener: false,
    verboseMemoryLeak: true
  });

  options = {
    max: 200,
    dispose: function(key) {
      return ev.emit(key);
    },
    maxAge: 20 * 60 * 1000
  };

  lru_cache = LRU(options);

  static_cache = new Map();

  CouchDB = class CouchDB {
    constructor(uri, use_lru) {
      if (uri.match(/\/$/)) {
        this.uri = uri;
      } else {
        this.uri = `${uri}/`;
      }
      if (use_lru) {
        this.cache = lru_cache;
        this.limit = most.fromEvent(this.uri, lru_dispose);
      } else {
        this.cache = static_cache;
        this.limit = most.never();
      }
      switch (false) {
        case !uri.match(/^http:/):
          this.agent = agent.agent(http.globalAgent);
          break;
        case !uri.match(/^https:/):
          this.agent = agent.agent(https.globalAgent);
          break;
        default:
          this.agent = agent;
      }
      return;
    }

    put(doc) {
      var _id, uri;
      ({_id} = doc);
      uri = new URL(ec(_id), this.uri);
      return this.agent.put(uri.toString()).type('json').accept('json').send(doc).then(function({body}) {
        return body;
      });
    }

    get(_id, {rev} = {}) {
      var uri;
      uri = new URL(ec(_id), this.uri);
      if (rev != null) {
        uri.searchParams.set('rev', rev);
      }
      return this.agent.get(uri.toString()).accept('json').then(function({body}) {
        return body;
      });
    }

    delete({_id, _rev}) {
      var uri;
      uri = new URL(ec(_id), this.uri);
      if (_rev != null) {
        uri.searchParams.set('rev', _rev);
      }
      return this.agent.delete(uri.toString()).accept('json').then(function({body}) {
        return body;
      });
    }

    changes(options = {}) {
      var include_docs, live, s, since, stream, uri;
      ({live, include_docs, since} = options);
      if (live == null) {
        live = true;
      }
      if (live !== true) {
        throw new Error('Only live streaming is supported');
      }
      if (this.cache.has(this.uri)) {
        return this.cache.get(this.uri);
      }
      if (since == null) {
        since = 'now';
      }
      uri = new URL('_changes', this.uri);
      uri.searchParams.set('feed', 'eventsource');
      uri.searchParams.set('heartbeat', true);
      uri.searchParams.set('include_docs', include_docs != null ? include_docs : false);
      // The stream might end because the server disconnects or other errors.
      // In all cases we let it finish cleanly.
      s = function() {
        uri.searchParams.set('since', since);
        return fromEventSource(new EventSource(uri.toString())).continueWith(function() {
          console.error('retry', uri.host, uri.pathname, since);
          return most.never();
        }).recoverWith(function(error) {
          var ref;
          console.error('retry', (ref = error.stack) != null ? ref : error, uri.host, uri.pathname, since);
          return most.never();
        });
      };
      stream = autoRestart(s).map(function({data}) {
        return data;
      }).map(JSON.parse).tap(function({seq}) {
        if (seq != null) {
          return since = seq;
        }
      }).until(this.limit).multicast();
      this.cache.set(this.uri, stream);
      return stream;
    }

  };

  module.exports = CouchDB;

  ec = encodeURIComponent;

  most = require('most');

  EventSource = require('eventsource');

  ({fromEventSource} = require('most-w3msg'));

  ({URL} = require('url'));

  agent = require('superagent');

  // When the stream finishes, we restart it with a small delay.
  sleep = function(timeout) {
    return new Promise(function(resolve) {
      return setTimeout(resolve, timeout);
    });
  };

  autoRestart = function(s) {
    return most.generate(function*() {
      return (yield 200);
    }).startWith(0).map(async function(delay) {
      await sleep(delay);
      return s();
    }).chain(most.fromPromise).switchLatest();
  };

}).call(this);