2 replies

How to efficiently decode a byte stream?

effectstream✅Solved!

Hi, from what I gather, Sinks must be stateless and pure. For instance, in the

@effect/platform-node-shared

@effect/platform-node-shared

package, there's this Sink that illustrates what I mean by this:

export const collectUint8Array: Sink.Sink<Uint8Array, Uint8Array> = Sink.foldLeftChunks(
    new Uint8Array(),
    (bytes, chunk: Chunk.Chunk<Uint8Array>) =>
    Chunk.reduce(chunk, bytes, (acc, curr) => {
        const newArray = new Uint8Array(acc.length + curr.length)
        newArray.set(acc)
        newArray.set(curr, acc.length)
        return newArray
    })
)

export const collectUint8Array: Sink.Sink<Uint8Array, Uint8Array> = Sink.foldLeftChunks(
    new Uint8Array(),
    (bytes, chunk: Chunk.Chunk<Uint8Array>) =>
    Chunk.reduce(chunk, bytes, (acc, curr) => {
        const newArray = new Uint8Array(acc.length + curr.length)
        newArray.set(acc)
        newArray.set(curr, acc.length)
        return newArray
    })
)

As you can see, the folding function is pure, depending only on

acc

acc

and

curr

curr

and returning a new copy of the uint8array. There is no state other than the

acc

acc

, which is not modified in the folding function.

This ensures that the sink can be used with multiple streams concurrently or the same sink can be used multiple times and there will be no data crossing from one stream into the other.

On the other hand, I wanted to do something like this instead:

  const decoder = new TextDecoder("utf-8");

  const sink = pipe(
    Sink.foldLeftChunks(
    Sink.foldLeftChunks(
      Chunk.empty<string>(),
      (partials, chunk: Chunk.Chunk<Uint8Array>) =>
        Chunk.reduce(chunk, partials, (acc, curr) => {
          const partial = decoder.decode(curr, { stream: true });
          return Chunk.append(acc, partial);
        })
    ),
    Sink.map(Chunk.join('')),
  );

  const decoder = new TextDecoder("utf-8");

  const sink = pipe(
    Sink.foldLeftChunks(
    Sink.foldLeftChunks(
      Chunk.empty<string>(),
      (partials, chunk: Chunk.Chunk<Uint8Array>) =>
        Chunk.reduce(chunk, partials, (acc, curr) => {
          const partial = decoder.decode(curr, { stream: true });
          return Chunk.append(acc, partial);
        })
    ),
    Sink.map(Chunk.join('')),
  );

This is a sink that will decode a stream of bytes into a string, without creating multiple copies of the same data in memory. The problem with this approach is that it is stateful (decoder is modified in the folding function) and data from one stream will interfere with data from another stream when used concurrently or the same sink is used for multiple streams.

A way to solve this problem would be to specify a new instance of the state for every