pubsub

Documentation

Index

• type Handler
• type LegacySubscriber
  • func NewLegacySubscriber(node Handler) *LegacySubscriber
  • func (s *LegacySubscriber) Broadcast(msg *common.StreamMessage)
  • func (s *LegacySubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)
  • func (s *LegacySubscriber) IsMultiNode() bool
  • func (LegacySubscriber) Shutdown(ctx context.Context) error
  • func (LegacySubscriber) Start(done chan (error)) error
  • func (LegacySubscriber) Subscribe(stream string)
  • func (LegacySubscriber) Unsubscribe(stream string)
• type NATSSubscriber
  • func NewNATSSubscriber(node Handler, config *nconfig.NATSConfig, l *slog.Logger) (*NATSSubscriber, error)
  • func (s *NATSSubscriber) Broadcast(msg *common.StreamMessage)
  • func (s *NATSSubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)
  • func (s *NATSSubscriber) IsMultiNode() bool
  • func (s *NATSSubscriber) Publish(stream string, msg interface{})
  • func (s *NATSSubscriber) Shutdown(ctx context.Context) error
  • func (s *NATSSubscriber) Start(done chan (error)) error
  • func (s *NATSSubscriber) Subscribe(stream string)
  • func (s *NATSSubscriber) Unsubscribe(stream string)
• type RedisSubscriber
  • func NewRedisSubscriber(node Handler, config *rconfig.RedisConfig, l *slog.Logger) (*RedisSubscriber, error)
  • func (s *RedisSubscriber) Broadcast(msg *common.StreamMessage)
  • func (s *RedisSubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)
  • func (s *RedisSubscriber) IsMultiNode() bool
  • func (s *RedisSubscriber) Publish(stream string, msg interface{})
  • func (s *RedisSubscriber) Shutdown(ctx context.Context) error
  • func (s *RedisSubscriber) Start(done chan (error)) error
  • func (s *RedisSubscriber) Subscribe(stream string)
  • func (s *RedisSubscriber) Unsubscribe(stream string)
• type Subscriber

Types

type Handler

type Handler interface {
	Broadcast(msg *common.StreamMessage)
	ExecuteRemoteCommand(msg *common.RemoteCommandMessage)
}

type LegacySubscriber

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

func NewLegacySubscriber

func NewLegacySubscriber(node Handler) *LegacySubscriber

NewLegacySubscriber creates a legacy subscriber implementation to work with legacy Redis and NATS broadcasters

func (*LegacySubscriber) Broadcast

func (s *LegacySubscriber) Broadcast(msg *common.StreamMessage)

func (*LegacySubscriber) BroadcastCommand

func (s *LegacySubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)

func (*LegacySubscriber) IsMultiNode

func (s *LegacySubscriber) IsMultiNode() bool

func (LegacySubscriber) Shutdown

func (LegacySubscriber) Shutdown(ctx context.Context) error

func (LegacySubscriber) Start

func (LegacySubscriber) Start(done chan (error)) error

func (LegacySubscriber) Subscribe

func (LegacySubscriber) Subscribe(stream string)

func (LegacySubscriber) Unsubscribe

func (LegacySubscriber) Unsubscribe(stream string)

type NATSSubscriber

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

func NewNATSSubscriber

func NewNATSSubscriber(node Handler, config *nconfig.NATSConfig, l *slog.Logger) (*NATSSubscriber, error)

NewNATSSubscriber creates a NATS subscriber using pub/sub

func (*NATSSubscriber) Broadcast

func (s *NATSSubscriber) Broadcast(msg *common.StreamMessage)

func (*NATSSubscriber) BroadcastCommand

func (s *NATSSubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)

func (*NATSSubscriber) IsMultiNode

func (s *NATSSubscriber) IsMultiNode() bool

func (*NATSSubscriber) Publish

func (s *NATSSubscriber) Publish(stream string, msg interface{})

func (*NATSSubscriber) Shutdown

func (s *NATSSubscriber) Shutdown(ctx context.Context) error

func (*NATSSubscriber) Start

func (s *NATSSubscriber) Start(done chan (error)) error

func (*NATSSubscriber) Subscribe

func (s *NATSSubscriber) Subscribe(stream string)

func (*NATSSubscriber) Unsubscribe

func (s *NATSSubscriber) Unsubscribe(stream string)

type RedisSubscriber

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

func NewRedisSubscriber

func NewRedisSubscriber(node Handler, config *rconfig.RedisConfig, l *slog.Logger) (*RedisSubscriber, error)

NewRedisSubscriber creates a Redis subscriber using pub/sub

func (*RedisSubscriber) Broadcast

func (s *RedisSubscriber) Broadcast(msg *common.StreamMessage)

func (*RedisSubscriber) BroadcastCommand

func (s *RedisSubscriber) BroadcastCommand(cmd *common.RemoteCommandMessage)

func (*RedisSubscriber) IsMultiNode

func (s *RedisSubscriber) IsMultiNode() bool

func (*RedisSubscriber) Publish

func (s *RedisSubscriber) Publish(stream string, msg interface{})

func (*RedisSubscriber) Shutdown

func (s *RedisSubscriber) Shutdown(ctx context.Context) error

func (*RedisSubscriber) Start

func (s *RedisSubscriber) Start(done chan (error)) error

func (*RedisSubscriber) Subscribe

func (s *RedisSubscriber) Subscribe(stream string)

func (*RedisSubscriber) Unsubscribe

func (s *RedisSubscriber) Unsubscribe(stream string)

type Subscriber

type Subscriber interface {
	Start(done chan (error)) error
	Shutdown(ctx context.Context) error
	Broadcast(msg *common.StreamMessage)
	BroadcastCommand(msg *common.RemoteCommandMessage)
	Subscribe(stream string)
	Unsubscribe(stream string)
	IsMultiNode() bool
}

Subscriber is responsible for subscribing to individual streams and and publishing messages to streams