<commit_before>/**
 * Created by ndyumin on 15.09.2016.
 */

const {removeItem} = require('./arrays');
const {fromCallbacks, notifyAll} = require('./observers');
const symbolObservable = require('symbol-observable');

function project(events, initial) {
    return events.reduce(function (state, event) {
        return event.reduce(state, event.update);
    }, initial);
}

const eventCreatorFactory = reduce => update => ({
    reduce,
    update
});

const commandCreatorFactory = executor => executor;

function store(initial) {
    let events = [];
    let replicas = [];

    return {
        [symbolObservable]: function () {
            return this;
        },
        _eventLog (observer) {
            if (typeof observer === 'function') {
                observer = fromCallbacks(...arguments);
            }
            events.forEach(observer.next);
            replicas.push(observer);

            return {
                unsubscribe() {
                    replicas = removeItem(replicas, observer);
                }
            };
        },
        subscribe(observer) {
            if (typeof observer === 'function') {
                observer = fromCallbacks(...arguments);
            }

            return this.view(project).subscribe(observer);
        },
        plug(observable, reducer) {
            const event = this.eventCreatorFactory(reducer);
            return observable.subscribe({
                next: event
            });
        },
        view(projectFn) {
            let _viewObservers = [];
            const eventLog = this._eventLog;

            return {
                subscribe (observer) {
                    if (typeof observer === 'function') {
                        observer = fromCallbacks(...arguments);
                    }
                    _viewObservers.push(observer);
                    const subs = eventLog(function () {
                        observer.next(projectFn(events, initial));
                    });

                    return {
                        unsubscribe() {
                            subs.unsubscribe();
                            _viewObservers = removeItem(_viewObservers, observer);
                        }
                    };
                }
            }
        },
        transaction() {
            const _branchRev = events.length;
            const _altEvents = [];
            const branch = store(initial);
            const subs = this._eventLog(branch.dispatch);
            const bSubs = branch._eventLog(ev => _altEvents.push(ev));
            return {
                store: () => branch,
                dispatch: branch.dispatch,
                commit: () => {
                    subs.unsubscribe();
                    bSubs.unsubscribe();
                    events = events.slice(0, _branchRev).concat(_altEvents);
                    _altEvents.length = 0;
                    notifyAll(replicas, eventCreatorFactory(s=>s)())
                },
                cancel: () => {
                    subs.unsubscribe();
                    bSubs.unsubscribe();
                    _altEvents.length = 0;
                }
            };
        },
        dispatch(event) {
            events.push(event);
            notifyAll(replicas, event);
        },
        commandCreatorFactory,
        eventCreatorFactory(reduce) {
            return update =>
                this.dispatch({
                    reduce,
                    update
                });
        }
    };
}

module.exports = {
    store,
    eventCreatorFactory,
    commandCreatorFactory,
    defaultProjection: project
};<commit_msg>- notify of an initial value on a subscription<commit_after>/**
 * Created by ndyumin on 15.09.2016.
 */

const {removeItem} = require('./arrays');
const {fromCallbacks, notifyAll} = require('./observers');
const symbolObservable = require('symbol-observable');

function project(events, initial) {
    return events.reduce(function (state, event) {
        return event.reduce(state, event.update);
    }, initial);
}

const eventCreatorFactory = reduce => update => ({
    reduce,
    update
});

const commandCreatorFactory = executor => executor;

function store(initial) {
    let events = [];
    let replicas = [];

    return {
        [symbolObservable]: function () {
            return this;
        },
        _eventLog (observer) {
            if (typeof observer === 'function') {
                observer = fromCallbacks(...arguments);
            }
            events.forEach(observer.next);
            replicas.push(observer);

            return {
                unsubscribe() {
                    replicas = removeItem(replicas, observer);
                }
            };
        },
        subscribe(observer) {
            if (typeof observer === 'function') {
                observer = fromCallbacks(...arguments);
            }

            return this.view(project).subscribe(observer);
        },
        plug(observable, reducer) {
            const event = this.eventCreatorFactory(reducer);
            return observable.subscribe({
                next: event
            });
        },
        view(projectFn) {
            let _viewObservers = [];
            const onEvent = this._eventLog;

            return {
                subscribe (observer) {
                    if (typeof observer === 'function') {
                        observer = fromCallbacks(...arguments);
                    }
                    _viewObservers.push(observer);
                    const projectAndNotify = () =>
                        observer.next(projectFn(events, initial));

                    projectAndNotify();

                    const subscription = onEvent(projectAndNotify);

                    return {
                        unsubscribe() {
                            subscription.unsubscribe();
                            _viewObservers = removeItem(_viewObservers, observer);
                        }
                    };
                }
            }
        },
        transaction() {
            const _branchRev = events.length;
            const _altEvents = [];
            const branch = store(initial);
            const subs = this._eventLog(branch.dispatch);
            const bSubs = branch._eventLog(ev => _altEvents.push(ev));
            return {
                store: () => branch,
                dispatch: branch.dispatch,
                commit: () => {
                    subs.unsubscribe();
                    bSubs.unsubscribe();
                    events = events.slice(0, _branchRev).concat(_altEvents);
                    _altEvents.length = 0;
                    notifyAll(replicas, eventCreatorFactory(s=>s)())
                },
                cancel: () => {
                    subs.unsubscribe();
                    bSubs.unsubscribe();
                    _altEvents.length = 0;
                }
            };
        },
        dispatch(event) {
            events.push(event);
            notifyAll(replicas, event);
        },
        commandCreatorFactory,
        eventCreatorFactory(reduce) {
            return update =>
                this.dispatch({
                    reduce,
                    update
                });
        }
    };
}

module.exports = {
    store,
    eventCreatorFactory,
    commandCreatorFactory,
    defaultProjection: project
};
{'fraction_non_alphanumeric': 0.10660721524118362, 'fraction_numerical': 0.0029725712741521418, 'mean_word_length': 1.2966180577102078, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2520723', 'n_tokens_mistral': 1819, 'n_tokens_neox': 1706, 'n_words': 333}