More readable implementation of callback based duplex stream about callbag HOT 13 OPEN

type Func = (...args: any[]) => any;
type Command = 0 | 1 | 2;

type Callbag = (cmd: Command, data?: any) => void;
type Producer = Callbag;
type Consumer = Callbag;

type Source = (c: Consumer) => void;
type Sink = (s: Source) => void;

type ToSource<T> = (x:T) => Source;
type FromSource<T> = (s:Source) => T;
type SourceTransformer = (s: Source) => Source;

type ToSink<T> = (x:T) => Sink;

function compose(...fs: Func[])
{
    return (x: any) =>
    {
        let res = x;
        for(let i = 0, n = fs.length; i < n; i++)res = fs[i](res);
        return res;
    };
}

const genericCallbag: Callbag = (c, d) => null;

const genericSource: Source = consumer =>
{
    const producer = genericCallbag;
    consumer(0, producer);
};

const genericSink: Sink = source =>
{
    const consumer = genericCallbag;
    source(consumer);
};

const pullableSource:ToSource<Iterable<any>> = iterable => (consumer =>
{
    const iterator = iterable[Symbol.iterator]();

    const producer: Callbag = (c, d) =>
    {
        switch(c)
        {
            case 0: break;
            case 1:
                const res = iterator.next();
                if(res.done)consumer(2);
                else consumer(1, res.value);
                break;
            case 2: break;
        }
    };

    consumer(0, producer);
});

const pullingSink:ToSink<Func> = f => (source =>
{
    let producer = genericCallbag;

    const consumer: Callbag = (cmd, data) =>
    {
        switch(cmd)
        {
            case 0: producer = data; producer(1); break;
            case 1: f(data); producer(1); break;
            case 2: break;
        }
    };

    source(consumer);
});

const map = (f: Func):SourceTransformer => (originalSource => consumer =>
{
    const mappedConsumer: Callbag = (cmd, data) =>
    {
        if(cmd === 1)return consumer(1, f(data));
        return consumer(cmd, data);
    }
    originalSource(mappedConsumer);
});

const take = (amount: number):SourceTransformer => (originalSource => consumer =>
{
    let taken = 0;
    let producer = genericCallbag;

    const boundedProducer: Callbag = (c, d) =>
    {
        if(c === 1)
        {
            if(taken < amount)
            {
                taken++;
                return producer(1, d);
            }
            else
            {
                producer(2);
                return consumer(2);
            }
        }
        if(c === 2)producer(2, d);
    };

    const boundedConsumer: Callbag = (cmd, data) =>
    {
        switch(cmd)
        {
            case 0:
                producer = data;
                consumer(0, boundedProducer);
                break;
            case 1: consumer(1, data); break;
            case 2: consumer(2); break;
        }
    }

    originalSource(boundedConsumer);
});

function* genMaker()
{
    let i = 0;
    while(i < i + 1)yield i++;
}

const gen = genMaker();

const log_square_of_first_4_from_iterable = compose(pullableSource, take(4), map(x => x * x), pullingConsumer(x => console.log(x)));

log_square_of_first_4_from_iterable(gen);

type Producer = Callbag;
type Consumer = Callbag;