kafka

Documentation

Index

• func DefaultGroupName() string
• type Client
  • func NewClient(cfg *Config) (*Client, error)
  • func (client *Client) ProducerLoop()
  • func (client *Client) Send(event deployment.Event)
• type Config
• type SASL
• type TLS

Functions

func DefaultGroupName

func DefaultGroupName() string

DefaultGroupName returns a automatically generated group name based on the host

Types

type Client

type Client struct {
	RecvQ         chan sarama.ConsumerMessage
	Producer      sarama.SyncProducer
	ProducerTopic string
	SendQueue     chan deployment.Event
}

Client is a Kafka client

func NewClient

func NewClient(cfg *Config) (*Client, error)

NewClient takes a Kafka config object and returns a new client, or an error if the config is invalid.

func (*Client) ProducerLoop

func (client *Client) ProducerLoop()

ProducerLoop sends messages from the event queue in perpetuity

func (*Client) Send

func (client *Client) Send(event deployment.Event)

Send sends a message to Kafka

type Config

type Config struct {
	Enabled      bool     `json:"enabled"`
	Brokers      []string `json:"brokers"`
	Topic        string   `json:"topic"`
	ClientID     string   `json:"client-id"`
	GroupID      string   `json:"group-id"`
	LogVerbosity string   `json:"log-verbosity"`
	TLS          TLS      `json:"tls"`
	SASL         SASL     `json:"sasl"`
}

Config is the Kafka configuration

type SASL

type SASL struct {
	Enabled   bool   `json:"enabled"`
	Handshake bool   `json:"handshake"`
	Username  string `json:"username"`
	Password  string `json:"password"`
}

SASL config

type TLS

type TLS struct {
	Enabled  bool `json:"enabled"`
	Insecure bool `json:"insecure"`
}

TLS config