trs_kafka

Documentation

Index

• func GenerateSendReceiveConsumerGroupName(appName, workerClass, consumerGroup string) (sendTopic string, rcvTopic string, cgroupName string)
• type TRSKafka
  • func (trsKafka *TRSKafka) Init(ctx context.Context, initialReceiveTopics []string, consumerGroup string, ...) (err error)
  • func (trsKafka *TRSKafka) SetTopics(topics []string) (err error)
  • func (trsKafka *TRSKafka) Shutdown()
  • func (trsKafka *TRSKafka) Write(topic string, payload []byte)
• type TRSKafkaClient
• type TRSKafkaConsumer
  • func (consumer *TRSKafkaConsumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (consumer *TRSKafkaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *TRSKafkaConsumer) Setup(sarama.ConsumerGroupSession) error

Functions

func GenerateSendReceiveConsumerGroupName

func GenerateSendReceiveConsumerGroupName(appName, workerClass, consumerGroup string) (sendTopic string, rcvTopic string, cgroupName string)

Types

type TRSKafka

type TRSKafka struct {
	Client           *TRSKafkaClient
	ConsumerShutdown chan int
	RcvTopicNames    []string
	Mux              sync.Mutex
	Logger           *logrus.Logger
}

func (*TRSKafka) Init

func (trsKafka *TRSKafka) Init(ctx context.Context,
	initialReceiveTopics []string,
	consumerGroup string,
	brokerSpec string,
	tasksChan chan *sarama.ConsumerMessage, logger *logrus.Logger) (err error)

func (*TRSKafka) SetTopics

func (trsKafka *TRSKafka) SetTopics(topics []string) (err error)

func (*TRSKafka) Shutdown

func (trsKafka *TRSKafka) Shutdown()

func (*TRSKafka) Write

func (trsKafka *TRSKafka) Write(topic string, payload []byte)

type TRSKafkaClient

type TRSKafkaClient struct {
	Config        *sarama.Config
	Producer      *sarama.AsyncProducer
	ConsumerGroup *sarama.ConsumerGroup
	Consumer      *TRSKafkaConsumer
	Logger        *logrus.Logger
}

type TRSKafkaConsumer

type TRSKafkaConsumer struct {
	Errors    chan error
	Responses chan *sarama.ConsumerMessage
	Logger    *logrus.Logger
}

func (*TRSKafkaConsumer) Cleanup

func (consumer *TRSKafkaConsumer) Cleanup(sarama.ConsumerGroupSession) error

func (*TRSKafkaConsumer) ConsumeClaim

func (consumer *TRSKafkaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

func (*TRSKafkaConsumer) Setup

func (consumer *TRSKafkaConsumer) Setup(sarama.ConsumerGroupSession) error