kafka

Documentation

Index

• type Consumer
  • func NewConsumer(committerInterval time.Duration, bootstrapServer string, ...) (*Consumer, error)
  • func (c *Consumer) Register(registration *transport.BundleRegistration)
  • func (c *Consumer) Start()
  • func (c *Consumer) Stop()
• type KafkaConsumerConfig
• type Message

Types

type Consumer

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

Consumer abstracts hub-of-hubs-kafka-transport kafka-consumer's generic usage.

func NewConsumer

func NewConsumer(committerInterval time.Duration, bootstrapServer string, consumerConfig *KafkaConsumerConfig, conflationManager *conflator.ConflationManager, statistics *statistics.Statistics,
	log logr.Logger,
) (*Consumer, error)

NewConsumer creates a new instance of Consumer.

func (*Consumer) Register

func (c *Consumer) Register(registration *transport.BundleRegistration)

Register function registers a msgID to the bundle updates channel.

func (*Consumer) Start

func (c *Consumer) Start()

Start function starts the consumer.

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop stops the consumer.

type KafkaConsumerConfig

type KafkaConsumerConfig struct {
	ConsumerID    string
	ConsumerTopic string
}

type Message

type Message struct {
	ID      string `json:"id"`
	MsgType string `json:"msgType"`
	Version string `json:"version"`
	Payload []byte `json:"payload"`
}

Message abstracts a message object to be used by different transport components.