consumerserver

Documentation

Index

• Constants
• func DefaultConfig() *config
• type BatchHandler
• type BatchListener
  • func (l *BatchListener) Handle(ctx context.Context, message *ekafka.Message, optFuncs ...handleOption) (bool, error)
• type Component
  • func NewConsumerServerComponent(name string, config *config, ekafkaComponent *ekafka.Component, ...) *Component
  • func (cmp *Component) Consumer() *ekafka.Consumer
  • func (cmp *Component) ConsumerGroup() *ekafka.ConsumerGroup
  • func (cmp *Component) GracefulStop(ctx context.Context) error
  • func (cmp *Component) Info() *server.ServiceInfo
  • func (cmp *Component) Init() error
  • func (cmp *Component) Name() string
  • func (cmp *Component) OnConsumeEachMessage(handler OnConsumeEachMessageHandler) error
  • func (cmp *Component) OnConsumerGroupStart(handler OnConsumerGroupStartHandler) error
  • func (cmp *Component) OnEachMessage(consumptionErrors chan<- error, handler OnEachMessageHandler) error
  • func (cmp *Component) OnStart(handler OnStartHandler) error
  • func (cmp *Component) PackageName() string
  • func (cmp *Component) Start() error
  • func (cmp *Component) Stop() error
  • func (cmp *Component) Subscribe(listener Listener)
  • func (cmp *Component) SubscribeBatchHandler(handler BatchHandler, batchSize int, timeout time.Duration)
  • func (cmp *Component) SubscribeSingleHandler(handler Handler)
• type Container
  • func DefaultContainer() *Container
  • func Load(key string) *Container
  • func (c *Container) Build(options ...Option) *Component
• type Handler
• type Listener
• type OnConsumeEachMessageHandler
• type OnConsumerGroupStartHandler
• type OnEachMessageHandler
• type OnStartHandler
• type Option
  • func WithDebug(debug bool) Option
  • func WithEkafka(ekafkaComponent *ekafka.Component) Option
• type SyncListener
  • func (l *SyncListener) Handle(ctx context.Context, message *ekafka.Message, opts ...handleOption) (bool, error)

Constants

const PackageName = "component.ekafka.consumerserver"

PackageName is the name of this component.

Functions

func DefaultConfig

func DefaultConfig() *config

DefaultConfig returns a default config.

Types

type BatchHandler

type BatchHandler func(lastCtx context.Context, messages []*ekafka.CtxMessage) error

type BatchListener

type BatchListener struct {
	Batch           []*ekafka.CtxMessage
	BatchUpdateSize int
	Timeout         time.Duration
	Handler         BatchHandler
	// contains filtered or unexported fields
}

func (*BatchListener) Handle

func (l *BatchListener) Handle(ctx context.Context, message *ekafka.Message, optFuncs ...handleOption) (bool, error)

type Component

type Component struct {
	ServerCtx context.Context
	// contains filtered or unexported fields
}

Component starts an Ego server for message consuming.

func NewConsumerServerComponent

func NewConsumerServerComponent(name string, config *config, ekafkaComponent *ekafka.Component, logger *elog.Component) *Component

NewConsumerServerComponent creates a new server instance.

func (*Component) Consumer

func (cmp *Component) Consumer() *ekafka.Consumer

Consumer returns the default Consumer.

func (*Component) ConsumerGroup

func (cmp *Component) ConsumerGroup() *ekafka.ConsumerGroup

ConsumerGroup returns the default ConsumerGroup.

func (*Component) GracefulStop

func (cmp *Component) GracefulStop(ctx context.Context) error

GracefulStop stops the server.

func (*Component) Info

func (cmp *Component) Info() *server.ServiceInfo

Info returns server info, used by governor and consumer balancer.

func (*Component) Init

func (cmp *Component) Init() error

Init ...

func (*Component) Name

func (cmp *Component) Name() string

Name returns the name of this instance.

func (*Component) OnConsumeEachMessage

func (cmp *Component) OnConsumeEachMessage(handler OnConsumeEachMessageHandler) error

OnConsumeEachMessage register a handler for each message. When the handler returns an error, the message will be retried if the error is ErrRecoverableError else the message will not be committed.

func (*Component) OnConsumerGroupStart

func (cmp *Component) OnConsumerGroupStart(handler OnConsumerGroupStartHandler) error

OnConsumerGroupStart ...

func (*Component) OnEachMessage

func (cmp *Component) OnEachMessage(consumptionErrors chan<- error, handler OnEachMessageHandler) error

OnEachMessage ... Deprecated: use OnConsumeEachMessage instead.

func (*Component) OnStart

func (cmp *Component) OnStart(handler OnStartHandler) error

OnStart ...

func (*Component) PackageName

func (cmp *Component) PackageName() string

PackageName returns the package name.

func (*Component) Start

func (cmp *Component) Start() error

Start will start consuming.

func (*Component) Stop

func (cmp *Component) Stop() error

Stop stops the server.

func (*Component) Subscribe

func (cmp *Component) Subscribe(listener Listener)

Subscribe append a handler for each message.

func (*Component) SubscribeBatchHandler

func (cmp *Component) SubscribeBatchHandler(handler BatchHandler, batchSize int, timeout time.Duration)

SubscribeBatchHandler append a batch listener with this handler for each message. A batch messages will be handled when batch size or timeout reached

func (*Component) SubscribeSingleHandler

func (cmp *Component) SubscribeSingleHandler(handler Handler)

SubscribeSingleHandler append a single listener with this handler for each message

type Container

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

func DefaultContainer

func DefaultContainer() *Container

DefaultContainer 返回默认Container

func Load

func Load(key string) *Container

Load 载入配置，初始化Container

func (*Container) Build

func (c *Container) Build(options ...Option) *Component

Build 构建Container

type Handler

type Handler func(ctx context.Context, message *ekafka.Message) error

type Listener

type Listener interface {
	Handle(ctx context.Context, message *ekafka.Message, opts ...handleOption) (bool, error)
}

type OnConsumeEachMessageHandler

type OnConsumeEachMessageHandler = func(ctx context.Context, message *ekafka.Message) error

OnConsumeEachMessageHandler ...

type OnConsumerGroupStartHandler

type OnConsumerGroupStartHandler = func(ctx context.Context, consumerGroup *ekafka.ConsumerGroup) error

OnConsumerGroupStartHandler ...

type OnEachMessageHandler

type OnEachMessageHandler = func(ctx context.Context, message kafka.Message) error

OnEachMessageHandler ...

type OnStartHandler

type OnStartHandler = func(ctx context.Context, consumer *ekafka.Consumer) error

OnStartHandler ...

type Option

type Option func(c *Container)

func WithDebug

func WithDebug(debug bool) Option

WithDebug enables debug mode.

func WithEkafka

func WithEkafka(ekafkaComponent *ekafka.Component) Option

WithEkafka ...

type SyncListener

type SyncListener struct {
	Handler Handler
	// contains filtered or unexported fields
}

func (*SyncListener) Handle

func (l *SyncListener) Handle(ctx context.Context, message *ekafka.Message, opts ...handleOption) (bool, error)