kafka

README

Kafka

Install

go get github.com/aacfactory/fns-contrib/message-queues/kafka

Usage

app.Deploy(kafka.New(kafka.WithConsumeHandler(name, handleFn)))

Config

kafka:
  brokers: 
    - "192.168.0.1:9092"
  producers:
    enable: true
    num: 4
    partitioner:
      name: "round_robin"
  consumers:
    nameA:
      group: "groupId"
      topics:
        - "topicA"

As proxy

publishErr := kafka.Publish(ctx, kafka.NewMessage(topic, key, body))

Documentation

Index

• Constants
• func New(options ...Option) services.Listenable
• func Publish(ctx context.Context, message ...ProducerMessage) (err error)
• func PublishAsync(ctx context.Context, message ...ProducerMessage) (err error)
• type ConsumeErrorHandler
• type ConsumeHandler
• type Consumer
  • func NewGroupConsumer(name string, log logs.Logger, maxPollRecords int, partitionBuffer int, ...) (v Consumer, err error)
• type ConsumerOptions
• type GroupConsumer
  • func (consumer *GroupConsumer) Listen(ctx context.Context) (err error)
  • func (consumer *GroupConsumer) Shutdown(ctx context.Context)
• type GroupPartitionConsumer
  • func (consumer *GroupPartitionConsumer) Consume(ctx context.Context)
• type Header
• type Headers
  • func (headers Headers) ConvertToKafkaHeaders() (v []kgo.RecordHeader)
• type Meta
• type Option
  • func WithConsumeErrorHandler(handler ConsumeErrorHandler) Option
  • func WithConsumeHandler(name string, handler ConsumeHandler) Option
• type Options
• type Producer
  • func NewProducer(log logs.Logger, num int, opts []kgo.Opt) (v *Producer, err error)
  • func (producer *Producer) Client() (v *kgo.Client)
  • func (producer *Producer) Publish(ctx context.Context, messages []ProducerMessage, async bool) (err error)
  • func (producer *Producer) Shutdown(_ context.Context)
• type ProducerMessage
  • func NewMessage(topic string, key []byte, body []byte) ProducerMessage
  • func (msg ProducerMessage) AddHeader(key string, value []byte) ProducerMessage
  • func (msg ProducerMessage) Validate() (err error)
• type TimestampType
• type TopicPartitionKey

Constants

const (
	UnknownTimestamp = iota - 1
	DefaultTimestamp
	AlternativeTimestamp
)

Functions

func New

func New(options ...Option) services.Listenable

func Publish

func Publish(ctx context.Context, message ...ProducerMessage) (err error)

func PublishAsync

func PublishAsync(ctx context.Context, message ...ProducerMessage) (err error)

Types

type ConsumeErrorHandler

type ConsumeErrorHandler func(topic string, partition int32, cause error)

type ConsumeHandler

type ConsumeHandler func(ctx context.Context, value []byte, meta Meta) (err error)

ConsumeHandler value is uncompressed

type Consumer

type Consumer interface {
	Listen(ctx context.Context) (err error)
	Shutdown(ctx context.Context)
}

func NewGroupConsumer

func NewGroupConsumer(name string, log logs.Logger, maxPollRecords int, partitionBuffer int, opts []kgo.Opt, handler ConsumeHandler, errorHandler ConsumeErrorHandler) (v Consumer, err error)

type ConsumerOptions

type ConsumerOptions struct {
	Log logs.Logger

	Config configs.ConsumerConfig
}

type GroupConsumer

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

func (*GroupConsumer) Listen

func (consumer *GroupConsumer) Listen(ctx context.Context) (err error)

func (*GroupConsumer) Shutdown

func (consumer *GroupConsumer) Shutdown(ctx context.Context)

type GroupPartitionConsumer

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

func (*GroupPartitionConsumer) Consume

func (consumer *GroupPartitionConsumer) Consume(ctx context.Context)

type Header

type Header struct {
	Key   string `json:"key" avro:"key"`
	Value []byte `json:"value" avro:"value"`
}

type Headers

type Headers []Header

func (Headers) ConvertToKafkaHeaders

func (headers Headers) ConvertToKafkaHeaders() (v []kgo.RecordHeader)

type Meta

type Meta struct {
	Topic           string
	Headers         Headers
	Key             []byte
	Partition       int32
	Offset          int64
	Time            time.Time
	ProducerEpoch   int16
	ProducerId      int64
	LeaderEpoch     int32
	TimestampType   TimestampType
	IsTransactional bool
	IsControl       bool
}

type Option

type Option func(options *Options)

func WithConsumeErrorHandler

func WithConsumeErrorHandler(handler ConsumeErrorHandler) Option

func WithConsumeHandler

func WithConsumeHandler(name string, handler ConsumeHandler) Option

type Options

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

type Producer

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

func NewProducer

func NewProducer(log logs.Logger, num int, opts []kgo.Opt) (v *Producer, err error)

func (*Producer) Client

func (producer *Producer) Client() (v *kgo.Client)

func (*Producer) Publish

func (producer *Producer) Publish(ctx context.Context, messages []ProducerMessage, async bool) (err error)

func (*Producer) Shutdown

func (producer *Producer) Shutdown(_ context.Context)

type ProducerMessage

type ProducerMessage struct {
	Topic   string  `json:"topic" avro:"topic"`
	Key     []byte  `json:"key" avro:"key"`
	Body    []byte  `json:"body" avro:"body"`
	Headers Headers `json:"headers" avro:"headers"`
}

func NewMessage

func NewMessage(topic string, key []byte, body []byte) ProducerMessage

func (ProducerMessage) AddHeader

func (msg ProducerMessage) AddHeader(key string, value []byte) ProducerMessage

func (ProducerMessage) Validate

func (msg ProducerMessage) Validate() (err error)

type TimestampType

type TimestampType int8

type TopicPartitionKey

type TopicPartitionKey struct {
	Topic     string
	Partition int32
}