Misc collection of shorthand helper functions for Flux implemented with Rx

rxjs-flux-utils.js

/**
Misc collection of shorthand helper functions for Flux implemented with Rx
**/

import Rx from 'rx';
import _ from 'lodash';
import formSerialize from 'form-serialize';

Rx.Observable.prototype.byTag = byTag;

// Enforce actions sent through the dispatcher go through one of the action
// creator functions below.
var DispatcherSub = new Rx.Subject();
export var Dispatcher = DispatcherSub.asObservable();

/**
Send tagged events through the dispatcher

@example
    sendTag('foo/bar', {foo: 'Foo!'})
    sendTag('foo/bar')
**/
export function sendTag(
        ltag: string,
        data: Object = {}): void {
    var msg = {tag: ltag, data};
    return DispatcherSub.onNext(msg);
}

/**
Helper function for `sendTag()` for use with callbacks

A function may be passed in the `data` argument. It will be called before
calling `sendTag()`. This allows pre-processing event data before sending it
through the dispatcher. The function is called with the same parameters as the
callback.

@example
    myComponent = React.createClass({
        render: function() {
            return h('p', [
                h('a', {onClick: send('my/tag/foo')}),
                h('a', {onClick: send('my/tag/bar', {msg: 'Hello'})}),
                h('input', {onChange: send('my/tag/baz', {
                    thedata: (ev) => ev.target.value})}),
            ]);
        },
    });
**/
export function send(ltag: string, initData: Object = {}): Function {
    return function(...args: Array<any>): void {
        var data = _.merge({}, initData);
        _.forEach(initData, function(val, key) {
            if (_.isFunction(val)) {
                data[key] = val.apply(null, args);
            }
        });

        return sendTag(ltag, data);
    };
}

/**
Catch a form submit event and send the form values through the dispatcher

@example
    h('form', {
        onSubmit: sendForm('some/tag'),
    }, [...]);

    h('form', {
        onSubmit: sendForm('some/tag', {extra: 'data'}),
    }, [...]);
**/
export function sendForm(ltag: string, data: Object = {}): Function {
    return function(ev: Object): void {
        ev.preventDefault();

        var combinedData = _.merge(
            formSerialize(ev.target, {hash: true}),
            data);
        sendTag(ltag, combinedData);
    };
}

/**
Filter messages through the event stream by tag pattern matching

The pattern may contain glob-style tokens which will be matched against tags
coming through the event stream.
**/
// @example
// var Store = Dispatcher.byTag('foo/*/baz');
export function byTag(...args: Array<string>): boolean {
    // reuse the regex objs.
    var fnList = [];
    for (let pattern of args) {
        fnList.push(fnmatch(pattern));
    }

    return this.filter(function(stream) {
        if (!stream.tag) { return false; }

        for (let fn of fnList) {
            if (fn(stream.tag)) { return true; }
        }

        return false;
    });
}


function globStringToRegex(str) {
    // http://stackoverflow.com/a/13818704/127816
    return new RegExp('^' + preg_quote(str).replace(/\\\*/g, '.*').replace(/\\\?/g, '.') + '$');
}
function preg_quote(str, delimiter) {
    return (str + '').replace(new RegExp('[.\\\\+*?\\[\\^\\]$(){}=!<>|:\\' + (delimiter || '') + '-]', 'g'), '\\$&');
}

export function fnmatch(match, string) {
    /**
    Perform shell-style globbing on strings

    Function can be curried to cache the generated regex object for performance.

    Usage:
        fnmatch('foo*')('foobarbaz')
        // true
    **/
    var glob = globStringToRegex(match);
    if (arguments.length < 1) {
        return fnmatch;
    } else if (arguments.length < 2) {
        return x => glob.test(x);
    } else {
        return glob.test(string);
    }
}

rxjs-xhr-utils.js

import Rx from 'rx';
import _ from 'lodash';
import 'rx-dom-ajax';

export const defHeaders = {
    'Accept': 'application/json',
    'Content-type': 'application/json',
};

/**
Allow external parties to watch the status of XHR requests and responses

Emits Salt-style event tags of XMLHttpRequest2 progress events. The tag is the
URL path and the event data contains the request data as well as the raw
progress event.
**/
export const Progress$ = new Rx.Subject();

/**
A wrapper around Rx.DOM.ajax that provides pre/post processing

@returns {Observable} Completion and error results are normalized to the same
    data structure and both returned in the `onNext` callback.

@prop {object} params.qs - Key/val pairs that will be transformed into query
    string key=val pairs. Arrays of strings as the value will be repeated as
    `key=val1&keyval2` pairs.
@prop {any} params.body - A value that will be used as the request body. If the
    `Content-Type` header is JSON this value will first be serialized as JSON.
@prop {object} params.headers - An object of key/vals to use for request
    headers. This object will augment the default headers in `xhr.defHeaders`.
@prop {string} params.etag - An optional Etag to add to the request headers.
@prop {string} params.progress - A Salt-style event tag used to prefix progress
    events. Defaults to the path if omitted (leading slash omitted).

@example
    xhrNext('GET', '/some/path');
    xhrNext('GET', '/some/path', {qs: {foo: 'Foo!'}});
    xhrNext('POST', '/some/path', {body: {foo: 'Foo!'}});
    xhrNext('GET', '/some/path', {headers: {'Accept': 'text/html'}});
    xhrNext('GET', '/some/path', {etag: '123456'});
    xhrNext('GET', '/some/path', {progress: 'foo/bar/*'});
**/
export function xhrNext(method: string, path: string, params: Object = {}) {
    var {
        body,
        qs: qs = {},
        headers: headers = {},
        etag,
    } = params;

    _.defaults(headers, defHeaders);
    var url = _.isEmpty(qs) ? path : [path, obj2qs(qs)].join('?');

    // Automatically make a conditional-GET request if possible.
    if (etag) { headers['If-None-Match'] = etag; }

    const emitProgress = ev => Progress$.onNext({
        tag: params.progress || _.trimStart(url, '/'),
        data: {qs, body, headers, url, method, ev, xhr: ev.target},
    });

    var reqObj = {
        method,
        url,
        body: getOrParseBody(body, headers['Content-type']),
        headers,

        progressObserver: Rx.Observer.create(
            emitProgress,
            emitProgress,
            () => {}),

        normalizeSuccess,
        normalizeError,
    };

    return Rx.DOM.ajax(reqObj)
        .catch(x => Rx.Observable.just(x));
}

