diff --git a/src/fuwaroid.mbt b/src/fuwaroid.mbt index 0cd542c..6e2cb99 100644 --- a/src/fuwaroid.mbt +++ b/src/fuwaroid.mbt @@ -207,17 +207,6 @@ fn LoopStop::fresh() -> LoopStop { { reason: None, cleanup_error: None, task: None, diag: Diag::fresh(), } } -///| -/// No-op query handler for `spawn_fold`. -#inline -fn[State, Cmd] fold_receipt_query( - _ctx : Ctx[Cmd, Unit, Unit], - state : State, - _query : Unit, -) -> (State, Unit) { - (state, ()) -} - ///| /// Shared admission path for `tell` and `ask`. #inline @@ -302,13 +291,24 @@ pub fn[State, Cmd] Fuwaroid::spawn_fold( on_cmd~ : (Ctx[Cmd, Unit, Unit], State, Cmd) -> State, mailbox? : Mailbox = Mailbox::Unbounded, ) -> Fuwaroid[Cmd, Unit, Unit] { - Fuwaroid::spawn( - group~, - init~, - on_cmd~, - on_query=fold_receipt_query, - mailbox~, - ) + let kind : @aqueue.Kind = match mailbox { + Unbounded => @aqueue.Unbounded + Bounded(n) => + match mailbox_capacity_error(n) { + Some(message) => + abort("fuwaroid: invalid mailbox configuration: \{message}") + None => @aqueue.Blocking(n) + } + } + let queue : @aqueue.Queue[Envelope[Cmd, Unit, Unit]] = @aqueue.Queue(kind~) + let stop = LoopStop::fresh() + let fuwaroid : Fuwaroid[Cmd, Unit, Unit] = { mailbox: queue, stop, } + let ctx : Ctx[Cmd, Unit, Unit] = { group, address: fuwaroid, } + let task = group.spawn(no_wait=true, () => { + fuwaroid_loop_fold(ctx, init, on_cmd, queue, stop~) + }) + stop.task = Some(task) + fuwaroid } ///| @@ -349,6 +349,66 @@ fn[Cmd, Query, Reply] abandon_envelope( } } +///| +async fn[State, Cmd] fuwaroid_loop_fold( + ctx : Ctx[Cmd, Unit, Unit], + init : State, + on_cmd : (Ctx[Cmd, Unit, Unit], State, Cmd) -> State, + queue : @aqueue.Queue[Envelope[Cmd, Unit, Unit]], + stop~ : LoopStop, +) -> Unit { + match @async.handle_cancellation(() => { + fuwaroid_serve_fold(ctx, init, on_cmd, queue, stop~) + }) { + Some(_) => () + None => { + fuwaroid_cleanup(queue, Cancelled, stop) + @async.pause() + } + } +} + +///| +#locals(on_cmd) +async fn[State, Cmd] fuwaroid_serve_fold( + ctx : Ctx[Cmd, Unit, Unit], + init : State, + on_cmd : (Ctx[Cmd, Unit, Unit], State, Cmd) -> State, + queue : @aqueue.Queue[Envelope[Cmd, Unit, Unit]], + stop~ : LoopStop, +) -> Unit { + let reason : StopReason = try { + for envelope = queue.get(), + state = init, + served = 0 { + let state = match envelope { + Tell(cmd) => on_cmd(ctx, state, cmd) + Ask(_, reply) => { + let _ = reply.try_put(()) catch { _ => false } + state + } + } + let served = served + 1 + stop.diag.processed += 1L + + if served >= fuwaroid_yield_batch { + @async.pause() + continue queue.get(), state, 0 + } + + match queue.try_get() { + Some(next) => continue next, state, served + None => continue queue.get(), state, served + } + } + Graceful + } catch { + @aqueue.QueueAlreadyClosed => Graceful + original => Failed(original) + } + fuwaroid_cleanup(queue, reason, stop) +} + ///| /// Cancellation is handled separately from ordinary loop errors. async fn[State, Cmd, Query, Reply] fuwaroid_loop(