Immutable State Management in a Stream Accumulator

I have this scan accumulator that appends chunks to the chunk array, which represents chunks of the Uint8array of a readable stream. For every scan call (and there may be many), it actually pushes the next uint8array on to the accumulator, which is stateeful and not properly functional because technically its mutating previous state that's already been emitted. What is an elegant way to solve this without compromising performance?

I feel like maybe using a Ref, but I don't know how to is it in the pipe way, only the generator way, and generators don't play nice with streams it seems.

        readableStream.pipe(Stream.scan(
            { done: false, length: 0, chunks: [] as Uint8Array[] },
            (accumulator, newChunk) => {
                const length = accumulator.length + newChunk.length
                accumulator.chunks.push(newChunk)
                return { length, chunks: accumulator.chunks, done: false }
            },
        ))

        readableStream.pipe(Stream.scan(
            { done: false, length: 0, chunks: [] as Uint8Array[] },
            (accumulator, newChunk) => {
                const length = accumulator.length + newChunk.length
                accumulator.chunks.push(newChunk)
                return { length, chunks: accumulator.chunks, done: false }
            },
        ))

Effect Community•13mo ago

Peter Adrian

Immutable State Management in a Stream Accumulator

        readableStream.pipe(Stream.scan(
            { done: false, length: 0, chunks: [] as Uint8Array[] },
            (accumulator, newChunk) => {
                const length = accumulator.length + newChunk.length
                accumulator.chunks.push(newChunk)
                return { length, chunks: accumulator.chunks, done: false }
            },
        ))

        readableStream.pipe(Stream.scan(
            { done: false, length: 0, chunks: [] as Uint8Array[] },
            (accumulator, newChunk) => {
                const length = accumulator.length + newChunk.length
                accumulator.chunks.push(newChunk)
                return { length, chunks: accumulator.chunks, done: false }
            },
        ))

Immutable State Management in a Stream Accumulator

Similar Threads

Immutable State Management in a Stream Accumulator

Similar Threads

Similar Threads

Similar Threads