Untyped Dispatch

Untyped Dispatch

Code that routes messages without naming their types — a job queue, an inbox, a scheduler — sends envelopes. Request.envelope, Notification.envelope, and Stream.envelope seal a message with its kind; Mediator.dispatch, Mediator.broadcast, and Mediator.streamEnvelope answer an Answer holding the outcome, whether it failed, and a one-line summary. The kind that sealed a message reads the typed outcome back with answerOf, and a stream's items with itemOf:

module Ex08Envelopes

import Std.Io as Io
import PuduLangMediator.Context as Context
import PuduLangMediator.Mediator as Mediator
import PuduLangMediator.Message as Message
import PuduLangMediator as Messaging
import PuduLangMediator.Notification as Notification
import PuduLangMediator.Registration as Registration
import PuduLangMediator.Request as Request
import PuduLangMediator.Stream as Stream

export fn main() -> Int {
  let square: Request.Kind[Int, Int, Str] = Request.kind("math.square")
  let ticked: Notification.Kind[Int, Str] = Notification.kind("clock.ticked")
  let digits: Stream.Kind[Int, Int, Str] = Stream.kind("math.digits")
  let unknown: Request.Kind[Int, Int, Str] = Request.kind("math.cube")
  let mediator = match Mediator.build([
      Registration.handler(&square, |n: Int, _context: Context.Context| Ok(n * n)),
      Registration.streamHandler(&digits, fn(n: Int, _context: Context.Context, sink: Stream.Sink[Int]) -> Messaging.Outcome[(), Str] {
          var rest = n
          var found: Array[Int] = []
          while rest > 0 {
            found = [rest % 10].concat(found)
            rest = rest / 10
          }
          Stream.each(&found, sink)
        })
    ]) {
    case Ok(built) => built
    case Err(invalid) => panic(Mediator.explain(&invalid))
  }
  let queue = [Request.envelope(&square, 12), Request.envelope(&unknown, 3), Notification.envelope(&ticked, 1)]
  var answers: Array[Message.Answer] = []
  for envelope in queue {
    let answer = match envelope.info.shape {
      case Message.Event => Mediator.broadcast(&mediator, envelope)
      case _ => Mediator.dispatch(&mediator, envelope)
    }
    let _said = Io.writeLine(envelope.info.name + " " + envelope.summary + " -> " + answer.summary)
    answers = answers.push(answer)
  }
  let streamed = Mediator.streamEnvelope(&mediator, Stream.envelope(&digits, 2026), Context.create(), fn(packed: fn() -> ()) -> Bool {
      if let Some(digit) = Stream.itemOf(&digits, &packed) { let _said = Io.writeLine("digit " + show(digit)) }
      true
    })
  let squared = Request.answerOf(&square, &answers[0])
  let _typed = Io.writeLine("read back: " + show(squared) + ", stream: " + streamed.summary)
  if squared == Some(Ok(144)) && Request.answerOf(&unknown, &answers[1]) == Some(Err(Messaging.Unhandled("math.cube"))) { 0 } else { 1 }
}

Check it and run it:

pudu check src/Ex08Envelopes.pudu
pudu run src/Ex08Envelopes.pudu

Output:

math.square 12 -> Ok(144)
math.cube 3 -> No handler is registered for 'math.cube'.
clock.ticked 1 -> Ok(())
digit 2
digit 0
digit 2
digit 6
read back: Some(Ok(144)), stream: Ok(())

An envelope of a kind nobody registered answers Unhandled; a notification nobody hears succeeds. Values cross the untyped boundary through the kind's own slots rather than casts, so an answer never reads back as another kind's type.

Related

• Senders Publishers And Testing
• Reference Index