kafka

type Client ¶

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

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

func (c *Client) AbortTxn() error

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

func (c *Client) BeginTxn() error

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

func (c *Client) Close() error

func (c *Client) Closed() bool

func (c *Client) CommitTxn() error

func (c *Client) EnqueueMessage(msg Message) (err error)

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

func (c *Client) GetMessages() <-chan Message

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

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

func (c *Client) GetQueueLength() int

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

func (c *Client) Ready() bool

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

type ConsumerGroupHandler ¶

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

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

func (c *ConsumerGroupHandler) Commit() error

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

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

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

type Message ¶ added in v1.2.0

type Message struct {
	Topic     string
	Value     []byte
	Header    map[string][]byte
	Key       []byte
	Offset    int64
	Partition int32
}

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

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

type NewClientOptions ¶

type NewClientOptions struct {
	ListenTopicRegex        *regexp.Regexp
	SenderTag               SenderTag
	ConsumerGroupId         string
	TransactionalID         string
	ClientID                string
	Brokers                 []string
	StartOffset             int64
	OpenDeadLine            time.Duration
	Partitions              int32
	ReplicationFactor       int16
	EnableTLS               bool
	AutoCommit              bool
	ProducerReturnSuccesses bool
	AutoMark                bool
}

NewClientOptions are the options for creating a new kafka client. ListenTopicRegex is the regex to match topics to listen to. ConsumerName is the name of the consumer group (group.id). Brokers is the list of brokers to connect to. StartOffset is the offset to start consuming from. Partitions is the number of partitions to create for new topics. ReplicationFactor is the replication factor to use for new topics. EnableTLS enables TLS for the connection. SenderTag controls the sender tagging feature. ClientID is the client ID to use for the connection, only relevant for debugging. AutoCommit enables auto-commit (Default: true) [Note: AutoCommit only commits Marked Messages. You shouldn't need to set this to false] OpenDeadLine is the deadline, until connection to the brokers must be established. ProducerReturnSuccesses enables the success output channel of back to the user. TransactionalID is required for transaction of producers, e.g.,BeginTxn(). AutoMark enabled automatic marking of messages as consumed. [Set this to false if you want to manually mark messages as consumed]

type SenderTag ¶ added in v1.4.0

type SenderTag struct {
	OverwriteSerialNumber     *string
	OverwriteMicroserviceName *string
	Enabled                   bool
}

SenderTag controls the sender tagging feature. Enabled enables the feature. OverwriteSerialNumber overwrites the serial number with the given value, else the SERIAL_NUMBER env variable is used. OverwriteMicroserviceName overwrites the microservice name with the given value, else the MICROSERVICE_NAME env variable is used.

type TraceValue ¶ added in v1.4.0

type TraceValue struct {
	Traces map[int64]string `json:"trace"`
}

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