Getting started¶

By the end of this you will have a bus running on channels, a subscription receiving events, and an understanding of the one call that matters at start-up.

No broker is involved. The in-memory backend is a real backend rather than a test double, so everything here is the same code path a NATS-backed bus takes.

1. Build a bus, with its handler¶

import (
    "gitlab.com/phpboyscout/go/messaging"
    "gitlab.com/phpboyscout/go/messaging/memory"
)

bus, err := messaging.New(memory.New(), messaging.Settings{
    Subscriptions: []messaging.SubscriptionSpec{
        {
            Name:    "greeter",
            Pattern: "greeting.>",
            Shape:   messaging.Fanout,
            Handler: func(ctx context.Context, d messaging.Delivery) error {
                log.Info("received", "subject", d.Subject, "id", d.Event.ID)

                return nil
            },
        },
    },
})
if err != nil {
    return err
}

The handler is part of the subscription's declaration, not a separate step: a subscription without one is a construction error, so there is no window where a subscription exists with nothing to run against it.

d.Subject is the concrete subject, not the pattern that selected it. A subscription on greeting.> needs to know whether it got greeting.en or greeting.fr.

Two things have not happened yet: nothing is connected, and nothing is receiving. Construction registers your subscriptions as intent, which is why it does not need a broker to be reachable.

2. Start, and gate once¶

if err := bus.Start(ctx); err != nil {
    return err
}
defer bus.Stop(context.Background())

select {
case <-bus.Ready():
    // publish and consume freely from here
case <-time.After(30 * time.Second):
    return errors.New("the bus did not become ready")
}

This is the call that matters. Ready() closes only once every subscription you registered at construction exists on the connection, so a publisher never has to know whether any subscriber is ready — only whether the bus is.

The budget is yours to choose, because how long your service should wait for messaging before declaring itself broken is a property of your service and not of this module.

3. Publish¶

err := bus.Publish(ctx, "greeting.en", cloudevents.Event{
    SpecVersion: "1.0",
    ID:          "01JQ...",
    Source:      "urn:example:tutorial",
    Type:        "com.example.greeting.v1",
    Data:        []byte(`{"text":"hello"}`),
})

Publishing before Ready() closes returns an error, and the message is not held and delivered later. That matters more than it looks: "it was rejected" and "it was quietly held and delivered afterwards" are opposite contracts that look identical from the publisher's side.

What you have¶

A bus on channels. To put it on NATS, swap memory.New() for the NATS backend and change nothing else — that is the whole point of the exercise.

Next: wire a bus into a controller, which is how it looks in a real service.