<commit_before>"use strict"

const Parser = require("./Parser");

module.exports = (scope, sequence_id,  event) => {

    // TODO error handling in async buffer
    return new Promise((resolve, reject) => {

        let sequence = scope.cache.get(sequence_id);
        if (sequence) {
            if (sequence.roles[event.role]) {
                sequence.call(event, resolve);
            } else {
                reject(new Error('Flow: Sorry, no access. Roles:' + Object.keys(sequence.roles) + ' Role:' + event.role));
            }
            return;
        }

        // create a sequence
        sequence = Sequence(sequence_id, event, resolve, reject);
        scope.cache.set(sequence_id, sequence);

        const parser = Parser(scope, sequence, event.role);
        const error = (err) => {};
        parser.on('finish', sequence.fin);
        parser.on('error', (err) => {
            scope.cache.del(sequence_id);
            reject(err);
        });

        setImmediate(() => {
            const triples = scope.seq(scope, sequence_id, event.role);
            triples.on('error', (err) => {
                scope.cache.del(sequence_id);
                reject(err);
            });
            triples.pipe(parser);
        });
    });
};

// TODO move to debug version
function seqErr (msg, sequence) {
    sequence.calls = [];
    return new Error(
        "Flow.sequence.done: " + msg + ". Sequence:\n" +
        "ID:    " + sequence.id + "\n" +
        "First: " + sequence.first + "\n" +
        "Roles: " + JSON.stringify(sequence.roles, null, 4)
    );
}

function Sequence (sequence_id, event, resolve, reject) {

    const sequence = {
        id: sequence_id,
        fns: 1,
        seq: {},
        roles: {},
        calls: [[event, resolve]],

        done: () => {

            if (!sequence.first) {
                return reject(seqErr("No first handler set on sequence", sequence));
            }

            if (!sequence.seq[sequence.first]) {
                return reject(seqErr("First handler not in sequence.", sequence));
            }

            if (Object.keys(sequence.roles) < 1) {
                return reject(seqErr("No roles on sequence.", sequence));
            }

            // TODO reject all calls on error

            sequence.ready = true;
            sequence.calls.forEach((event) => {sequence.call(event[0], event[1])});
            delete sequence.calls;
        },

        fin: () => {
            sequence.parsed = true;
            if (--sequence.fns < 1) {
                sequence.done();
            }
        },

        call: (event, resolve) => {
            sequence.ready ? resolve([event, sequence]) : sequence.calls.push([event, resolve]);
        } 
    };

    return sequence;
};<commit_msg>use promises in combination with event emitter<commit_after>"use strict"

const EventEmitter = require("events");
const vocab = "<http://schema.jillix.net/vocab/";

module.exports = (scope, sequence_id, event) => {

    const sequence = scope.cache.get(sequence_id);
    if (sequence) {

        // wait till sequence is loaded
        if (sequence.loading) {
            return new Promise((resolve, reject) => {
                sequence.once("reject", reject);
                sequence.once("resolve", (sequence) => {
                    resolve([event, sequence]);
                });
            });

        // resolve promise if event role has accesses to event
        } else if (sequence.roles[event.role]) {
            return Promise.resolve([event, sequence]);

        // reject promise
        } else {
           return Promise.reject(new Error('Flow: Sorry, no access. Roles:' + Object.keys(sequence.roles) + ' Role:' + event.role));
        }
    }

    return new Promise((resolve, reject) => {

        const loader = new EventEmitter();
        loader.loading = true;
        loader.wait = [];
        loader.once("reject", (err) => {
            scope.cache.del(sequence_id);
            reject(err);
        });
        loader.once("resolve", (sequence) => {
            scope.cache.set(sequence_id, sequence);
            resolve([event, sequence]);
        });
        scope.cache.set(sequence_id, loader);

        const sequence = {
            id: sequence_id,
            roles: {}
        };

        setImmediate(() => {
            const triples = scope.seq(scope, sequence_id, event.role);
            triples.on('end', () => {
                if (loader.wait.length) {
                    Promise.all(loader.wait)
                    .then(() => {loader.emit("resolve", sequence)})
                    .catch((err) => {loader.emit("reject", err)});
                } else {
                    loader.emit("resolve", sequence);
                }
            });
            triples.on('error', (err) => {loader.emit("reject", err)});
            triples.on("data", (triple) => {

                let subject = triple[0] || triple.subject;
                let predicate = triple[1] || triple.predicate;
                let object = triple[2] || triple.object;

                switch (predicate) {

                    case vocab + 'onError>':
                        sequence.err = object;
                        break;

                    case vocab + 'role>':
                        sequence.roles[object] = true;
                        break; 

                    case vocab + 'next>':
                        if (subject === sequence.id) {
                            sequence.first = object;
                        } else {
                            setHandler(sequence, subject, 'next', object);
                        }
                        break;

                    case vocab + 'args>':
                        try {
                            setHandler(sequence, subject, 'args', Object.freeze(JSON.parse(object)));
                        } catch (err) {
                            return loader.emit("reject", err);
                        }
                        break;

                    case vocab + 'state>':
                        setHandler(sequence, subject ,'state', getState(scope, object));
                        break;

                    case vocab + 'fn>':
                        loader.wait.push(Fn(scope, sequence, subject, object, event.role));
                        break;

                    default:
                        reject(new Error('[Parser] Unknown triple: ' + predicate));
                }
            });
        });
    });
};

function Fn (scope, sequence, handler_id, fn_iri, role) {

    const fn = scope.cache.get(fn_iri);
    if (fn) {

        // wait till sequence is loaded
        if (fn.loading) {
            return new Promise((resolve, reject) => {
                fn.once("resolve", (fn) => {
                    setHandler(sequence, handler_id, "fn", fn);
                    resolve(fn);
                });
                fn.once("reject", reject);
            });

        // resolve promise if event role has accesses to event
        } else {
            setHandler(sequence, handler_id, "fn", fn);
            return Promise.resolve(fn);
        }
    }

    return new Promise((resolve, reject) => {

        const loader = new EventEmitter();
        loader.loading = true;
        loader.wait = [];
        loader.once("reject", (err) => {
            scope.cache.del(fn_iri);
            reject(err);
        });
        loader.once("resolve", (fn) => {
            setHandler(sequence, handler_id, "fn", fn);
            scope.cache.set(fn_iri, fn);
            resolve(fn);
        });
        scope.cache.set(fn_iri, loader);

        scope.fn(fn_iri, role, (err, fn) => {

            if (err) {
                return loader.emit("reject", err);
            }

            loader.emit("resolve", fn);
        });
    });
};

function getState (scope, state_id) {
    let state = scope.cache.get(state_id);
    if (state) {
        return state;
    }

    state = {};
    scope.cache.set(state_id, state);
    return state;
}

function setHandler (seq, handler, key, value) {
    seq[handler] = seq[handler] || {};
    seq[handler][key] = value;
}
{'fraction_non_alphanumeric': 0.13811274509803922, 'fraction_numerical': 0.0011029411764705882, 'mean_word_length': 1.4507507507507507, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 37, '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': '179064', 'n_tokens_mistral': 2208, 'n_tokens_neox': 2110, 'n_words': 486}