kafka

Documentation

Index

• Variables
• func AddXOriginIfMissing(message *[]sarama.RecordHeader) error
• func AddXTrace(message *[]sarama.RecordHeader, value string) error
• func GetKafkaStats() (send uint64, received uint64)
• func NewConsumerGroupHandler(queue *chan Message) sarama.ConsumerGroupHandler
• type Client
  • 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 Message) (err error)
  • func (c *Client) GetMessages() <-chan Message
  • func (c *Client) Ready() bool
  • func (c *Client) TopicCreator(topic string) (err error)
• type ConsumerGroupHandler
  • 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 Message
  • func ToKafkaMessage(topic string, message interface{}) (Message, error)
• type NewClientOptions
• type SenderTag
• type TraceValue
  • func GetTrace(message *[]sarama.RecordHeader, key string) *TraceValue

Variables

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

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

Functions

func AddXOriginIfMissing

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

func AddXTrace

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

func GetKafkaStats

func GetKafkaStats() (send uint64, received uint64)

func NewConsumerGroupHandler

func NewConsumerGroupHandler(queue *chan Message) sarama.ConsumerGroupHandler

Types

type Client

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

func NewKafkaClient

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

func (*Client) ChangeSubscribedTopics

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

func (*Client) Close

func (c *Client) Close() error

func (*Client) EnqueueMessage

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

func (*Client) GetMessages

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

func (*Client) Ready

func (c *Client) Ready() bool

func (*Client) TopicCreator

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

type ConsumerGroupHandler

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

func (ConsumerGroupHandler) Cleanup

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

func (ConsumerGroupHandler) ConsumeClaim

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

func (ConsumerGroupHandler) Setup

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

type Message

type Message struct {
	Topic  string
	Value  []byte
	Header map[string][]byte
	Key    []byte
}

func ToKafkaMessage

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

type NewClientOptions

type NewClientOptions struct {
	ListenTopicRegex  *regexp.Regexp
	SenderTag         SenderTag
	ConsumerName      string
	Brokers           []string
	StartOffset       int64
	Partitions        int32
	ReplicationFactor int16
	EnableTLS         bool
	ClientID          string
}

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.

type SenderTag

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

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

func GetTrace

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