GoChannel event provider

In-process pub/sub backed by Watermill's GoChannel transport. Single-process delivery with optional fan-out replay for late subscribers (not durable persistence), with no broker or network in the loop, and so no horizontal scale.

Overview

The GoChannel provider implements the events Provider port (the same port as the NATS provider) using Watermill's gochannel.GoChannel for transport. A compile-time assertion in the package binds the provider to that port, so GoChannel and NATS are drop-in peers behind one interface. Watermill routes messages through buffered Go channels inside the running process, no network hop, no on-disk persistence, no cross-replica delivery.

The OutputChannelBuffer setting controls per-subscriber back-pressure (field type int64). A zero value yields an unbuffered channel, so set it through DefaultConfig(), which supplies a buffer of 8192. When BlockPublishUntilSubscriberAck is true, Publish blocks until every subscriber acknowledges, which guarantees delivery at the cost of throughput. When false, delivery runs asynchronously, and a saturated subscriber drops messages. Even with blocking enabled, Publish does not block when a topic has no subscribers.

The Persistent option does not mean "survives a process restart". It controls whether the provider buffers messages published before any subscription exists and replays them when a subscriber connects (a fan-out pattern). Restarting the process loses the buffer either way.

The provider reuses piko machinery. Watermill's internal logging bridges to piko's structured logger, the router goroutine runs under piko's panic recovery, and start and close operations emit OpenTelemetry metrics.

The architectural choice between in-process and clustered messaging matters here. GoChannel handles single-process workloads, background-task fan-out, internal state-change broadcasts, and tests, where every publisher and subscriber lives in the same binary. Reach for NATS when subscribers run in a different process or replica, when messages must survive a crash, or when you want JetStream's replay semantics. The two providers are interchangeable at the events Provider boundary. To swap, construct the new provider, start it, and pass it to the bootstrap option.

Configuration

Start from DefaultConfig(), which is what the bootstrap container itself uses. It sets OutputChannelBuffer to 8192, BlockPublishUntilSubscriberAck to false (fire-and-forget), and Persistent to false. Override a single field instead of re-specifying defaults.

import (
    "piko.sh/piko/wdk/events"
    "piko.sh/piko/wdk/events/events_provider_gochannel"
)

config := events_provider_gochannel.DefaultConfig()
config.BlockPublishUntilSubscriberAck = true // opt in to back-pressure
config.RouterConfig = events.RouterConfig{}   // public alias for Watermill router settings

provider, err := events_provider_gochannel.NewGoChannelProvider(config)
if err != nil {
    return err
}

Import shared types such as RouterConfig from the public piko.sh/piko/wdk/events surface. The piko.sh/piko/internal/... tree is not importable from application code.

Bootstrap

If you set no events provider, the container builds a GoChannel provider from DefaultConfig(), starts it, and registers its shutdown. Events work with no configuration. The frontmatter tags this the default for that reason.

To override the default, pass your own provider. The events service takes a single provider. There is no name registry on this option.

if err := provider.Start(ctx); err != nil {
    return err
}

ssr := piko.New(
    piko.WithEventsProvider(provider),
)

A provider passed through WithEventsProvider is not lifecycle-managed by the container. Call Start(ctx) before you pass it, or the events router never runs. The provided context controls the router's lifetime. Call Close() yourself on shutdown to stop the router cleanly. Automatic start and shutdown apply only to the default GoChannel provider that the container creates when you set no override.

See also

Other event providers:

  • NATS, distributed pub/sub with optional JetStream persistence; reach for this when subscribers span processes or replicas.

Framework docs:

External: