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

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 *kafka.Message, 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 {
	Producer *kafka.Producer
	Consumer *TRSKafkaConsumer
	Logger   *logrus.Logger
}

type TRSKafkaConsumer

type TRSKafkaConsumer struct {
	Errors    chan error
	Responses chan *kafka.Message
	Logger    *logrus.Logger
}