export function streamFrom(startWatching) {
    let unsub;
    return new ReadableStream({
        start(ctrl) {
            unsub = startWatching({
                next: v => ctrl.enqueue(v),
                complete: () => ctrl.close(),
                error: err => ctrl.error(err)
            });
        },
        cancel: () => unsub()
    });
}
export function map(fn) {
    return new TransformStream({
        transform: (v, ctrl) => ctrl.enqueue(fn(v))
    });
}
export function filter(predicate) {
    return new TransformStream({
        transform: (v, ctrl) => {
            if (predicate(v)) {
                ctrl.enqueue(v);
            }
        }
    });
}
export function doWithEach(fn) {
    return new TransformStream({
        transform: (v, ctrl) => {
            try {
                fn(v);
            }
            catch (err) {
                console.error(err);
            }
            ctrl.enqueue(v);
        }
    });
}
export function doAndAwaitWithEach(fn) {
    return new TransformStream({
        transform: async (v, ctrl) => {
            try {
                await fn(v);
            }
            catch (err) {
                console.error(err);
            }
            ctrl.enqueue(v);
        }
    });
}
export async function* readerToIterable(reader) {
    while (true) {
        const { done, value } = await reader.read();
        if (done) {
            break;
        }
        else {
            yield value;
        }
    }
}
