diff --git a/.changeset/eager-client-dispatch.md b/.changeset/eager-client-dispatch.md new file mode 100644 index 0000000..5f8907f --- /dev/null +++ b/.changeset/eager-client-dispatch.md @@ -0,0 +1,5 @@ +--- +"effect-machine": patch +--- + +Commit synchronous client transitions before `ActorClient.send` returns and notify Actor Atom subscribers without a timer delay. diff --git a/src/atom.ts b/src/atom.ts index 97a5714..1795199 100644 --- a/src/atom.ts +++ b/src/atom.ts @@ -36,6 +36,14 @@ export type ActorAcquireAtom, Cause.NoSuchElementError> >; +const stateAtom = ( + actor: ActorRef, +): Atom.Atom => + Atom.readable((get) => { + get.addFinalizer(actor.subscribe((state) => get.setSelf(state))); + return actor.client.getSnapshot(); + }); + /** Observe the current ActorRef for one ActorSystem key. */ export const fromSystem: { ( @@ -102,7 +110,7 @@ export const acquire: { export const make = ( actor: ActorRef, ): ActorAtom => { - const state = Atom.subscriptionRef(actor.state); + const state = stateAtom(actor); return Atom.writable( (get) => get(state), (_ctx, event) => actor.client.send(event), @@ -170,7 +178,7 @@ export const can: { actor: ActorRef, event: Event, ): CanAtom => { - const state = Atom.subscriptionRef(actor.state); + const state = stateAtom(actor); return Atom.make((get) => { get(state); return actor.can(event); diff --git a/src/internal/runtime.ts b/src/internal/runtime.ts index cfa08f3..6ab3843 100644 --- a/src/internal/runtime.ts +++ b/src/internal/runtime.ts @@ -550,6 +550,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function eventQueue, pendingRequests, exitDeferred, + () => loopFiberRef.current?.currentDispatcher?.flush(), ), stop: stop.pipe(Effect.provide(services)), start: start.pipe(Effect.provide(services)), @@ -568,6 +569,7 @@ const makeHandle = >, pendingRequests: Set<(error: ActorStoppedError) => Effect.Effect>, exitDeferred: Deferred.Deferred>, + flushLoop: () => void, ): RuntimeHandle => { const track = ( deferred: Deferred.Deferred, @@ -625,7 +627,10 @@ const makeHandle = { const stopped = Effect.runSync(Ref.get(stoppedRef)); - if (!stopped) Effect.runSync(Queue.offer(eventQueue, { _tag: "send", event })); + if (stopped) return; + Effect.runSync(Queue.offer(eventQueue, { _tag: "send", event })); + eventQueue.dispatcher.flush(); + flushLoop(); }, getState: SubscriptionRef.get(stateRef), stateRef, diff --git a/test/actor.test.ts b/test/actor.test.ts index b9af965..e1ace1e 100644 --- a/test/actor.test.ts +++ b/test/actor.test.ts @@ -1055,7 +1055,7 @@ describe("ActorRef", () => { // ============================================================================ describe("sendSync", () => { - it.scopedLive("sendSync sends event synchronously", () => + it.scopedLive("sendSync commits a synchronous transition before returning", () => Effect.gen(function* () { const machine = Machine.make({ state: TestState, @@ -1071,9 +1071,8 @@ describe("ActorRef", () => { const actor = yield* Machine.spawn(machine); yield* actor.start; actor.client.send(TestEvent.Start({ value: 7 })); - yield* yieldFibers; - const state = yield* actor.snapshot; + const state = actor.client.getSnapshot(); expect(state._tag).toBe("Active"); if (state._tag === "Active") { expect(state.value).toBe(7); @@ -1102,6 +1101,54 @@ describe("ActorRef", () => { expect(state._tag).toBe("Idle"); }), ); + + it.scopedLive("sendSync commits a transition with a synchronous Effect guard", () => + Effect.gen(function* () { + const machine = Machine.make({ + state: TestState, + event: TestEvent, + initial: TestState.Idle, + }).when( + TestState.Idle, + TestEvent.Start, + () => Effect.succeed(true), + ({ event }) => TestState.Active({ value: event.value }), + ); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + + actor.client.send(TestEvent.Start({ value: 9 })); + + expect(actor.client.getSnapshot()).toEqual(TestState.Active({ value: 9 })); + }), + ); + + it.scopedLive("sendSync preserves nested event order", () => + Effect.gen(function* () { + const machine = Machine.make({ + state: TestState, + event: TestEvent, + initial: TestState.Idle, + }) + .on(TestState.Idle, TestEvent.Start, ({ event }) => + TestState.Active({ value: event.value }), + ) + .on(TestState.Active, TestEvent.Stop, () => TestState.Done); + const actor = yield* Machine.spawn(machine); + yield* actor.start; + const states: Array = []; + const unsubscribe = actor.subscribe((state) => { + states.push(state._tag); + if (state._tag === "Active") actor.client.send(TestEvent.Stop); + }); + + actor.client.send(TestEvent.Start({ value: 1 })); + + expect(actor.client.getSnapshot()._tag).toBe("Done"); + expect(states).toEqual(["Active", "Done"]); + unsubscribe(); + }), + ); }); // ============================================================================ diff --git a/test/atom.test.ts b/test/atom.test.ts index 645b8d7..47b9318 100644 --- a/test/atom.test.ts +++ b/test/atom.test.ts @@ -43,8 +43,6 @@ it.scopedLive("make exposes actor state and sends events through Atom writes", ( }); registry.set(stateAtom, CounterEvent.Increment); - yield* actor.waitFor((state) => state.count === 1); - yield* Effect.yieldNow; expect(registry.get(stateAtom).count).toBe(1); expect(values).toEqual([0, 1]);