mule - v0.1.0

mule/runtime/producer

Types

Message

</> 
pub type Message {
  NotificationReceived(notifier.Notification)
  Dispatch
  Poll
  Refresh
  Check(reply: process.Subject(engine.QueueState))
  InitiateShutdown(
    drained: process.Subject(String),
    reply: process.Subject(Nil),
  )
  MonitorDown(process.Down)
  ParentExit
}

Constructors

State

</> 
pub type State {
  State(
    config: config.Config,
    dispatch_timer: option.Option(process.Timer),
    drained: option.Option(process.Subject(String)),
    foreman: foreman.Foreman,
    meta: engine.QueueMeta,
    refresh_timer: option.Option(process.Timer),
    running: dict.Dict(
      process.Monitor,
      #(process.Pid, executor.Executor),
    ),
    self_subject: process.Subject(Message),
  )
}

Constructors

Values

check

</> 
pub fn check(
  producer producer: process.Subject(Message),
) -> engine.QueueState

Synchronous queue introspection (producer.ex check/1). Panics when the producer is down or does not reply in time — mule.check_queue wraps this in a rescue, the mule.ex safe_check analog. Uses shutdown_call_timeout because the watchman’s timed-out drain calls this too, and the queue supervisor budgets the watchman’s shutdown as grace + two such calls.

initiate_shutdown

</> 
pub fn initiate_shutdown(
  producer producer: process.Subject(Message),
  drained drained: process.Subject(String),
) -> Nil

Ask the producer to begin draining (producer.ex shutdown/1), telling it to send the queue name on drained once no jobs are running. Panics when the producer is down — the watchman wraps this in a rescue, mirroring watchman.ex’s catch :exit.

shutdown_call_timeout

</> 
pub const shutdown_call_timeout: Int

How long initiate_shutdown waits for the producer’s reply — GenServer.call’s 5000ms default, which watchman.ex’s Producer.shutdown/1 rides. A producer mid-fetch against a loaded database can hold the call past a shorter window, and a timeout here aborts the whole graceful drain. The queue supervisor’s watchman child-spec margin accounts for it.

start

</> 
pub fn start(
  config config: config.Config,
  limit limit: Int,
  queue queue: String,
  paused paused: Bool,
) -> Result(
  actor.Started(process.Subject(Message)),
  actor.StartError,
)

subject

</> 
pub fn subject(
  instance_name instance_name: String,
  queue queue: String,
) -> Result(process.Subject(Message), Nil)

The registered producer subject for queue under instance_name, when one is alive — the typed accessor over the registry’s {:producer, q} row (registry handles are stored untyped; this module owns the type).

wake_local_producers

</> 
pub fn wake_local_producers(
  instance_name instance_name: String,
) -> Nil

Wake every producer registered locally for instance_name, bypassing the notifier — the Stager’s local-mode notify (stager.ex notify_queues’s :local clause sends the insert message straight to the registry’s producer pids). Lives here because this module owns the registry row’s handle type.

✨ Search Document