pubsub

Documentation

Index

• func AsKafkaMessage(msg Message) kafka.Message
• type Handler
• type KafkaMessage
  • func NewMessage(msg kafka.Message) *KafkaMessage
  • func (k KafkaMessage) Key() []byte
  • func (k KafkaMessage) Value() []byte
• type Message
• type Publisher
  • func NewPublisher(w *kafka.Writer) *Publisher
  • func (p *Publisher) Publish(ctx context.Context, msgs ...Message) error
• type Subscriber
  • func NewSubscriber(r *kafka.Reader) *Subscriber
  • func (s *Subscriber) Receive(ctx context.Context, h Handler) (func(), <-chan error)

Functions

func AsKafkaMessage

func AsKafkaMessage(msg Message) kafka.Message

Types

type Handler

type Handler func(ctx context.Context, msg Message) error

type KafkaMessage

type KafkaMessage struct {
	kafka.Message
}

func NewMessage

func NewMessage(msg kafka.Message) *KafkaMessage

func (KafkaMessage) Key

func (k KafkaMessage) Key() []byte

func (KafkaMessage) Value

func (k KafkaMessage) Value() []byte

type Message

type Message interface {
	Key() []byte
	Value() []byte
}

type Publisher

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

func NewPublisher

func NewPublisher(w *kafka.Writer) *Publisher

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, msgs ...Message) error

type Subscriber

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

func NewSubscriber

func NewSubscriber(r *kafka.Reader) *Subscriber

func (*Subscriber) Receive

func (s *Subscriber) Receive(ctx context.Context, h Handler) (func(), <-chan error)

Receive handles the message received from the message queue. Returning an error will not commit the offset.