consumer

Jan 6, 2020

Changes in this version

+ var ErrNoQueueAddresses = errors.New("no kafka-rest-proxy addresses configured")

+ type AgeingClient struct

+ HTTPClient *http.Client

+ Logger *log.UPPLogger

+ MaxAge time.Duration

+ func NewAgeingClient(client *http.Client, maxAge time.Duration, logger *log.UPPLogger) (*AgeingClient, error)

+ func (c AgeingClient) StartAgeingProcess()

+ type Consumer struct

+ func (c *Consumer) ConnectivityCheck() (string, error)

+ func (c *Consumer) Start()

+ func (c *Consumer) Stop()

+ type Message struct

+ Body string

+ Headers map[string]string

+ type MessageConsumer interface

+ ConnectivityCheck func() (string, error)

+ Start func()

+ Stop func()

+ func NewAgeingConsumer(config QueueConfig, handler func(m Message), client *AgeingClient) MessageConsumer

+ func NewBatchedConsumer(config QueueConfig, handler func(m []Message), client *http.Client, ...) MessageConsumer

+ func NewConsumer(config QueueConfig, handler func(m Message), client *http.Client, ...) MessageConsumer

+ type QueueConfig struct

+ Addrs []string

+ AuthorizationKey string

+ AutoCommitEnable bool

+ BackoffPeriod int

+ ConcurrentProcessing bool

+ Group string

+ NoOfProcessors int

+ Offset string

+ Queue string

+ StreamCount int

+ Topic string