kafka

Documentation

Index

• type Config
• type MsgHandler
• type Publisher
  • func NewPublisher(c Config, opts ...kgo.Opt) (*Publisher, error)
  • func (p *Publisher) GetKafkaProducer() *sdkkafka.Client
  • func (p *Publisher) Produce(ctx context.Context, rec *kgo.Record, fn func(record *kgo.Record, err error))
  • func (p *Publisher) ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults
  • func (p *Publisher) Publish(ctx context.Context, rec *kgo.Record, fn func(record *kgo.Record, err error))
  • func (p *Publisher) Stop(ctx context.Context) error
• type Subscriber
  • func NewSubscriber(c Config, opts ...kgo.Opt) (*Subscriber, error)
  • func (s *Subscriber) Subscribe(ctx context.Context) chan error
  • func (s *Subscriber) Unsubscribe() error

Types

type Config

type Config struct {
	// Application name that will be used in a serviceName provided to tracer spans
	ApplicationName string
	// Kafka configuration provided by go-sdk
	KafkaConfig pubsub.Kafka
	// AWS session reference, it will be used in case AWS MSK IAM authentication mechanism is used
	//
	// Deprecated: Use AwsConfig instead
	AwsSession *session.Session
	// MsgHandler is a function that will be called when a message is received
	MsgHandler MsgHandler
	// AWS configuration reference, it will be used in case AWS MSK IAM authentication mechanism is used
	AwsConfig *aws.Config
	Logger    sdklogger.Logger
}

Config provides a common configuration for Kafka PubSub clients.

type MsgHandler

type MsgHandler func(msg *kgo.Record)

type Publisher

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

func NewPublisher

func NewPublisher(c Config, opts ...kgo.Opt) (*Publisher, error)

NewPublisher is a tiny wrapper around the go-sdk kafka.Client and provides API to Publish kafka messages.

func (*Publisher) GetKafkaProducer

func (p *Publisher) GetKafkaProducer() *sdkkafka.Client

GetKafkaProducer returns underlying kafka.Producer for fine-grained tuning purposes.

func (*Publisher) Produce

func (p *Publisher) Produce(ctx context.Context, rec *kgo.Record, fn func(record *kgo.Record, err error))

Produce is an alias to Publish to satisfy kafka go-kit transport.

func (*Publisher) ProduceSync

func (p *Publisher) ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults

ProduceSync publishes kgo.Record messages synchronously.

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, rec *kgo.Record, fn func(record *kgo.Record, err error))

Publish publishes kgo.Record message.

func (*Publisher) Stop

func (p *Publisher) Stop(ctx context.Context) error

Stop flushes and waits for outstanding messages and requests to complete delivery. It also closes a Producer instance.

type Subscriber

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

func NewSubscriber

func NewSubscriber(c Config, opts ...kgo.Opt) (*Subscriber, error)

NewSubscriber is a tiny wrapper around the sdk kafka.Client and provides API to subscribe to a kafka topic.

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context) chan error

Subscribe subscribes to a configured topic and reads messages. Returns unbuffered channel to inspect possible errors.

func (*Subscriber) Unsubscribe

func (s *Subscriber) Unsubscribe() error