kafka

Sep 20, 2023

Changes in this version

type Client

+ func (c *Client) MarkMessage(msg *Message) error

type ConsumerGroupHandler

+ func (c *ConsumerGroupHandler) Commit() error

+ func (c *ConsumerGroupHandler) MarkMessage(msg *sarama.ConsumerMessage, metadata string) error

type Message

+ Offset int64

+ Partition int32

type NewClientOptions

+ AutoMark bool

Jul 14, 2023

Changes in this version

type NewClientOptions

+ ConsumerGroupId string

May 30, 2023

Changes in this version

type Client

+ func (c *Client) GetQueueLength() int

May 26, 2023

Changes in this version

type NewClientOptions

+ TransactionalID string

May 17, 2023

Changes in this version

type Client

+ func (c *Client) Closed() bool

May 15, 2023

Changes in this version

+ func MessageToConsumerMessage(message *Message) *sarama.ConsumerMessage

+ func MessageToProducerMessage(message *Message) *sarama.ProducerMessage

+ func ProducerMessageToConsumerMessage(message *sarama.ProducerMessage) (*sarama.ConsumerMessage, error)

type Client

+ func (c *Client) AbortTxn() error

+ func (c *Client) AddMessageToTxn(msg *Message) error

+ func (c *Client) BeginTxn() error

+ func (c *Client) CommitTxn() error

+ func (c *Client) GetConsumerErrorsChannel() <-chan error

+ func (c *Client) GetProducerErrorsChannel() <-chan *sarama.ProducerError

+ func (c *Client) GetProducerSuccessesChannel() <-chan *sarama.ProducerMessage

type Message

+ func ProducerMessageToMessage(prodMsg *sarama.ProducerMessage) (*Message, error)

type NewClientOptions

+ AutoCommit bool

+ ProducerReturnSuccesses bool

May 10, 2023

Changes in this version

type NewClientOptions

+ OpenDeadLine time.Duration

May 3, 2023

Changes in this version

type NewClientOptions

+ ClientID string

May 3, 2023

Changes in this version

+ var MicroserviceName = env.GetAsString("MICROSERVICE_NAME", false, "")

+ var SerialNumber = env.GetAsString("SERIAL_NUMBER", false, "")

+ func AddXOriginIfMissing(message *[]sarama.RecordHeader) error

+ func AddXTrace(message *[]sarama.RecordHeader, value string) error

type NewClientOptions

+ SenderTag SenderTag

+ type SenderTag struct

+ Enabled bool

+ OverwriteMicroserviceName *string

+ OverwriteSerialNumber *string

+ type TraceValue struct

+ Traces map[int64]string

+ func GetTrace(message *[]sarama.RecordHeader, key string) *TraceValue

May 3, 2023

Changes in this version

type Client

+ func (c *Client) TopicCreator(topic string) (err error)

Mar 21, 2023

Changes in this version

+ func GetKafkaStats() (send uint64, received uint64)

Mar 21, 2023

Changes in this version

+ type Message struct

+ Header map[string][]byte

+ Key []byte

+ Topic string

+ Value []byte

Mar 16, 2023

Changes in this version

type NewClientOptions

+ StartOffset int64

Mar 9, 2023

Changes in this version

type NewClientOptions

+ EnableTLS bool

Mar 6, 2023

Changes in this version

+ func NewConsumerGroupHandler(queue *chan KafkaMessage) sarama.ConsumerGroupHandler

+ type Client struct

+ func NewKafkaClient(opts NewClientOptions) (client *Client, err error)

+ func (c *Client) ChangeSubscribedTopics(newTopicRegex *regexp.Regexp)

+ func (c *Client) Close() error

+ func (c *Client) EnqueueMessage(msg KafkaMessage) (err error)

+ func (c *Client) GSub(ctx context.Context, cncl context.CancelFunc)

+ func (c *Client) GetMessages() <-chan KafkaMessage

+ func (c *Client) Ready() bool

+ type ConsumerGroupHandler struct

+ func (c ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

+ func (c ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (c ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

+ type KafkaMessage struct

+ Topic string

+ Value []byte

+ func ToKafkaMessage(topic string, message interface{}) (KafkaMessage, error)

+ type NewClientOptions struct

+ Brokers []string

+ ConsumerName string

+ ListenTopicRegex *regexp.Regexp

+ Partitions int32

+ ReplicationFactor int16