events

The eventsource.example and eventsink.example microservices demonstrate how events reverse the dependency between two microservices. The event source publishes events without knowing who (if anyone) will consume them; the event sink subscribes to those events without the source needing to be told about it.

In this example the source is a mock user-registration service. It fires two events per registration attempt:

  • OnAllowRegister — a synchronous event whose subscribers can veto the registration (multicast request/response — the source waits for every subscriber).
  • OnRegistered — an asynchronous fire-and-forget event announcing that a registration succeeded.

The sink is one filter implementation: it disallows gmail.com and hotmail.com addresses and remembers what got registered. Additional filter providers could be added later — by writing a new sink microservice — without changing a line of the source’s code.

Try It

With the examples app running, attempt these URLs in order:

Code Walkthrough — Source

Source: exampleservices/eventsource/service.go.

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

The service holds no state. Configs, downstream calls, and subscriber knowledge are all absent — which is the point. OnStartup and OnShutdown are empty.

Wiring (intermediate.go)

The only thing wired in intermediate.go is the Register functional endpoint:

func NewService() *Service {
    svc := &Service{}
    svc.Intermediate = NewIntermediate(svc)
    return svc
}
svc.Subscribe(
    "Register", svc.doRegister,
    sub.At(eventsourceapi.Register.Method, eventsourceapi.Register.Route),
    sub.Description(`Register attempts to register a new user.`),
    sub.Function(eventsourceapi.RegisterIn{}, eventsourceapi.RegisterOut{}),
)

Notice what’s not here: no list of subscribers, no event-bus configuration, no knowledge of eventsink at all. The events the source declares (OnAllowRegister, OnRegistered) live in the eventsourceapi package as part of the source’s public API and are fired through the generated eventsourceapi.NewMulticastTrigger(svc) helper. Subscribers wire themselves up on the other side.

Register

func (svc *Service) Register(ctx context.Context, email string) (allowed bool, err error) {
    // Trigger a synchronous event to check if any event sink blocks the registration
    for r := range eventsourceapi.NewMulticastTrigger(svc).OnAllowRegister(ctx, email) {
        allowed, err := r.Get()
        if err != nil {
            return false, errors.Trace(err)
        }
        if !allowed {
            return false, nil
        }
    }

    // OK to register the user
    // ...

    // Trigger an asynchronous fire-and-forget event to inform all event sinks of the new registration
    svc.Go(ctx, func(ctx context.Context) (err error) {
        eventsourceapi.NewMulticastTrigger(svc).OnRegistered(ctx, email)
        return nil
    })

    return true, nil
}

Two different event shapes, both delivered through the same NewMulticastTrigger(svc) helper:

  • Synchronous, with veto. OnAllowRegister returns a channel of responses — one per subscriber. The handler waits for every subscriber’s verdict and short-circuits on the first false. If no sink is wired up, the loop simply exits with allowed still its zero value (false); in this example the calling test fixtures and the demo app always include a sink.
  • Asynchronous, fire-and-forget. OnRegistered is fired inside svc.Go(ctx, ...), which runs the closure on a tracked goroutine that the framework will drain on shutdown. The result channel of OnRegistered is not even read — subscribers’ responses are dropped. This is the right shape for a notification that should not block the caller.

The Register handler never imports eventsink. If a second filter provider is added, neither the source nor the existing sink changes.

Code Walkthrough — Sink

Source: exampleservices/eventsink/service.go.

var (
    // Emulated data store - keyed by plane for test isolation
    mux           sync.Mutex
    registrations = make(map[string][]string)
)

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

The sink keeps a process-global mutex-guarded map of registrations. It is keyed by plane — the framework’s concept for test isolation between parallel test runs, accessed via svc.Plane(). In production every replica would naturally share mux and registrations only within its own process; this example just demonstrates the in-memory side, not the persistence story.

Wiring (intermediate.go)

Two subscriptions are made — both via the eventsourceapi.NewHook(svc) mechanism — and one functional endpoint to expose the recorded list:

svc.Subscribe(
    "Registered", svc.doRegistered,
    sub.At(eventsinkapi.Registered.Method, eventsinkapi.Registered.Route),
    sub.Description(`Registered returns the list of registered users.`),
    sub.Function(eventsinkapi.RegisteredIn{}, eventsinkapi.RegisteredOut{}),
)

// Inbound event sinks
eventsourceapi.NewHook(svc).OnAllowRegister(svc.OnAllowRegister)
eventsourceapi.NewHook(svc).OnRegistered(svc.OnRegistered)

eventsourceapi.NewHook(svc) is the generated subscriber-side counterpart to the source’s NewMulticastTrigger. It binds the sink’s local handler methods (svc.OnAllowRegister, svc.OnRegistered) to the source’s event subjects. This is where the dependency arrow lives: the sink imports eventsourceapi, not the other way around.

OnAllowRegister

func (svc *Service) OnAllowRegister(ctx context.Context, email string) (allow bool, err error) {
    svc.LogInfo(ctx, "OnAllowRegister", "email", email)
    address, err := mail.ParseAddress(strings.ToLower(email))
    if err != nil {
        return false, nil
    }
    email = address.Address
    if strings.HasSuffix(email, "@gmail.com") || strings.HasSuffix(email, ".gmail.com") ||
        strings.HasSuffix(email, "@hotmail.com") || strings.HasSuffix(email, ".hotmail.com") {
        return false, nil
    }
    plane := svc.Plane()
    mux.Lock()
    taken := slices.Contains(registrations[plane], email)
    mux.Unlock()
    if taken {
        return false, nil
    }
    return true, nil
}

The veto handler. Returns false for malformed addresses, gmail/hotmail domains, and already-registered emails; otherwise true. Since the source treats any false as a deny, only one sink needs to object to block a registration — even if other filter providers approve.

The handler returns (false, nil) for “deny” rather than (false, err). Errors are reserved for actual failures; a policy decision to deny is a successful answer with the value false.

OnRegistered

func (svc *Service) OnRegistered(ctx context.Context, email string) (err error) {
    svc.LogInfo(ctx, "OnRegistered", "email", email)
    plane := svc.Plane()
    mux.Lock()
    registrations[plane] = append(registrations[plane], strings.ToLower(email))
    mux.Unlock()
    return nil
}

The notification handler. Since the source fires this asynchronously and ignores the return channel, this handler’s error return is effectively dropped — it exists for symmetry with the other handler shape, and for logging on the sink side.

Registered

func (svc *Service) Registered(ctx context.Context) (emails []string, err error) {
    plane := svc.Plane()
    emails = []string{}
    mux.Lock()
    emails = append(emails, registrations[plane]...)
    mux.Unlock()
    return emails, nil
}

A plain functional endpoint exposing the recorded list, so the /registered browser URL works. Allocates a fresh slice rather than returning the underlying one — the caller may iterate it after the lock is released.

See Also

  • Events — the concept and motivation for reversed dependencies between services.
  • Multicast — the multicast transport that backs every event delivery.
  • Messaging example — multicast/unicast at the raw Publish/Request level, without the event-API generation.
Backward embedder flightbooking Forward