kafka

Documentation

Index

• Variables
• type Option
  • func WithBrokers(brokers ...string) Option
  • func WithClientID(clientID string) Option
  • func WithClusterConfig(clusterConfig *cluster.Config) Option
  • func WithCompression(codec sarama.CompressionCodec, level int) Option
  • func WithErrorHandler(h nc.ErrorHandler) Option
  • func WithFlashFrequency(frequency time.Duration) Option
  • func WithFlashMessages(messageCount int) Option
  • func WithGroupName(name string) Option
  • func WithKafkaURL(urlString string) Option
  • func WithKafkaVersion(version sarama.KafkaVersion) Option
  • func WithPanicHandler(h nc.PanicHandler) Option
  • func WithPublisherErrorHandler(h PublisherErrorHandler) Option
  • func WithPublisherSuccessHandler(h PublisherSuccessHandler) Option
  • func WithSaramaConfig(streamConfig *sarama.Config) Option
  • func WithSubscriberNotificationHandler(h SubscriberNotificationHandler) Option
  • func WithTopics(topics ...string) Option
• type Options
• type Publisher
  • func MustNewPublisher(ctx context.Context, options ...Option) *Publisher
  • func NewPublisher(ctx context.Context, options ...Option) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, messages ...interface{}) (err error)
• type PublisherErrorHandler
• type PublisherSuccessHandler
• type Subscriber
  • func NewSubscriber(options ...Option) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Listen(ctx context.Context) (err error)
• type SubscriberNotificationHandler

Variables

var ErrMessageInvalidConsumer = errors.New(`[message] invalid consumer`)

ErrMessageInvalidConsumer error

Types

type Option

type Option func(options *Options)

Option function type

func WithBrokers

func WithBrokers(brokers ...string) Option

WithBrokers overrides the list of brokers to connect

func WithClientID

func WithClientID(clientID string) Option

WithClientID value

func WithClusterConfig

func WithClusterConfig(clusterConfig *cluster.Config) Option

WithClusterConfig custom config

func WithCompression

func WithCompression(codec sarama.CompressionCodec, level int) Option

WithCompression of the pipe

func WithErrorHandler

func WithErrorHandler(h nc.ErrorHandler) Option

WithErrorHandler set handler of error processing

func WithFlashFrequency

func WithFlashFrequency(frequency time.Duration) Option

WithFlashFrequency of flushing

func WithFlashMessages

func WithFlashMessages(messageCount int) Option

WithFlashMessages minimal count

func WithGroupName

func WithGroupName(name string) Option

WithGroupName of the message consuming

func WithKafkaURL

func WithKafkaURL(urlString string) Option

WithKafkaURL is an Option to set the URL the client should connect to. The url can contain username/password semantics. e.g. kafka://derek:pass@localhost:4222/{groupName}?topics=topic1,topic2 Comma separated arrays are also supported, e.g. urlA, urlB.

func WithKafkaVersion

func WithKafkaVersion(version sarama.KafkaVersion) Option

WithKafkaVersion minimal version

func WithPanicHandler

func WithPanicHandler(h nc.PanicHandler) Option

WithPanicHandler set handler of panic processing

func WithPublisherErrorHandler

func WithPublisherErrorHandler(h PublisherErrorHandler) Option

WithPublisherErrorHandler set handler of the sarama errors

func WithPublisherSuccessHandler

func WithPublisherSuccessHandler(h PublisherSuccessHandler) Option

WithPublisherSuccessHandler set handler of the sarama success

func WithSaramaConfig

func WithSaramaConfig(streamConfig *sarama.Config) Option

WithSaramaConfig custom config

func WithSubscriberNotificationHandler

func WithSubscriberNotificationHandler(h SubscriberNotificationHandler) Option

WithSubscriberNotificationHandler set handler of the cluster group notifications

func WithTopics

func WithTopics(topics ...string) Option

WithTopics will set the list of topics for publishing or subscribing

type Options

type Options struct {
	ClusterConfig cluster.Config

	// IsSynchronous type of producer
	// TODO: make it work for sync publisher
	IsSynchronous bool

	// Brokers contains list of broker hosts with port
	Brokers []string

	// Name of the subscription group
	GroupName string

	// Names of topics for subscribing or publishing
	Topics []string

	// ErrorHandler of message processing
	ErrorHandler nc.ErrorHandler

	// PanicHandler process panic
	PanicHandler nc.PanicHandler

	// PublisherErrorHandler provides handler of message send errors
	PublisherErrorHandler PublisherErrorHandler

	// PublisherSuccessHandler provides handler of message send success
	PublisherSuccessHandler PublisherSuccessHandler

	// SubscriberNotificationHandler provides handler of received messages
	SubscriberNotificationHandler SubscriberNotificationHandler

	// Message encoder interface
	Encoder encoder.Encoder

	// Logger of subscriber
	Logger loggerInterface
}

Options for publisher or subscriber

type Publisher

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

Publisher implementation of Publisher interface for the Kafka driver

func MustNewPublisher

func MustNewPublisher(ctx context.Context, options ...Option) *Publisher

MustNewPublisher connection or panic

func NewPublisher

func NewPublisher(ctx context.Context, options ...Option) (*Publisher, error)

NewPublisher to the kafka with some brokers and topics for sending

func (*Publisher) Close

func (p *Publisher) Close() error

Close kafka producer

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, messages ...interface{}) (err error)

Publish one or more messages to the pub-service

type PublisherErrorHandler

type PublisherErrorHandler func(*sarama.ProducerError)

PublisherErrorHandler callback function

type PublisherSuccessHandler

type PublisherSuccessHandler func(*sarama.ProducerMessage)

PublisherSuccessHandler callback function

type Subscriber

type Subscriber struct {
	notificationcenter.ModelSubscriber
	// contains filtered or unexported fields
}

Subscriber for kafka

func NewSubscriber

func NewSubscriber(options ...Option) (*Subscriber, error)

NewSubscriber connection to kafka "group" from list of topics

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close kafka consumer

func (*Subscriber) Listen

func (s *Subscriber) Listen(ctx context.Context) (err error)

Listen kafka consumer

type SubscriberNotificationHandler

type SubscriberNotificationHandler func(notification *cluster.Notification)

SubscriberNotificationHandler callback function