pubsub

Documentation

Index

• type ConfluentKafkaConsumer
• type ConfluentKafkaConsumerProperty
• type ConfluentKafkaProducer
• type DLQHandler
• type DLQHandlerAdapter
  • func NewDLQHandlerAdapter(topic string, publisher Publisher) *DLQHandlerAdapter
  • func (dlqHandlerAdapter *DLQHandlerAdapter) Send(ctx context.Context, dlqMessage *DeadLetterQueueMessage) (err error)
• type DeadLetterQueueMessage
• type DefaultEventHandler
  • func (h DefaultEventHandler) Handle(ctx context.Context, message interface{}) (err error)
• type EventHandler
• type MessageHeaders
  • func (mh MessageHeaders) Add(key, value string)
• type Publisher
  • func PublisherFromConfluentKafkaProducer(logger *logrus.Logger, producer ConfluentKafkaProducer) Publisher
• type Subscriber
  • func SubscriberFromConfluentKafkaConsumer(props ConfluentKafkaConsumerProperty) Subscriber

Types

type ConfluentKafkaConsumer

type ConfluentKafkaConsumer interface {
	Assign(partitions []ck.TopicPartition) (err error)
	Assignment() (partitions []ck.TopicPartition, err error)
	Unassign() (err error)
	SubscribeTopics(topics []string, rb ck.RebalanceCb) (err error)
	Poll(ms int) ck.Event
	Commit() (partitions []ck.TopicPartition, err error)
	Close() (err error)
}

type ConfluentKafkaConsumerProperty

type ConfluentKafkaConsumerProperty struct {
	Logger       *logrus.Logger
	Topic        string
	EventHandler EventHandler
	Consumer     ConfluentKafkaConsumer
}

type ConfluentKafkaProducer

type ConfluentKafkaProducer interface {
	Events() chan ck.Event
	Produce(msg *ck.Message, deliveryChan chan ck.Event) error
	Flush(timeoutMs int) int
	Close()
}

type DLQHandler

type DLQHandler interface {
	Send(ctx context.Context, dlqMessage *DeadLetterQueueMessage) (err error)
}

DLQHandler is an handler to handler dead letter queue or an unprocessed message

type DLQHandlerAdapter

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

DLQHandlerAdapter is an dead letter queue adapter.

func NewDLQHandlerAdapter

func NewDLQHandlerAdapter(topic string, publisher Publisher) *DLQHandlerAdapter

NewDLQHandlerAdapter is a constructor.

func (*DLQHandlerAdapter) Send

func (dlqHandlerAdapter *DLQHandlerAdapter) Send(ctx context.Context, dlqMessage *DeadLetterQueueMessage) (err error)

Send will publish the dlq message to the assigned topic.

type DeadLetterQueueMessage

type DeadLetterQueueMessage struct {
	Channel           string         `json:"channel"`
	Publisher         string         `json:"publisher"`
	Consumer          string         `json:"consumer"`
	Key               string         `json:"key"`
	Headers           MessageHeaders `json:"headers"`
	Message           string         `json:"message"`
	CausedBy          string         `json:"caused_by"`
	FailedConsumeDate string         `json:"failed_consume_date"`
}

DeadLetterQueueMessage is an entity.

type DefaultEventHandler

type DefaultEventHandler struct {
	Logger *logrus.Logger
}

func (DefaultEventHandler) Handle

func (h DefaultEventHandler) Handle(ctx context.Context, message interface{}) (err error)

type EventHandler

type EventHandler interface {
	Handle(ctx context.Context, message interface{}) (err error)
}

EventHandler is an event handler. It will be called after message is arrived to consumer

type MessageHeaders

type MessageHeaders map[string]string

MessageHeaders is type of message headers

func (MessageHeaders) Add

func (mh MessageHeaders) Add(key, value string)

Add will add the key and value to headers.

type Publisher

type Publisher interface {
	// Will send the message to the assigned topic.
	Publish(ctx context.Context, topic string, key string, headers MessageHeaders, message []byte) (err error)
	Close() (err error)
}

Publisher is a collection of behavior of a publisher

func PublisherFromConfluentKafkaProducer

func PublisherFromConfluentKafkaProducer(logger *logrus.Logger, producer ConfluentKafkaProducer) Publisher

type Subscriber

type Subscriber interface {
	Subscribe()
	Close() (err error)
}

Subscriber is a collection of behavior of a subscriber

func SubscriberFromConfluentKafkaConsumer

func SubscriberFromConfluentKafkaConsumer(props ConfluentKafkaConsumerProperty) Subscriber