Standard library · Time and concurrency

events

An in-process event emitter: dotted topics with wildcards, one-shot handlers, and an optional worker thread.

import events

On this page

Register, emit, remove
Handle events on a worker thread
Structs as payloads
Emitting from a handler
Removing a component's handlers
Types
Functions

events is a small publish and subscribe hub for the inside of one program. You register handlers on a topic, and Emit calls every handler whose topic matches. The payload type is a parameter, so one emitter carries one kind of data: Emitter<int>, Emitter<str>, or an emitter of your own struct.

Topics are dot-separated words such as user.login. In a handler's topic, * stands for exactly one word and # for the rest of the topic, including nothing at all, so user.* matches user.login but not user.a.b, and order.# matches order, order.new and order.new.eu.

Things to know:

A handler is a lambda, and a lambda captures the variables around it by value. Share state with a pointer, a heap collection or a mut global, as the examples do.
Handlers run outside the lock, so a handler may call On, Off or Emit on the same emitter. Handlers added during an Emit first see the next event.
On returns an id; Off takes that id. Functions are not compared.
An async emitter handles events on the thread you give it. NewAsync makes the queue and starts no thread; run Drain on one thread of your own, and its handlers run one at a time, in the order the events were queued. Flush waits until the queue is empty and the last handler has returned; never call it from inside a handler.
Release an emitter with Free. For an async one, call Close and join the thread running Drain first, since Free does not wait for it.

Register, emit, remove

On and Once take a topic and a handler. OnAny receives every event. Emit returns how many handlers were called.

events-basic.salam

import events

mut e := events.New() as events.Emitter<int>
mut log := Vector {} as Vector<str>

func note(s: str):
    log.push(s)
end

func main:
    e = events.New() as events.Emitter<int>
    a := events.On(e, "user.login", (ev: events.Event<int>): note("a " + ev.topic + " " + ev.data) end)
    b := events.On(e, "user.*", (ev: events.Event<int>): note("b " + ev.topic + " " + ev.data) end)
    _c := events.Once(e, "user.login", (ev: events.Event<int>): note("once " + ev.data) end)
    _d := events.OnAny(e, (ev: events.Event<int>): note("any " + ev.topic) end)
    _f := events.On(e, "order.#", (ev: events.Event<int>): note("order " + ev.topic) end)

    println "count login", events.Count(e, "user.login")
    println "count user.a.b", events.Count(e, "user.a.b")
    println "count order", events.Count(e, "order")
    println "count order.x.y", events.Count(e, "order.x.y")
    println "calls", events.Emit(e, "user.login", 1)
    println "calls", events.Emit(e, "user.login", 2)
    println "calls", events.Emit(e, "user.logout", 3)
    println "calls", events.Emit(e, "order.new.eu", 4)
    println "calls", events.Emit(e, "other", 5)

    println "off a", events.Off(e, a)
    println "off a again", events.Off(e, a)
    println "off topic", events.OffTopic(e, "user.*")
    println "off b", events.Off(e, b)
    println "len", events.Len(e)
    println "calls", events.Emit(e, "user.login", 6)

    selfoff := events.On(e, "x", (ev: events.Event<int>): note("x " + ev.data) end)
    _adder := events.On(e, "x", (_ev: events.Event<int>):
        _id := events.On(e, "y", (_v: events.Event<int>): note("y") end)
        _ok := events.Off(e, selfoff)
    end)
    println "len before", events.Len(e)
    events.Emit(e, "x", 7)
    println "len after", events.Len(e)
    events.Emit(e, "x", 8)

    repeat log.len() in i:
        println log.get(i)
    end
    log.free()
    events.Free(e)
    ret 0
end

Outputcount login 4 count user.a.b 1 count order 2 count order.x.y 2 calls 4 calls 3 calls 2 calls 2 calls 1 off a true off a again false off topic 1 off b false len 2 calls 1 len before 4 len after 4 a user.login 1 b user.login 1 once 1 any user.login a user.login 2 b user.login 2 any user.login b user.logout 3 any user.logout any order.new.eu order order.new.eu any other any user.login any x x 7 any x

Handle events on a worker thread

NewAsync makes a queue of the given size, and Drain run on a thread of your own is its worker. EmitAsync puts an event on the queue and returns at once, and waits only when the queue is full. Close stops accepting events; Drain finishes what is queued and returns, so join the thread after it.

events-async.salam

import events
import mem

