bus

Documentation

Index

• type Consumer
• type Message
  • func NewMessage(topic string, payload interface{}) (m Message)
  • func (m Message) IsEqual(ingest Message) (res bool)
  • func (m Message) Payload() (res Payload)
  • func (m Message) String() (res string)
  • func (m Message) Topic() string
• type Payload
  • func NewPayload(v interface{}) (p Payload)
  • func (p Payload) Hash() uint64
  • func (p Payload) IsEmpty() bool
  • func (p Payload) String() (res string)
  • func (p Payload) Unmarshal(v interface{}) error
• type Pipe
• type Producer
• type ReplicatedPipe
• type TestingConsumer
  • func NewTestingConsumer(ctx context.Context) (c *TestingConsumer)
  • func (c *TestingConsumer) ConsumeMessage(message Message) (err error)
  • func (c *TestingConsumer) ExpectLastMessageFn(message Message) (fn func() error)
  • func (c *TestingConsumer) ExpectMessagesByIdFn(expect map[string][]Message) (fn func() error)
  • func (c *TestingConsumer) ExpectMessagesFn(expect ...Message) (fn func() error)

Types

type Consumer

type Consumer interface {
	ConsumeMessage(message Message) (err error)
}

Consumer consumes messages

type Message

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

Message encapsulates topic with given payload

func NewMessage

func NewMessage(topic string, payload interface{}) (m Message)

Create new message

func (Message) IsEqual

func (m Message) IsEqual(ingest Message) (res bool)

Is message equal to given message

func (Message) Payload

func (m Message) Payload() (res Payload)

Get message payload

func (Message) String

func (m Message) String() (res string)

func (Message) Topic

func (m Message) Topic() string

Get topic

type Payload

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

func NewPayload

func NewPayload(v interface{}) (p Payload)

func (Payload) Hash

func (p Payload) Hash() uint64

func (Payload) IsEmpty

func (p Payload) IsEmpty() bool

func (Payload) String

func (p Payload) String() (res string)

func (Payload) Unmarshal

func (p Payload) Unmarshal(v interface{}) error

type Pipe

type Pipe interface {
	GetConsumer() (c Consumer)
}

Pipe consumes messages and pipes them to downstream

type Producer

type Producer interface {
	Subscribe(ctx context.Context, consumer Consumer)
}

type ReplicatedPipe

type ReplicatedPipe interface {
	GetConsumers() (c []Consumer)
}

Replicated pipes messages to multiple pipes

type TestingConsumer

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

func NewTestingConsumer

func NewTestingConsumer(ctx context.Context) (c *TestingConsumer)

func (*TestingConsumer) ConsumeMessage

func (c *TestingConsumer) ConsumeMessage(message Message) (err error)

func (*TestingConsumer) ExpectLastMessageFn

func (c *TestingConsumer) ExpectLastMessageFn(message Message) (fn func() error)

func (*TestingConsumer) ExpectMessagesByIdFn

func (c *TestingConsumer) ExpectMessagesByIdFn(expect map[string][]Message) (fn func() error)

func (*TestingConsumer) ExpectMessagesFn

func (c *TestingConsumer) ExpectMessagesFn(expect ...Message) (fn func() error)