kafka

Documentation

Index

• Constants
• func SplitBrokerString(S string) []string
• type ClientConfig
  • func DefaultConfig() *ClientConfig
  • func SetDefaultConfig(c *ClientConfig) *ClientConfig
  • func (c *ClientConfig) Copy() *ClientConfig
  • func (c *ClientConfig) Validate() error
• type Log
  • func New(cfg *ClientConfig) (*Log, error)
  • func (l *Log) Close() error
  • func (l *Log) LogMessage(ctx context.Context, m logger.Message) error

Constants

const (
	DefaultBrokerAddr   = "localhost:9092"
	DefaultBufferSize   = 1024
	DefaultNumProducers = 1
	DefaultNumWorkers   = 2
)

The default option values.

const (
	EnvBrokerAddrs = "PCAS_KAFKA_BROKERS"
)

The environment variables consulted

Functions

func SplitBrokerString

func SplitBrokerString(S string) []string

SplitBrokerString splits S on commas, returning the distinct non-empty trimmed pieces.

Types

type ClientConfig

type ClientConfig struct {
	Formatter    logger.Formatter // The formatter for the log messages
	BrokerAddrs  []string         // The broker addresses for the Kafka producers
	BufferSize   int              // The size of the message buffer
	NumProducers int              // The number of Kafka producers
	NumWorkers   int              // The number of workers feeding log messages to the producers
	Log          log.Interface    // The destination log for debugging
}

ClientConfig describes the options that the user can set on a client connection

func DefaultConfig

func DefaultConfig() *ClientConfig

DefaultConfig returns a new client configuration initialised with the default values. The initial default value for the broker addresses will be read from the environment variable PCAS_KAFKA_BROKERS.

func SetDefaultConfig

func SetDefaultConfig(c *ClientConfig) *ClientConfig

SetDefaultConfig sets the default client configuration to c and returns the old default configuration. This change will be reflected in future calls to DefaultConfig.

func (*ClientConfig) Copy

func (c *ClientConfig) Copy() *ClientConfig

Copy returns a deep copy of the configuration.

func (*ClientConfig) Validate

func (c *ClientConfig) Validate() error

Validate validates the client configuration, returning an error if there's a problem.

type Log

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

Log is a logger that sends log messages to Kafka.

func New

func New(cfg *ClientConfig) (*Log, error)

New returns a Kafka logger. It is the caller's responsibility to call Close on the returned *Log when finished with it, otherwise resources will leak.

func (*Log) Close

func (l *Log) Close() error

Close closes the logger.

func (*Log) LogMessage

func (l *Log) LogMessage(ctx context.Context, m logger.Message) error

LogMessage submits the given log message to Kafka.