mut total := 0
mut last := 0
mut ordered := true
mut seen := 0

func add(ev: events.Event<int>):
    total = total + ev.data
    if ev.data != last + 1:
        ordered = false
    end
    last = ev.data
    seen = seen + 1
end

func worker(p: events.Emitter<int>*):
    events.Drain(p[0])
end

func main:
    e := events.NewAsync(4) as events.Emitter<int>
    box := mem.AllocateZeroed(sizeof(events.Emitter<int>) as u64) as events.Emitter<int>*
    box[0] = e
    th := spawn(worker, box)
    _id := events.On(e, "tick", (ev: events.Event<int>): add(ev) end)

    println "first queued", events.EmitAsync(e, "tick", 1)
    repeat 2 to 200 in i:
        _ok := events.EmitAsync(e, "tick", i)
    end
    events.Flush(e)
    println "seen", seen
    println "total", total
    println "ordered", ordered
    println "pending", events.Pending(e)

    _ok := events.EmitAsync(e, "tick", 201)
    _ok2 := events.EmitAsync(e, "other", 999)
    events.Close(e)
    join(th)
    println "after close seen", seen
    println "total", total
    println "emit after close", events.EmitAsync(e, "tick", 5)
    events.Close(e)

    plain := events.New() as events.Emitter<int>
    println "async on sync emitter", events.EmitAsync(plain, "x", 1)
    events.Free(plain)
    events.Free(e)
    mem.Free(box as void*)
    ret 0
end

Outputfirst queued true seen 200 total 20100 ordered true pending 0 after close seen 201 total 20301 emit after close false async on sync emitter false

Structs as payloads

The payload type is a parameter, so a struct travels as easily as an int: Emitter<Order> hands every handler a whole Order in ev.data. OnAny sees every topic, including ones nobody else listens to, which makes it a good place for an audit counter or a log line. Emit returns how many handlers ran, so order.shipped below reports 1 (only the audit handler).

events-guide-payload.salam

import events

mut revenue := 0
mut audited := 0

struct Order:
    pub id: int = 0
    pub item: str = ""
    pub qty: int = 0
end

func main:
    bus := events.New() as events.Emitter<Order>
    _a := events.On(bus, "order.paid", (ev: events.Event<Order>):
        revenue += ev.data.qty * 10
        println "paid", ev.data.id, ev.data.item
    end)
    _b := events.On(bus, "order.cancelled", (ev: events.Event<Order>):
        println "cancelled", ev.data.id
    end)
    _c := events.OnAny(bus, (_ev: events.Event<Order>): audited += 1 end)

    println events.Emit(bus, "order.paid", Order { id = 1, item = "tea", qty = 3 })
    println events.Emit(bus, "order.cancelled", Order { id = 2 })
    println events.Emit(bus, "order.shipped", Order { id = 1 })
    println "revenue", revenue, "audited", audited
    events.Free(bus)
end

Outputpaid 1 tea 2 cancelled 2 2 1 revenue 30 audited 3

Emitting from a handler

Handlers run outside the emitter's lock, so a handler may emit again on the same emitter. The nested Emit runs its handlers at once, before the outer handler returns, and the outer Emit still counts only its own handlers. Keep such chains short and make sure they end, because nothing stops a handler that emits its own topic unconditionally.

events-guide-cascade.salam

import events

mut bus := events.New() as events.Emitter<int>
mut depth := 0

func main:
    bus = events.New() as events.Emitter<int>
    _a := events.On(bus, "count", (ev: events.Event<int>):
        println "count", ev.data
        if ev.data < 3:
            depth += 1
            _n := events.Emit(bus, "count", ev.data + 1)
        end
    end)
    _b := events.On(bus, "count", (ev: events.Event<int>):
        if ev.data == 3:
            _n := events.Emit(bus, "done", ev.data)
        end
    end)
    _c := events.On(bus, "done", (ev: events.Event<int>): println "done at", ev.data end)

    println "handlers called", events.Emit(bus, "count", 1)
    println "nested emits", depth
    events.Free(bus)
end

Outputcount 1 count 2 count 3 done at 3 handlers called 2 nested emits 2

Removing a component's handlers

A component that registers handlers should remember their ids and give them back when it goes away, otherwise the emitter keeps calling code that is no longer meant to run. Off takes one id, and Len after teardown is an easy check that nothing was left behind. Emitting a topic with no handlers is not an error: it simply returns 0.

events-guide-teardown.salam

