libraries

rxjs.ts

const observable = (predicate) => (listener) => predicate(listener);
const pipe = (obs) => (...operators) => operators.reduce((acc, op) => op(acc), obs);

const map = (mapper) => (obs) => observable((resolve) => obs((data) => resolve(mapper(data))));
const mergeMap = (mapper) => (obs) => observable((resolve) => obs((data) => mapper(data)(resolve)))

const filter = (predicate) => (obs) => observable((resolve) => obs((data) => predicate(data) && resolve(data)));

// {
//     return (listener) => {
//         predicate(listener);
//     };
// };

pipe(observable((resolve) => {
    let k = setTimeout(() => {
        resolve(Math.random());
    }, 0);

    return () => clearTimeout(k);
}))(
    map(x => x * 2),
    mergeMap((value) => {
        // some async operation
        return observable((resolve) => {
            setTimeout(() => {
                resolve(`value after ${value * 1000}ms is ${value}`);
            }, value * 1000)
        });
    }),
)(console.log)