kafkax

Documentation

Index

• Constants
• func ConsumerLog(logger log.Logger) consumer.Middleware
• func ConsumerRequestId() consumer.Middleware
• func GetHeaderValue(headers []kafka.Header, key string) string
• func NewResultHandler(logger log.Logger, adapter handler.SyncHandlerAdapter) handler.Sync
• func PlainAuth(auth *Auth) sasl.Mechanism
• func PublisherLog(logger log.Logger) publisher.Middleware
• func PublisherMetrics(storage PublisherMetricStorage) publisher.Middleware
• func PublisherRequestId() publisher.Middleware
• type Auth
• type Client
  • func New(logger log.Logger) *Client
  • func (c *Client) Close()
  • func (c *Client) Healthcheck(ctx context.Context) error
  • func (c *Client) UpgradeAndServe(ctx context.Context, config Config) error
• type Config
  • func NewConfig(opts ...ConfigOption) Config
• type ConfigOption
  • func WithConsumers(consumers ...consumer.Consumer) ConfigOption
  • func WithPublishers(publishers ...*publisher.Publisher) ConfigOption
• type ConsumerConfig
  • func (c ConsumerConfig) DefaultConsumer(logger log.Logger, handler consumer.Handler, ...) consumer.Consumer
  • func (c ConsumerConfig) GetCommitInterval() time.Duration
  • func (c ConsumerConfig) GetMaxBatchSizeMb() int
• type KafkaClient
• type LogObserver
  • func NewLogObserver(ctx context.Context, logger log.Logger) LogObserver
  • func (l LogObserver) ClientError(err error)
  • func (l LogObserver) ClientReady()
  • func (l LogObserver) ShutdownDone()
  • func (l LogObserver) ShutdownStarted()
• type NoopObserver
  • func (n NoopObserver) ClientError(err error)
  • func (n NoopObserver) ClientReady()
  • func (n NoopObserver) ShutdownDone()
  • func (n NoopObserver) ShutdownStarted()
• type Observer
• type PublisherConfig
  • func (p PublisherConfig) DefaultPublisher(logger log.Logger, restMiddlewares ...publisher.Middleware) *publisher.Publisher
  • func (p PublisherConfig) GetBatchSizePerPartition() int
  • func (p PublisherConfig) GetBatchTimeoutPerPartition() time.Duration
  • func (p PublisherConfig) GetMaxMessageSizePerPartition() int64
  • func (p PublisherConfig) GetRequiredAckLevel() kafka.RequiredAcks
  • func (p PublisherConfig) GetWriteTimeout() time.Duration
• type PublisherMetricStorage

Constants

const RequestIdHeader = "x-request-id"

Functions

func ConsumerLog

func ConsumerLog(logger log.Logger) consumer.Middleware

func ConsumerRequestId

func ConsumerRequestId() consumer.Middleware

func GetHeaderValue

func GetHeaderValue(headers []kafka.Header, key string) string

func NewResultHandler

func NewResultHandler(logger log.Logger, adapter handler.SyncHandlerAdapter) handler.Sync

func PlainAuth

func PlainAuth(auth *Auth) sasl.Mechanism

func PublisherLog

func PublisherLog(logger log.Logger) publisher.Middleware

func PublisherMetrics

func PublisherMetrics(storage PublisherMetricStorage) publisher.Middleware

func PublisherRequestId

func PublisherRequestId() publisher.Middleware

Types

type Auth

type Auth struct {
	Username string `validate:"required" schema:"Логин"`
	Password string `validate:"required" schema:"Пароль"`
}

type Client

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

func New

func New(logger log.Logger) *Client

func (*Client) Close

func (c *Client) Close()

func (*Client) Healthcheck

func (c *Client) Healthcheck(ctx context.Context) error

func (*Client) UpgradeAndServe

func (c *Client) UpgradeAndServe(ctx context.Context, config Config) error

type Config

type Config struct {
	Publishers []*publisher.Publisher
	Consumers  []consumer.Consumer
}

func NewConfig

func NewConfig(opts ...ConfigOption) Config

type ConfigOption

type ConfigOption func(c *Config)

func WithConsumers

func WithConsumers(consumers ...consumer.Consumer) ConfigOption

func WithPublishers

