memory

Documentation

Overview

Package memory provides an in-process Backend implementation for testing and single-instance deployments. Multiple linked backends can be created via Fork to simulate cross-process fan-out within a single process.

Index

• Variables
• type Backend
  • func New() *Backend
  • func (b *Backend) Close() error
  • func (b *Backend) Fork() *Backend
  • func (b *Backend) Healthy() bool
  • func (b *Backend) OnReconnect(fn func())
  • func (b *Backend) Publish(_ context.Context, env backend.MessageEnvelope) error
  • func (b *Backend) SetHealthy(healthy bool)
  • func (b *Backend) Subscribe(_ context.Context, topic string) (<-chan backend.MessageEnvelope, error)
  • func (b *Backend) Unsubscribe(topic string) error

Variables

var ErrClosed = errClosed{}

ErrClosed is returned when an operation is attempted on a closed backend. It is a sentinel error safe for comparison with errors.Is.

Types

type Backend

type Backend struct {
	// contains filtered or unexported fields
}

Backend is an in-process implementation of backend.Backend backed by a shared bus. Use New to create one. Call Backend.Fork to create a linked instance that shares the same bus — publishes on either instance are visible to subscribers on the other.

Backend also implements backend.HealthAwareBackend for health checking and reconnection notifications.

func New

func New() *Backend

New creates a new in-process Backend with its own bus.

func (*Backend) Close

func (b *Backend) Close() error

Close shuts down the backend and closes all subscription channels.

func (*Backend) Fork

func (b *Backend) Fork() *Backend

Fork creates a new Backend that shares the same message bus as the receiver. Messages published on any fork are delivered to subscribers on all other forks (but not back to the publisher).

func (*Backend) Healthy

func (b *Backend) Healthy() bool

Healthy reports whether the backend is currently healthy. Returns false when the backend is closed or has been marked unhealthy via [SetHealthy].

func (*Backend) OnReconnect

func (b *Backend) OnReconnect(fn func())

OnReconnect registers a callback that fires when the backend transitions from unhealthy to healthy. The callback is invoked from [SetHealthy].

func (*Backend) Publish

func (b *Backend) Publish(_ context.Context, env backend.MessageEnvelope) error

Publish sends env to all other backends subscribed to the same topic.

func (*Backend) SetHealthy

func (b *Backend) SetHealthy(healthy bool)

SetHealthy sets the health state. When transitioning from unhealthy to healthy, the registered OnReconnect callback is fired.

func (*Backend) Subscribe

func (b *Backend) Subscribe(_ context.Context, topic string) (<-chan backend.MessageEnvelope, error)

Subscribe registers interest in a topic and returns a channel that receives envelopes published by other backends sharing the same bus.

func (*Backend) Unsubscribe

func (b *Backend) Unsubscribe(topic string) error

Unsubscribe removes the subscription for the given topic.