Events between resolvers

GraphQL subscription resolvers can consume events published by mutations or other application code. grx keeps this boundary backend-neutral with the github.com/grx-gql/grx/pubsub package.

Use the memory backend when one process owns all publishers and subscribers. Use Redis or another shared backend when replicas must fan out events. Define the delivery guarantees before choosing a backend.

Choosing a backendCopy Anchor Link

Both implementations satisfy the same pubsub.PubSub contract. Slow subscribers may miss messages, and messages published before subscription are not replayed. Use a durable event system separately when replay is required.

The interface does not promise ordering across topics, duplicate suppression, durability, replay, or delivery after reconnect. A custom backend must document which of these it provides.

The interfaceCopy Anchor Link

package pubsub

type Message struct {
	Topic   string
	Payload []byte
}

type PubSub interface {
	Publish(ctx context.Context, topic string, payload []byte) error
	Subscribe(ctx context.Context, topic string) (<-chan Message, error)
	Close() error
}

Topics are exact matches. Use hierarchical topic names such as message.posted.room-1 when a resolver needs scoped streams.

Payloads are opaque to the bus. The publisher owns the payload until Publish returns and must not mutate or reuse its backing array afterward. Subscribers must treat received payloads as read-only; copy them when they need to retain or modify the bytes.

Memory backendCopy Anchor Link

import "github.com/grx-gql/grx/pubsub/memory"

bus := memory.New()
defer bus.Close()

The Memory implementation avoids serialization and is suitable for tests, development, and single-replica deployments.

Redis backendCopy Anchor Link

The Redis backend is a separate module so applications that do not need Redis do not add a Redis dependency to the root module.

import (
	redisps "github.com/grx-gql/grx/pubsub/redis"
	"github.com/redis/go-redis/v9"
)

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
bus, err := redisps.New(redisps.Config{Client: rdb, Prefix: "grx:"})
if err != nil {
	panic(err)
}
defer bus.Close()

Resolver wiringCopy Anchor Link

Encode and decode domain values at the resolver boundary. This keeps every backend interchangeable and avoids putting application-specific codecs in the transport-agnostic bus.

Each resolver holds the same pubsub.PubSub directly:

type MessageMutation struct {
	Bus pubsub.PubSub
}

type MessageSubscription struct {
	Bus pubsub.PubSub
}

The subscription example composes these dependencies with grx.WithSchema(graph.Schema(bus)). Without a bus, graph.Schema(nil) registers only queries and non-event mutations, with no subscription root.

See Realtime subscriptions for the tested message resolver, JSON encoding and decoding, and shared bus wiring.

Lifecycle and backpressureCopy Anchor Link

Pass the resolver context into Publish and Subscribe. Cancel the subscription when the client disconnects and close the backend during service shutdown. Remove subscribers when their contexts end.

Choose a buffer policy deliberately. A bounded buffer can drop or reject events; an unbounded buffer can exhaust memory. Do not let one slow subscriber block unrelated topics or publishers unless that is an explicit guarantee.

For a new backend, read Custom pub/sub and test publish, subscribe, cancellation, cleanup, shutdown, and error behavior.