func WithPublishers(publishers ...*publisher.Publisher) ConfigOption

type ConsumerConfig

type ConsumerConfig struct {
	Addresses         []string `validate:"required" schema:"Список адресов брокеров для чтения сообщений"`
	Topic             string   `validate:"required" schema:"Топик"`
	GroupId           string   `validate:"required" schema:"Идентификатор консьюмера"`
	Concurrency       int      `schema:"Кол-во обработчиков, по умолчанию 1"`
	MaxBatchSizeMb    int      `` /* 133-byte string literal not displayed */
	CommitIntervalSec *int     `` /* 141-byte string literal not displayed */
	Auth              *Auth    `schema:"Параметры аутентификации"`
}

func (ConsumerConfig) DefaultConsumer

func (c ConsumerConfig) DefaultConsumer(
	logger log.Logger,
	handler consumer.Handler,
	restMiddlewares ...consumer.Middleware,
) consumer.Consumer

func (ConsumerConfig) GetCommitInterval

func (c ConsumerConfig) GetCommitInterval() time.Duration

func (ConsumerConfig) GetMaxBatchSizeMb

func (c ConsumerConfig) GetMaxBatchSizeMb() int

type KafkaClient

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

type LogObserver

type LogObserver struct {
	NoopObserver
	// contains filtered or unexported fields
}

func NewLogObserver

func NewLogObserver(ctx context.Context, logger log.Logger) LogObserver

func (LogObserver) ClientError

func (l LogObserver) ClientError(err error)

func (LogObserver) ClientReady

func (l LogObserver) ClientReady()

func (LogObserver) ShutdownDone

func (l LogObserver) ShutdownDone()

func (LogObserver) ShutdownStarted

func (l LogObserver) ShutdownStarted()

type NoopObserver

type NoopObserver struct{}

func (NoopObserver) ClientError

func (n NoopObserver) ClientError(err error)

func (NoopObserver) ClientReady

func (n NoopObserver) ClientReady()

func (NoopObserver) ShutdownDone

func (n NoopObserver) ShutdownDone()

func (NoopObserver) ShutdownStarted

func (n NoopObserver) ShutdownStarted()

type Observer

type Observer interface {
	ClientReady()
	ClientError(err error)
	ShutdownStarted()
	ShutdownDone()
}

type PublisherConfig

type PublisherConfig struct {
	Addresses                  []string `validate:"required" schema:"Список адресов брокеров для отправки сообщений"`
	Topic                      string   `` /* 160-byte string literal not displayed */
	MaxMsgSizeMbPerPartition   int64    `schema:"Максимальный размер сообщений в Мб, по умолчанию 64 Мб"`
	BatchSizePerPartition      int      `` /* 142-byte string literal not displayed */
	BatchTimeoutPerPartitionMs *int     `schema:"Периодичность записи батчей в кафку в мс, по умолчанию 500 мс"`
	WriteTimeoutSec            *int     `schema:"Таймаут отправки сообщений, по умолчанию 10 секунд"`
	RequiredAckLevel           int      `` /* 179-byte string literal not displayed */
	Auth                       *Auth    `schema:"Параметры аутентификации"`
}

func (PublisherConfig) DefaultPublisher

func (p PublisherConfig) DefaultPublisher(logger log.Logger, restMiddlewares ...publisher.Middleware) *publisher.Publisher

func (PublisherConfig) GetBatchSizePerPartition

func (p PublisherConfig) GetBatchSizePerPartition() int

func (PublisherConfig) GetBatchTimeoutPerPartition

func (p PublisherConfig) GetBatchTimeoutPerPartition() time.Duration

func (PublisherConfig) GetMaxMessageSizePerPartition

func (p PublisherConfig) GetMaxMessageSizePerPartition() int64

func (PublisherConfig) GetRequiredAckLevel

func (p PublisherConfig) GetRequiredAckLevel() kafka.RequiredAcks

func (PublisherConfig) GetWriteTimeout

func (p PublisherConfig) GetWriteTimeout() time.Duration

type PublisherMetricStorage

type PublisherMetricStorage interface {
	ObservePublishDuration(topic string, t time.Duration)
	ObservePublishMsgSize(topic string, size int)
	IncPublishError(topic string)
}