Common Beginner Pitfall in TypeScript PubSub Subscription

Not a question, just an easy beginner pitfall I ran into. It's really easy to do:

yield* pubsub.subscribe().pipe(
  Effect.map(queue => Stream.fromQueue(queue)),
  Effect.flatMap(Stream.runForEach(onEvent)),
  Effect.fork
)

yield* pubsub.subscribe().pipe(
  Effect.map(queue => Stream.fromQueue(queue)),
  Effect.flatMap(Stream.runForEach(onEvent)),
  Effect.fork
)

And not realize that you can easily miss the first event on the queue if it fires very soon (because the subscription action itself is forked).

Versus (I think this is safe):

const stream = yield* pubsub
  .subscribe()
  .pipe(Effect.map(Stream.fromQueue))

yield* stream.pipe(Stream.runForEach(onEvent), Effect.fork)

const stream = yield* pubsub
  .subscribe()
  .pipe(Effect.map(Stream.fromQueue))

yield* stream.pipe(Stream.runForEach(onEvent), Effect.fork)

Effect Community•2y ago

Jonathan Clem

Common Beginner Pitfall in TypeScript PubSub Subscription

Not a question, just an easy beginner pitfall I ran into. It's really easy to do:

yield* pubsub.subscribe().pipe(
  Effect.map(queue => Stream.fromQueue(queue)),
  Effect.flatMap(Stream.runForEach(onEvent)),
  Effect.fork
)

yield* pubsub.subscribe().pipe(
  Effect.map(queue => Stream.fromQueue(queue)),
  Effect.flatMap(Stream.runForEach(onEvent)),
  Effect.fork
)

And not realize that you can easily miss the first event on the queue if it fires very soon (because the subscription action itself is forked).

Versus (I think this is safe):

const stream = yield* pubsub
  .subscribe()
  .pipe(Effect.map(Stream.fromQueue))

yield* stream.pipe(Stream.runForEach(onEvent), Effect.fork)

const stream = yield* pubsub
  .subscribe()
  .pipe(Effect.map(Stream.fromQueue))

yield* stream.pipe(Stream.runForEach(onEvent), Effect.fork)

Common Beginner Pitfall in TypeScript PubSub Subscription

Similar Threads

Common Beginner Pitfall in TypeScript PubSub Subscription

Similar Threads

Similar Threads

Similar Threads