function normalizeSuccess(e, xhr, type) {   // eslint-disable-line no-shadow
    var ctype = xhr.getResponseHeader('content-type') || '';
    var body = _.get(xhr, 'response', xhr.responseText);
    var response = ctype.includes('json') ? JSON.parse(body || '{}') : body;

    return {
        status: xhr.status,
        response,
        responseType: xhr.responseType,
        xhr,
        originalEvent: e,
        type,

        errors: false,
    };
}

function normalizeError(e, xhr, type) {   // eslint-disable-line no-shadow
    return {
        status: xhr.status,
        response: xhr.response,
        responseType: xhr.responseType,
        xhr,
        originalEvent: e,
        type,

        errors: true,
    };
}

// - Util functions -----------------------------------------------------------

/**
Parse, or don't parse, the request body based on the content type
**/
function getOrParseBody(body: any, contentType: string): ?string {
    if (body == null) {
        return null;
    } else {
        if (contentType.includes('json')) {
            return JSON.stringify(body);
        } else {
            return body;
        }
    }
}

/**
Generate a querystring from an object

@example
    obj2qs({foo: 'Foo', bar: 'Bar'});
    // => foo=Foo&bar=Bar
**/
export function obj2qs(obj) {
    return _.map(obj, function(val, key) {
        if (_.isArray(val)) {
            return _.map(val, i => `${key}=${encodeURIComponent(i)}`).join('&');
        } else {
            return `${key}=${encodeURIComponent(val)}`;
        }
    }).join('&');
}

// - Caching decorator --------------------------------------------------------

/**
A decorator to cache xhr returns and make conditional GET requests for updates

The cache is stored in an Rx ReplaySubject that tracks active subscribers.

@example
    const xhr = cachingXhr(xhrNext);
    const sub1 = xhr('GET', '/minions/cache').subscribe();  // 200 response
    const sub2 = xhr('GET', '/minions/cache').subscribe();  // 304 response
    xhr.clean();    // noop
    sub1.displose();
    sub2.displose();
    xhr.clean();    // destroys cache
**/
export function cachingXhr(fn) {
    var cache = new Map();
    getXhr.cache = cache;
    getXhr.clean = cleanOldSubjects;
    return getXhr;

    function getXhr(...args) {
        var [method, path, params = {}] = args;
        var {qs: qs = {}} = params;

        // Only cache GET requests.
        if (method !== 'GET') { return fn(...args); }

        var objKey = [method, path, JSON.stringify(qs)].join('_|-');
        var {replay, etag: prevEtag} = getOrSetReplay(objKey);

        // Assemble args for the xhr call.
        var callArgs = [method, path];
        if (prevEtag) { params.etag = prevEtag; }
        if (!_.isEmpty(qs)) { params.qs = qs; }
        if (!_.isEmpty(params)) { callArgs.push(params); }

        var xhr$ = fn(...callArgs);

        // RaaS returns an empty body on 304 so emit the cached return instead.
        return xhr$
            .withLatestFrom(replay.startWith(false), function(curRet, prevRet) {
                if (curRet.status === 304) {
                    prevRet.status = 304;
                    return prevRet;
                } else {
                    return curRet;
                }
            })
            // Update the cache with the most recent Etag.
            .do(({xhr}) =>
                getOrSetReplay(objKey, xhr.getResponseHeader('Etag')))
            // Reuse the same Subject for each cached RaaS endpoint.
            .multicast(replay)
            .refCount();
    }

    /**
    Return a cached ReplaySubject or create one

    The replay is non-completing so it can retain subscribers over time and
    over multiple usages.
    **/
    function getOrSetReplay(objKey, newEtag) {
        var isCached = cache.has(objKey);

        var {replay, etag: oldEtag} = isCached
            ? cache.get(objKey)
            : {replay: new Rx.ReplaySubject(1)};
        replay.onCompleted = noop;

        cache.set(objKey, {replay, etag: newEtag || oldEtag});
        return cache.get(objKey);

        function noop() {}
    }

    /**
    Delete cached subjects if there are no current subscribers
    **/
    function cleanOldSubjects() {
        var cleaned = [];

        for (var [objKey, {replay}] of cache) {
            if (!replay.hasObservers()) {
                cache.delete(objKey);
                cleaned.push(objKey);
            }
        }

        return cleaned;
    }
}