1/**
2 * @title Broadcasting domain events with PubSub
3 *
4 * Build an in-process event bus with `PubSub` and expose it as a service.
5 */
6import { Context, Effect, Layer, PubSub, Stream } from "effect"
7
8export type OrderEvent =
9 | { readonly _tag: "OrderPlaced"; readonly orderId: string }
10 | { readonly _tag: "PaymentCaptured"; readonly orderId: string }
11 | { readonly _tag: "OrderShipped"; readonly orderId: string }
12
13export class OrderEvents extends Context.Service<OrderEvents, {
14 publish(event: OrderEvent): Effect.Effect<void>
15 publishAll(events: ReadonlyArray<OrderEvent>): Effect.Effect<void>
16 readonly subscribe: Stream.Stream<OrderEvent>
17}>()("acme/OrderEvents") {
18 static readonly layer = Layer.effect(
19 OrderEvents,
20 Effect.gen(function*() {
21 // Use PubSub.bounded to create a PubSub with backpressure support.
22 // You can also use PubSub.unbounded if you don't need backpressure.
23 const pubsub = yield* PubSub.bounded<OrderEvent>({
24 capacity: 256,
25 // Optionally add a replay buffer to let late subscribers catch up on
26 // recent events after restarts.
27 replay: 50
28 })
29
30 // Ensure the PubSub is properly shut down when the service is no longer
31 // needed.
32 yield* Effect.addFinalizer(() => PubSub.shutdown(pubsub))
33
34 const publish = Effect.fn("OrderEvents.publish")(function*(event: OrderEvent) {
35 yield* PubSub.publish(pubsub, event)
36 })
37
38 const publishAll = Effect.fn("OrderEvents.publishAll")(function*(events: ReadonlyArray<OrderEvent>) {
39 yield* PubSub.publishAll(pubsub, events)
40 })
41
42 // Create a Stream that emits events published to the PubSub.
43 //
44 // Each subscriber will receive all events published after they subscribe,
45 // and if a replay buffer is configured, they will also receive the most
46 // recent events that were published before they subscribed.
47 const subscribe = Stream.fromPubSub(pubsub)
48
49 return OrderEvents.of({
50 publish,
51 publishAll,
52 subscribe
53 })
54 })
55 )
56}