messaging

The messaging.example microservice demonstrates the framework’s transport patterns over the bus:

Load-balanced unicast — many replicas subscribed to the same endpoint, one chosen at random per request.
Multicast — many replicas all respond to the same request; the caller receives every response.
Direct addressing — request a specific instance by ID, bypassing the load balancer.
Distributed cache — a NATS-backed key-value cache shared across all replicas of a service.

The main app registers three replicas of this service so the patterns are visible in a single browser refresh:

app.Add(
    // ...
    messaging.NewService(),
    messaging.NewService(),
    messaging.NewService(),
    // ...
)

Try It

With the examples app running:

Walk through every pattern in one page: http://localhost:8080/messaging.example/home

Processed by: 4l284tgjfk

Unicast
GET https://messaging.example/default-queue
> DefaultQueue 4l284tgjfk

Direct addressing unicast
GET https://4l284tgjfk.messaging.example/default-queue
> DefaultQueue 4l284tgjfk

Multicast
GET https://messaging.example/no-queue
> NoQueue 4kfei93btu
> NoQueue ba62j2gjjp
> NoQueue 4l284tgjfk

Direct addressing multicast
GET https://4l284tgjfk.messaging.example/no-queue
> NoQueue 4l284tgjfk

Refresh the page to try again

Refresh repeatedly:

The processor ID may change (load balancing on /home).
The single unicast responder may change.
The order of the multicast responses may change.

Distributed cache:
Store: http://localhost:8080/messaging.example/cache-store?key=foo&value=bar
```
key: foo
value: bar

Stored by qtshc434b7
```
Load: http://localhost:8080/messaging.example/cache-load?key=foo
```
key: foo
found: yes
value: bar

Loaded by ucarmsii56
```
Refresh the load URL a few times and note that whichever replica handles the request, the value is found — the cache is shared across all three replicas via NATS.
Miss: http://localhost:8080/messaging.example/cache-load?key=fox
```
key: fox
found: no

Loaded by pv7lqgeu98
```

Code Walkthrough

Source: exampleservices/messaging/service.go.

The Service struct is bare — the per-replica state that matters (the instance ID, the distributed cache handle) is provided by the embedded *Intermediate:

type Service struct {
    *Intermediate // IMPORTANT: Do not remove
}

OnStartup and OnShutdown are empty.

Wiring (intermediate.go)

The constructor and the five web-endpoint subscriptions, from the generated intermediate.go:

func NewService() *Service {
    svc := &Service{}
    svc.Intermediate = NewIntermediate(svc)
    return svc
}

svc.Subscribe("Home",         svc.Home,         sub.At(...), sub.Web())
svc.Subscribe("NoQueue",      svc.NoQueue,      sub.At(...), sub.Web(), sub.NoQueue())
svc.Subscribe("DefaultQueue", svc.DefaultQueue, sub.At(...), sub.Web())
svc.Subscribe("CacheLoad",    svc.CacheLoad,    sub.At(...), sub.Web())
svc.Subscribe("CacheStore",   svc.CacheStore,   sub.At(...), sub.Web())

The pivotal difference is on the NoQueue subscription: the extra sub.NoQueue() option puts every replica in its own NATS queue group, so each replica receives every message published to that subject. Without it (the default — see DefaultQueue), the framework places all replicas in a shared queue group and NATS load-balances incoming messages across them, so only one replica responds per request.

That single option is the entire difference between unicast and multicast for the subscriber side.

Home

The /home handler stitches together all four request patterns in one page so they can be compared back-to-back. The handler is long; the four key request calls are:

// 1. Standard unicast — one of the three replicas responds (load-balanced by NATS).
res, err := svc.Request(r.Context(), pub.GET("https://messaging.example/default-queue"))
// ...
responderID := frame.Of(res).FromID()

// 2. Direct addressing unicast — the specific instance we just heard from.
res, err = svc.Request(r.Context(), pub.GET("https://"+responderID+".messaging.example/default-queue"))

// 3. Multicast — all three replicas respond. svc.Publish returns a channel.
ch := svc.Publish(r.Context(), pub.GET("https://messaging.example/no-queue"))
for i := range ch {
    res, err := i.Get()
    // ...
}

// 4. Direct addressing multicast — only the specific instance responds,
//    even though the endpoint is multicast-shaped.
ch = svc.Publish(r.Context(), pub.GET("https://"+lastResponderID+".messaging.example/no-queue"))

Two things to notice:

svc.Request returns a single response; svc.Publish returns a channel of responses. The choice of Request vs Publish on the caller side is independent of whether the subscription uses sub.NoQueue(). They compose: a Publish to a unicast endpoint yields one response; a Request to a multicast endpoint takes the first one. The example pairs them in the conventional way.
Direct addressing. Prefixing the hostname with <instance-id>. (4l284tgjfk.messaging.example) routes the request straight to that instance, regardless of subscription type. This is why the fourth section returns only one response: even though /no-queue is multicast, the direct address narrows the audience to a single subscriber.

NoQueue and DefaultQueue

The two endpoints are tiny — they’re identification beacons that report which replica handled the request. The subscription options (set up in Wiring) are where the behavior actually lives.

func (svc *Service) NoQueue(w http.ResponseWriter, r *http.Request) (err error) {
    w.Header().Set("Content-Type", "text/plain")
    w.Write([]byte("NoQueue " + svc.ID()))
    return nil
}

func (svc *Service) DefaultQueue(w http.ResponseWriter, r *http.Request) (err error) {
    w.Header().Set("Content-Type", "text/plain")
    w.Write([]byte("DefaultQueue " + svc.ID()))
    return nil
}

svc.ID() returns this replica’s instance ID — the same value seen as Microbus-From-Id in hello’s /echo.

CacheLoad and CacheStore

func (svc *Service) CacheStore(w http.ResponseWriter, r *http.Request) (err error) {
    key := r.URL.Query().Get("key")
    if key == "" {
        return errors.New("missing key")
    }
    value := r.URL.Query().Get("value")
    if value == "" {
        return errors.New("missing value")
    }
    err = svc.DistribCache().Set(r.Context(), key, []byte(value))
    if err != nil {
        return errors.Trace(err)
    }
    // ... write a "Stored by <id>" response ...
}

func (svc *Service) CacheLoad(w http.ResponseWriter, r *http.Request) (err error) {
    key := r.URL.Query().Get("key")
    if key == "" {
        return errors.New("missing key")
    }
    var value []byte
    ok, err := svc.DistribCache().Get(r.Context(), key, &value)
    if err != nil {
        return errors.Trace(err)
    }
    // ... write a "Loaded by <id>" response, possibly with the value ...
}

svc.DistribCache() returns a per-service handle to a NATS-backed key-value cache that is shared across every replica of this microservice. Storing on one replica and loading from another succeeds because the store sits in NATS, not in any one process’s memory. The Stored by / Loaded by IDs in the output make it visible that the two operations were handled by different replicas.

This is a deliberate, scoped sharing — the cache is shared within a service (across its replicas) but not across services.