Documentation
¶
Index ¶
Constants ¶
const ( TopicMockNext = "topic_mock_next" TopicUpload = "topic_upload" TypeOfStructInMessageValue = "type_of_struct_in_value" )
Variables ¶
var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
SHA256 is a hash generator function for SCRAM-SHA-256.
var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
SHA512 is a hash generator function for SCRAM-SHA-512.
Functions ¶
Types ¶
type ConsumerGroupHandler ¶
type ConsumerGroupHandler struct {
Handlers map[string]MessageHandler
Log *logger.Log
}
ConsumerGroupHandler struct that handles Kafka group handlers
func (*ConsumerGroupHandler) Cleanup ¶
func (cgh *ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error
func (*ConsumerGroupHandler) ConsumeClaim ¶
func (cgh *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
func (*ConsumerGroupHandler) Setup ¶
func (cgh *ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error
type HandlerConfig ¶
HandlerConfig struct to define the Kafka topic and consumer group for a specific message handler
type MessageConsumerClient ¶
type MessageConsumerClient struct {
SaramaConfig *sarama.Config
// contains filtered or unexported fields
}
MessageConsumerClient ATTENTION: Start max one instance of consumer client for each service to keep resource usage low
func NewConsumerClient ¶
func (*MessageConsumerClient) Close ¶
func (c *MessageConsumerClient) Close(ctx context.Context) error
Close closes the consumer client
func (*MessageConsumerClient) Start ¶
func (c *MessageConsumerClient) Start(ctx context.Context, handlerFactory func(string) sarama.ConsumerGroupHandler, handlerConfigs []HandlerConfig) error
Start starts the actual event consuming from specified kafka topics
type MessageHandler ¶
type MessageHandler interface {
HandleMessage(ctx context.Context, message *sarama.ConsumerMessage) error
}
MessageHandler definition of a generic Kafka message handler
type SyncProducerClient ¶
type SyncProducerClient struct {
// contains filtered or unexported fields
}
MessageConsumerClient ATTENTION: Start max one instance of Kafka consumer client for each service to keep resource usage low
func NewSyncProducerClient ¶
func NewSyncProducerClient(ctx context.Context, saramaConfig *sarama.Config, cfg *model.Cfg, tracer *trace.Tracer, log *logger.Log) (*SyncProducerClient, error)
NewSyncProducerClient creates a Kafka sync producer client
func (*SyncProducerClient) Close ¶
func (c *SyncProducerClient) Close(ctx context.Context) error
Close close the producer
func (*SyncProducerClient) PublishMessage ¶
func (c *SyncProducerClient) PublishMessage(topic string, key string, json []byte, headers []sarama.RecordHeader) error
PublishMessage publish a message to a Kafka topic
type XDGSCRAMClient ¶ added in v0.6.4
type XDGSCRAMClient struct {
*scram.ClientConversation
scram.HashGeneratorFcn
}
XDGSCRAMClient implements the sarama.SCRAMClient interface using xdg-go/scram.
func (*XDGSCRAMClient) Begin ¶ added in v0.6.4
func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
Begin starts a new SCRAM conversation.
func (*XDGSCRAMClient) Done ¶ added in v0.6.4
func (x *XDGSCRAMClient) Done() bool
Done returns true if the conversation is complete.
Source Files