import events

mut clicks := 0

struct Panel:
    pub ids: Vector<int> = Vector {}
end

func openPanel(bus: events.Emitter<str>): Panel:
    mut p := Panel {}
    p.ids.push(events.On(bus, "ui.click", (_ev: events.Event<str>): clicks += 1 end))
    p.ids.push(events.On(bus, "ui.key.*", (_ev: events.Event<str>): clicks += 10 end))
    ret p
end

func closePanel(bus: events.Emitter<str>, p: Panel):
    each id in p.ids:
        _ok := events.Off(bus, id)
    end
    p.ids.free()
end

func main:
    bus := events.New() as events.Emitter<str>
    p := openPanel(bus)
    println "open", events.Len(bus), events.Emit(bus, "ui.click", "a"), events.Emit(bus, "ui.key.enter", "b")
    closePanel(bus, p)
    println "closed", events.Len(bus), events.Emit(bus, "ui.click", "c")
    println "clicks", clicks
    events.Free(bus)
end

Outputopen 2 1 1 closed 0 0 clicks 11

Types

`Event`

struct Event<T>

One event: the topic it was emitted on and its data.

Name	Type	Description
`topic`	`str = ""`	The full topic, such as `user.login`.
`data`	`T`	The payload given to `Emit`.

`Handler`

struct Handler<T>

A registered handler with its id, topic pattern and callback.

Name	Type	Description
`id`	`int = 0`	The id that `On` returned.
`pattern`	`str = ""`	The topic pattern the handler was registered with.
`once`	`bool = false`	The handler is removed after its first call.
`cb`	`func(Event)`	The function that is called with the event.

`Emitter`

struct Emitter<T>

One emitter. Copies of it share the same handlers and queue.

Name	Type	Description
`hs`	`Handler*`	The handler table.
`mu`	`sync.Mutex`	The lock around the table and the queue.
`ctl`	`i64*`	Counters and flags of the queue.
`async`	`bool = false`	The emitter has a queue, so it needs a thread running `Drain`.
`qbuf`	`Event*`	The queue of events waiting for the worker.
`cv`	`sync.CondVar`	The condition that wakes the worker and the waiters.

Functions

`New`

func New<T>(): Emitter<T>

Creates a synchronous emitter; Emit calls handlers on the caller's thread.

events-Basic.salam

import events

mut e := events.New() as events.Emitter<int>
mut log := Vector {} as Vector<str>

func note(s: str):
    log.push(s)
end

func main:
    e = events.New() as events.Emitter<int>
    a := events.On(e, "user.login", (ev: events.Event<int>): note("a " + ev.topic + " " + ev.data) end)
    b := events.On(e, "user.*", (ev: events.Event<int>): note("b " + ev.topic + " " + ev.data) end)
    _c := events.Once(e, "user.login", (ev: events.Event<int>): note("once " + ev.data) end)
    _d := events.OnAny(e, (ev: events.Event<int>): note("any " + ev.topic) end)
    _f := events.On(e, "order.#", (ev: events.Event<int>): note("order " + ev.topic) end)

    println "count login", events.Count(e, "user.login")
    println "count user.a.b", events.Count(e, "user.a.b")
    println "count order", events.Count(e, "order")
    println "count order.x.y", events.Count(e, "order.x.y")
    println "calls", events.Emit(e, "user.login", 1)
    println "calls", events.Emit(e, "user.login", 2)
    println "calls", events.Emit(e, "user.logout", 3)
    println "calls", events.Emit(e, "order.new.eu", 4)
    println "calls", events.Emit(e, "other", 5)

    println "off a", events.Off(e, a)
    println "off a again", events.Off(e, a)
    println "off topic", events.OffTopic(e, "user.*")
    println "off b", events.Off(e, b)
    println "len", events.Len(e)
    println "calls", events.Emit(e, "user.login", 6)

    selfoff := events.On(e, "x", (ev: events.Event<int>): note("x " + ev.data) end)
    _adder := events.On(e, "x", (_ev: events.Event<int>):
        _id := events.On(e, "y", (_v: events.Event<int>): note("y") end)
        _ok := events.Off(e, selfoff)
    end)
    println "len before", events.Len(e)
    events.Emit(e, "x", 7)
    println "len after", events.Len(e)
    events.Emit(e, "x", 8)

    repeat log.len() in i:
        println log.get(i)
    end
    log.free()
    events.Free(e)
    ret 0
end