kafka_sarama

Documentation

Overview

Package kafka implements the Kafka CloudEvents binding.

Index

• Constants
• func EncodeKafkaProducerMessage(ctx context.Context, m binding.Message, ...) error
• func NewMessage(cm *sarama.ConsumerMessage) (binding.Message, error)
• func NewMessageFromRaw(key []byte, value []byte, contentType string, headers map[string][]byte) (binding.Message, error)
• func WithSkipKeyExtension(ctx context.Context) context.Context
• type Message
  • func (m *Message) Binary(ctx context.Context, encoder binding.BinaryEncoder) error
  • func (m *Message) Encoding() binding.Encoding
  • func (m *Message) Finish(error) error
  • func (m *Message) Structured(ctx context.Context, encoder binding.StructuredEncoder) error
• type Receiver
  • func NewReceiver(client sarama.Client, groupId string, topic string) *Receiver
  • func (r *Receiver) Cleanup(sarama.ConsumerGroupSession) error
  • func (r *Receiver) Close(ctx context.Context) error
  • func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (r *Receiver) Receive(ctx context.Context) (binding.Message, error)
  • func (r *Receiver) Setup(sess sarama.ConsumerGroupSession) error
• type Sender
  • func NewSender(client sarama.Client, topic string, options ...SenderOptionFunc) (*Sender, error)
  • func (s *Sender) Close(ctx context.Context) error
  • func (s *Sender) Send(ctx context.Context, m binding.Message) error
• type SenderOptionFunc
  • func WithTranscoder(factory binding.TransformerFactory) SenderOptionFunc

Constants

const ContentType = "content-type"

const (
	SKIP_KEY_EXTENSION = "SKIP_KEY_EXTENSION"
)

Functions

func EncodeKafkaProducerMessage

func EncodeKafkaProducerMessage(ctx context.Context, m binding.Message, producerMessage *sarama.ProducerMessage, transformerFactories binding.TransformerFactories) error

Fill the provided producerMessage with the message m. Using context you can tweak the encoding processing (more details on binding.Translate documentation).

func NewMessage

func NewMessage(cm *sarama.ConsumerMessage) (binding.Message, error)

NewMessage returns a Message with data from body. Reads and closes body.

func NewMessageFromRaw

func NewMessageFromRaw(key []byte, value []byte, contentType string, headers map[string][]byte) (binding.Message, error)

func WithSkipKeyExtension

func WithSkipKeyExtension(ctx context.Context) context.Context

Types

type Message

type Message struct {
	Key, Value  []byte
	Headers     map[string][]byte
	ContentType string
	// contains filtered or unexported fields
}

Message holds a sarama ConsumerMessage.

func (*Message) Binary

func (m *Message) Binary(ctx context.Context, encoder binding.BinaryEncoder) error

func (*Message) Encoding

func (m *Message) Encoding() binding.Encoding

func (*Message) Finish

func (m *Message) Finish(error) error

func (*Message) Structured

func (m *Message) Structured(ctx context.Context, encoder binding.StructuredEncoder) error

type Receiver

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

func NewReceiver

func NewReceiver(client sarama.Client, groupId string, topic string) *Receiver

func (*Receiver) Cleanup

func (r *Receiver) Cleanup(sarama.ConsumerGroupSession) error

func (*Receiver) Close

func (r *Receiver) Close(ctx context.Context) error

func (*Receiver) ConsumeClaim

func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*Receiver) Receive

func (r *Receiver) Receive(ctx context.Context) (binding.Message, error)

func (*Receiver) Setup

func (r *Receiver) Setup(sess sarama.ConsumerGroupSession) error

type Sender

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

func NewSender

func NewSender(client sarama.Client, topic string, options ...SenderOptionFunc) (*Sender, error)

func (*Sender) Close

func (s *Sender) Close(ctx context.Context) error

func (*Sender) Send

func (s *Sender) Send(ctx context.Context, m binding.Message) error

type SenderOptionFunc

type SenderOptionFunc func(sender *Sender)

func WithTranscoder

func WithTranscoder(factory binding.TransformerFactory) SenderOptionFunc