redpanda

Documentation

Index

• type AdminAuthOpts
• type AdminClient
  • func NewAdminClient(adminEndpoint string, kafkaBrokers string, opts *AdminAuthOpts) AdminClient
• type Consumer
  • func NewConsumer(brokerHosts string, consumerGroup string, topicName string, ...) (Consumer, error)
• type ConsumerError
• type ConsumerOpts
• type KafkaMessage
• type KafkaSASLAuth
• type Producer
  • func NewProducer(brokerHosts string, producerOpts ProducerOpts) (Producer, error)
• type ProducerOpts
• type ProducerOutput
• type ReaderFunc
• type SASLMechanism

Types

type AdminAuthOpts

type AdminAuthOpts struct {
	Username string
	Password string
}

type AdminClient

type AdminClient interface {
	CreateUser(username, password string) error
	DeleteUser(username string) error
	UserExists(username string) (bool, error)
	CreateTopic(topicName string, partitionCount int) error
	DeleteTopic(topicName string) error
	TopicExists(topicName string) (bool, error)
	AllowUserOnTopics(username string, allowedOperations string, topicNames ...string) error
}

func NewAdminClient

func NewAdminClient(adminEndpoint string, kafkaBrokers string, opts *AdminAuthOpts) AdminClient

type Consumer

type Consumer interface {
	Ping(ctx context.Context) error
	Close()
	StartConsuming(reader ReaderFunc)
}

func NewConsumer

func NewConsumer(brokerHosts string, consumerGroup string, topicName string, options ConsumerOpts) (Consumer, error)

type ConsumerError

type ConsumerError interface {
	GetKey() string
	error
}

type ConsumerOpts

type ConsumerOpts struct {
	SASLAuth       *KafkaSASLAuth
	Logger         logging.Logger
	MaxRetries     *int
	MaxPollRecords *int
}

type KafkaMessage

type KafkaMessage struct {
	Key        []byte
	Value      []byte
	Timestamp  time.Time
	Topic      string
	Partition  int32
	ProducerId int64
	Offset     int64
}

type KafkaSASLAuth

type KafkaSASLAuth struct {
	SASLMechanism SASLMechanism
	User          string
	Password      string
}

type Producer

type Producer interface {
	Ping(ctx context.Context) error
	Close()
	Produce(ctx context.Context, topic, key string, value []byte) (*ProducerOutput, error)
}

func NewProducer

func NewProducer(brokerHosts string, producerOpts ProducerOpts) (Producer, error)

type ProducerOpts

type ProducerOpts struct {
	SASLAuth *KafkaSASLAuth
	Logger   logging.Logger
}

type ProducerOutput

type ProducerOutput struct {
	Key        []byte    `json:"key,omitempty"`
	Timestamp  time.Time `json:"timestamp"`
	Topic      string    `json:"topic"`
	Partition  int32     `json:"partition,omitempty"`
	ProducerId int64     `json:"producerId,omitempty"`
	Offset     int64     `json:"offset"`
}

type ReaderFunc

type ReaderFunc func(msg KafkaMessage) error

type SASLMechanism

type SASLMechanism string

const (
	ScramSHA256 SASLMechanism = "SCRAM-SHA-256"
	ScramSHA512 SASLMechanism = "SCRAM-SHA-512"
)