Effect CommunityEC
Effect Community•17mo ago
danielo515

Stream.unwrap Simplifies Event Processing Pipeline in TypeScript

I just learnt about Stream.unwrap, and looking to simplify a pipeline that I have looking like this:

const events = yield* readByLAbel(label, query).pipe(
          Effect.andThen(
            flow(
              toStream,
              Stream.flatMap(Stream.fromIterable),
              Stream.take(limit),
              Stream.mapEffect(decodeEvents),
              Stream.map((x) => ({ ...x.value, ...x })),
              Stream.runCollect
            )
          )
        );
        return yield* Effect.succeed({ events: Chunk.toArray(events) });


Into this:

const events = yield* readByLAbel(label, query).pipe(
          Effect.map(toStream),
          Stream.unwrap,
          Stream.flatMap(Stream.fromIterable),
          Stream.take(limit),
          Stream.mapEffect(decodeEvents),
          Stream.map((x) => ({ ...x.value, ...x })),
          Stream.runCollect
        );
        return yield* Effect.succeed({ events: Chunk.toArray(events) });


I wonder, is this correct?
The types (an a simplified manner for your context):
- readByLAbel => Effect<Something,unknown,never>
- toStream => Something => Stream<A[]>

So far, nothing is scoped (as I assume the underlying libraries are handling that).
I recently asked if this patter of having Streams nested into Effects is correct, which to me is not but I am not very experienced in neither Effect.Stream or Effect itself, so I am never sure if I'm doing things that are semantically correct or if I'm going my way.
The naming is also a bit strange for me, as unwrap to me usually means going from Monad<A> => A with a helper function in between if needed. But here it means going from another thing to the stream one, which I could have never figured out
Was this page helpful?