bus

Documentation

Index

• Constants
• func DeserializePayload[T proto.Message](buf []byte) (T, error)
• func SerializePayload(m proto.Message) ([]byte, error)
• type EmptySubscription
  • func (s EmptySubscription[MessageType]) Channel() <-chan MessageType
  • func (s EmptySubscription[MessageType]) Close() error
• type MessageBus
  • func NewLocalMessageBus() MessageBus
  • func NewNatsMessageBus(nc *nats.Conn) MessageBus
  • func NewRedisMessageBus(rc redis.UniversalClient) MessageBus
  • func NewTestBus(bus MessageBus, opts ...TestBusOption) MessageBus
• type PublishHandler
• type PublishInterceptor
• type ReadHandler
• type Reader
• type SubscribeInterceptor
• type Subscription
  • func Subscribe[MessageType proto.Message](ctx context.Context, bus MessageBus, channel string, channelSize int) (Subscription[MessageType], error)
  • func SubscribeQueue[MessageType proto.Message](ctx context.Context, bus MessageBus, channel string, channelSize int) (Subscription[MessageType], error)
• type TestBusOption
• type TestBusOpts

Constants

const (
	DefaultChannelSize = 100
)

Functions

func DeserializePayload

func DeserializePayload[T proto.Message](buf []byte) (T, error)

func SerializePayload

func SerializePayload(m proto.Message) ([]byte, error)

Types

type EmptySubscription

type EmptySubscription[MessageType any] struct{}

func (EmptySubscription[MessageType]) Channel

func (s EmptySubscription[MessageType]) Channel() <-chan MessageType

func (EmptySubscription[MessageType]) Close

func (s EmptySubscription[MessageType]) Close() error

type MessageBus

type MessageBus interface {
	Publish(ctx context.Context, channel string, msg proto.Message) error
	Subscribe(ctx context.Context, channel string, channelSize int) (Reader, error)
	SubscribeQueue(ctx context.Context, channel string, channelSize int) (Reader, error)
}

func NewLocalMessageBus

func NewLocalMessageBus() MessageBus

func NewNatsMessageBus

func NewNatsMessageBus(nc *nats.Conn) MessageBus

func NewRedisMessageBus

func NewRedisMessageBus(rc redis.UniversalClient) MessageBus

func NewTestBus

func NewTestBus(bus MessageBus, opts ...TestBusOption) MessageBus

type PublishHandler

type PublishHandler func(ctx context.Context, channel string, msg proto.Message) error

type PublishInterceptor

type PublishInterceptor func(next PublishHandler) PublishHandler

type ReadHandler

type ReadHandler func() ([]byte, bool)

type Reader

type Reader interface {
	Close() error
	// contains filtered or unexported methods
}

type SubscribeInterceptor

type SubscribeInterceptor func(ctx context.Context, channel string, next ReadHandler) ReadHandler

type Subscription

type Subscription[MessageType proto.Message] interface {
	Channel() <-chan MessageType
	Close() error
}

func Subscribe

func Subscribe[MessageType proto.Message](
	ctx context.Context,
	bus MessageBus,
	channel string,
	channelSize int,
) (Subscription[MessageType], error)

func SubscribeQueue

func SubscribeQueue[MessageType proto.Message](
	ctx context.Context,
	bus MessageBus,
	channel string,
	channelSize int,
) (Subscription[MessageType], error)

type TestBusOption

type TestBusOption func(*TestBusOpts)

type TestBusOpts

type TestBusOpts struct {
	PublishInterceptors   []PublishInterceptor
	SubscribeInterceptors []SubscribeInterceptor
}