topic

Documentation

Index

• Variables
• func New() message.M
• func Subscribe(ctx context.Context, b *message.Broker, topics ...string) error
• func Unsubscribe(ctx context.Context, b *message.Broker, topics ...string) error
• type M
• type Names
• type Option
  • func Named(o Names) Option
  • func Pattern(o message.Pattern) Option
• type Pipe
  • func Piped(opts ...Option) *Pipe
  • func (a *Pipe) Add(topics ...string)
  • func (a *Pipe) Handle(c *message.Conn, m interface{}) error
  • func (a *Pipe) Remove(topics ...string)
  • func (a *Pipe) Send(ctx context.Context, m interface{}, conns ...*message.Conn) error
• type Subscription

Variables

var (
	Invalid         = errors.New("invalid topic")
	NoSubscriptions = errors.New("no subscriptions")
)

var SubscriptionMessageID uint32 = 10

Functions

func New

func New() message.M

New implements message.New for Subscription.

func Subscribe

func Subscribe(ctx context.Context, b *message.Broker, topics ...string) error

func Unsubscribe

func Unsubscribe(ctx context.Context, b *message.Broker, topics ...string) error

Types

type M

type M interface {
	Topic() string
}

type Names

type Names interface {
	Add(topics ...string)
	Remove(topics ...string)
	Search(patterns ...string) (topics []string)
}

type Option

type Option func(*Pipe)

func Named

func Named(o Names) Option

func Pattern

func Pattern(o message.Pattern) Option

type Pipe

type Pipe struct {
	message.Pipe
	// contains filtered or unexported fields
}

func Piped

func Piped(opts ...Option) *Pipe

func (*Pipe) Add

func (a *Pipe) Add(topics ...string)

func (*Pipe) Handle

func (a *Pipe) Handle(c *message.Conn, m interface{}) error

func (*Pipe) Remove

func (a *Pipe) Remove(topics ...string)

func (*Pipe) Send

func (a *Pipe) Send(ctx context.Context, m interface{}, conns ...*message.Conn) error

type Subscription

type Subscription struct {
	Added   []string
	Removed []string
	Topics  []string
}