rxtimeline.js

Rx.Observable.prototype.sendToTimeline = function(name) {
    const addToTimeline = (group, data, className) => {
        items.add({
            start: new Date(),
            content: stringify(data),
            group: group.id,
            className
        });

        if (shouldFit) timeline.fit();
    };

    const setGroupClass = (group, className) => groups.update(Object.assign({}, group, { className }));

    const source = this, key = this;

    // Get the group (creating it if necessary)
    let group;
    if (!streams.has(key)) {
        group = { id: cuid(), content: name };
        groups.add(group);
        streams.set(key, groups);
    } else {
        group = streams.get(key);
    }

    return Rx.Observable.create(o => {
        return source.subscribe(
            data => {
                addToTimeline(group, data);
                o.onNext(data);
            },
            err => {
                addToTimeline(group, err.message, 'error');
                setGroupClass(group, 'error');
                o.onError(err);
            },
            () => {
                setGroupClass(group, 'completed');
                o.onCompleted();
            }
        );
    });
};