| 563 | ) |
| 564 | |
| 565 | const subscribeDurable = (aggregateID: string) => |
| 566 | Effect.gen(function* () { |
| 567 | const wake = yield* PubSub.sliding<void>(1) |
| 568 | const subscription = yield* PubSub.subscribe(wake) |
| 569 | yield* Effect.acquireRelease( |
| 570 | Effect.sync(() => { |
| 571 | const wakes = pubsub.durable.get(aggregateID) ?? new Set() |
| 572 | wakes.add(wake) |
| 573 | pubsub.durable.set(aggregateID, wakes) |
| 574 | }), |
| 575 | () => |
| 576 | Effect.sync(() => { |
| 577 | const wakes = pubsub.durable.get(aggregateID) |
| 578 | wakes?.delete(wake) |
| 579 | if (wakes?.size === 0) pubsub.durable.delete(aggregateID) |
| 580 | }).pipe(Effect.andThen(PubSub.shutdown(wake))), |
| 581 | ) |
| 582 | return subscription |
| 583 | }) |
| 584 | |
| 585 | const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream<Payload> => |
| 586 | Stream.unwrap( |