librdkafka

Documentation

Index

• Variables
• func UnmarshalBuffer(bufferPtr *C.buffer_t) []byte
• type Client
  • func NewProducer(config Config, handler MessageDelivery) (*Client, error)
  • func (cl *Client) Close()
  • func (cl *Client) GetAllocCounter() int64
  • func (cl *Client) Poll(timeout time.Duration)
• type Config
  • func NewConfig() Config
  • func (c *Config) Close()
  • func (c *Config) Set(key, value string)
  • func (c *Config) SetB(key string, value bool)
  • func (c *Config) SetI(key string, value int)
• type ErrorHandle
  • func (l *ErrorHandle) Error() string
• type Message
• type MessageDelivery
• type ResponseError
  • func (r ResponseError) Error() string
• type Topic
  • func NewTopic(name string, config TopicConfig, client *Client) *Topic
  • func (t *Topic) Close()
  • func (t *Topic) GetName() string
  • func (t *Topic) Produce(msg Message) error
• type TopicConfig
  • func NewTopicConfig() TopicConfig
  • func (c *TopicConfig) Close()
  • func (c *TopicConfig) Set(key, value string)
  • func (c *TopicConfig) SetB(key string, value bool)
  • func (c *TopicConfig) SetI(key string, value int)
  • func (c *TopicConfig) SetRandomPartitioner()
  • func (c *TopicConfig) SetRoundRobinPartitioner()

Variables

var (
	// Log is the standard logger used for non-message related errors
	Log = log.New(os.Stderr, "librdkafka: ", log.Lshortfile)
)

Functions

func UnmarshalBuffer

func UnmarshalBuffer(bufferPtr *C.buffer_t) []byte

UnmarshalBuffer creates a byte slice copy from a buffer_t handle.

Types

type Client

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

Client is a wrapper handle for rd_kafka_t

func NewProducer

func NewProducer(config Config, handler MessageDelivery) (*Client, error)

NewProducer creates a new librdkafka client in producer mode. Make sure to call Close() when done.

func (*Client) Close

func (cl *Client) Close()

Close frees the native handle.

func (*Client) GetAllocCounter

func (cl *Client) GetAllocCounter() int64

GetAllocCounter returns the number of allocated native buffers

func (*Client) Poll

func (cl *Client) Poll(timeout time.Duration)

Poll polls for new data to be sent to the async handler functions

type Config

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

Config is a wrapper for rd_kafka_conf_t

func NewConfig

func NewConfig() Config

NewConfig creates a new main config wrapper. Make sure to call Close() when done.

func (*Config) Close

func (c *Config) Close()

Close frees up the native handle

func (*Config) Set

func (c *Config) Set(key, value string)

Set sets a string value in this config

func (*Config) SetB

func (c *Config) SetB(key string, value bool)

SetB sets a boolean value in this config

func (*Config) SetI

func (c *Config) SetI(key string, value int)

SetI sets an integer value in this config

type ErrorHandle

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

ErrorHandle is a convenience wrapper for handling librdkafka native errors. This struct fulfills the standard golang error interface.

func (*ErrorHandle) Error

func (l *ErrorHandle) Error() string

type Message

type Message interface {
	GetKey() []byte
	GetPayload() []byte
	GetUserdata() []byte
}

Message is the interface used to send exchange data with kafka

type MessageDelivery

type MessageDelivery interface {
	OnMessageError(reason string, userdata []byte)
	OnMessageDelivered(userdata []byte)
}

MessageDelivery is used to handle message delivery errors

type ResponseError

type ResponseError struct {
	Userdata []byte
	Code     int
}

ResponseError is used as a wrapper for errors generated by the batch producer. The Code member wraps directly to the librdkafka error number. The original message is attached to allow backtracking.

func (ResponseError) Error

func (r ResponseError) Error() string

type Topic

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

Topic wrapper handle for rd_kafka_topic_t

func NewTopic

func NewTopic(name string, config TopicConfig, client *Client) *Topic

NewTopic creates a new topic representation in librdkafka. You have to call Close() to free any internal state. As this struct holds a pointer to the client make sure that Client.Close is called after closing objects of this type.

func (*Topic) Close

func (t *Topic) Close()

Close frees the internal handle and tries to flush the queue.

func (*Topic) GetName

func (t *Topic) GetName() string

GetName returns the name of the topic

func (*Topic) Produce

func (t *Topic) Produce(msg Message) error

Produce produces a single messages. If a message cannot be produced because of internal (non-wire) problems an error is immediately returned instead of being asynchronously handled via MessageDelivery interface.

type TopicConfig

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

TopicConfig is a wrapper for rd_kafka_topic_conf_t

func NewTopicConfig

func NewTopicConfig() TopicConfig

NewTopicConfig creates a new topic config wrapper. Make sure to call Close() when done.

func (*TopicConfig) Close

func (c *TopicConfig) Close()

Close frees up the native handle

func (*TopicConfig) Set

func (c *TopicConfig) Set(key, value string)

Set sets a string value in this config

func (*TopicConfig) SetB

func (c *TopicConfig) SetB(key string, value bool)

SetB sets a boolean value in this config

func (*TopicConfig) SetI

func (c *TopicConfig) SetI(key string, value int)

SetI sets an integer value in this config

func (*TopicConfig) SetRandomPartitioner

func (c *TopicConfig) SetRandomPartitioner()

SetRandomPartitioner registers the random partitioner function.

func (*TopicConfig) SetRoundRobinPartitioner

func (c *TopicConfig) SetRoundRobinPartitioner()

SetRoundRobinPartitioner registers the round robin partitioner function.