Event Bus

Event Bus

Publish and subscribe to events for observability — monitoring runtime and application activity and reacting to it.

Use the event bus for observation only: monitoring, logging, metrics, and reactive side effects. It is a best-effort publish/subscribe channel, not a reliable transport — do not build business logic on it or depend on it for guaranteed delivery. For business-critical messaging use process messaging (`process.send`), channels, or the [message queue](lua/storage/queue.md).

Loading

local events = require("events")

Subscribing to Events

Subscribe to events from the event bus:

-- Subscribe to all order events
local sub, err = events.subscribe("orders.*")
if err then
    return nil, err
end

-- Subscribe to specific event kind
local sub = events.subscribe("users", "user.created")

-- Subscribe to all events from a system
local sub = events.subscribe("payments")

-- Process events
local ch = sub:channel()
while true do
    local evt, ok = ch:receive()
    if not ok then break end

    logger:info("Received event", {
        system = evt.system,
        kind = evt.kind,
        path = evt.path
    })
    handle_event(evt)
end

Parameter	Type	Description
system	string	System pattern (supports wildcards like “test.*“)
kind	string	Event kind filter (optional)

Returns: Subscription, error

Sending Events

Send an event to the event bus:

-- Send order created event
local ok, err = events.send("orders", "order.created", "/orders/123", {
    order_id = "123",
    customer_id = "456",
    total = 99.99
})
if err then
    return nil, err
end

-- Send user event
events.send("users", "user.registered", "/users/" .. user.id, {
    user_id = user.id,
    email = user.email,
    created_at = time.now():format("2006-01-02T15:04:05Z07:00")
})

-- Send payment event
events.send("payments", "payment.completed", "/payments/" .. payment.id, {
    payment_id = payment.id,
    order_id = payment.order_id,
    amount = payment.amount,
    method = payment.method
})

-- Send without data
events.send("system", "heartbeat", "/health")

Parameter	Type	Description
system	string	System identifier
kind	string	Event kind/type
path	string	Event path for routing
data	any	Event payload (optional)

Returns: boolean, error

Subscription Methods

Getting the Channel

Get the channel for receiving events:

local ch = sub:channel()

local evt, ok = ch:receive()
if ok then
    print("System:", evt.system)
    print("Kind:", evt.kind)
    print("Path:", evt.path)
    print("Data:", json.encode(evt.data))
end

Event fields: system, kind, path, data

Closing Subscription

Unsubscribe and close the channel:

sub:close()

Permissions

Action	Resource	Description
events.subscribe	system	Subscribe to events from a system
events.send	system	Send events to a system

Errors

Condition	Kind	Retryable
Empty system	errors.INVALID	no
Empty kind	errors.INVALID	no
Empty path	errors.INVALID	no
Policy denied	errors.INVALID	no

See Error Handling for working with errors.