core

Documentation

Index

• Constants
• type Admin
• type AsyncProducer
• type Consumer
• type ConsumerHandler
• type ConsumerMessage
  • func (m ConsumerMessage) String() string
• type Message
  • func (m Message) String() string
• type MessageHeader
  • func (m MessageHeader) String() string
• type ProducerError
  • func (pe ProducerError) Error() string
  • func (pe ProducerError) Unwrap() error
• type SyncProducer
• type TopicConfiguration

Constants

const SecurityProtocolTls = "TLS"

Types

type Admin

type Admin interface {

	// CreateTopics create multiple topics at once with custom configurations.
	// Returns error if any error occurred
	CreateTopics(configurations []TopicConfiguration) error

	// DeleteTopics delete multiple topics at once.
	// Returns error if any error occurred
	DeleteTopics(topics []string) error

	// DeleteGroups delete multiple groups at once.
	// Returns error if any error occurred
	DeleteGroups(groupIds []string) error
}

type AsyncProducer

type AsyncProducer interface {

	// Send a message to the brokers
	Send(m *Message)

	// Successes is the success output channel back to the user
	Successes() <-chan *Message

	// Errors is the error output channel back to the user
	Errors() <-chan *ProducerError

	// Close the producer
	Close() error
}

AsyncProducer publishes messages to the brokers

type Consumer

type Consumer interface {
	Start(ctx context.Context)
	WaitForReady() chan bool
	Stop()
}

type ConsumerHandler

type ConsumerHandler interface {
	HandlerFunc(*ConsumerMessage)
	Close()
}

type ConsumerMessage

type ConsumerMessage struct {
	Topic     string
	Key       []byte
	Value     []byte
	Headers   []MessageHeader
	Partition int32
	Offset    int64
	Timestamp time.Time
}

func (ConsumerMessage) String

func (m ConsumerMessage) String() string

type Message

type Message struct {
	Topic     string
	Key       []byte
	Value     []byte
	Headers   []MessageHeader
	Metadata  interface{}
	Partition int32
	Offset    int64
	Timestamp time.Time
}

func (Message) String

func (m Message) String() string

type MessageHeader

type MessageHeader struct {
	Key   []byte
	Value []byte
}

func (MessageHeader) String

func (m MessageHeader) String() string

type ProducerError

type ProducerError struct {
	Msg *Message
	Err error
}

ProducerError is the type of error generated when the producer fails to deliver a message. It contains the original Message as well as the actual error value.

func (ProducerError) Error

func (pe ProducerError) Error() string

func (ProducerError) Unwrap

func (pe ProducerError) Unwrap() error

type SyncProducer

type SyncProducer interface {

	// Send a message to the brokers
	Send(m *Message) (partition int32, offset int64, err error)

	// Close the producer
	Close() error
}

SyncProducer publishes messages to the brokers

type TopicConfiguration

type TopicConfiguration struct {
	Name          string
	Partitions    int32 `default:"1"`
	ReplicaFactor int16 `default:"1"`
	Retention     time.Duration
}