consumer

Documentation

Index

• Constants
• Variables
• type Event
  • func Ack(offset int64) Event
• type Message
• type Request
  • func NewRequest(group, topic string) Request
• type Response
• type T

Constants

const (
	// An event of this type should be sent to the message events channel
	// when the message is offered to a client.
	EvOffered eventType = iota

	// An event of this type should be sent to the message events channel
	// when the message is acknowledged by a client.
	EvAcked
)

Variables

var (
	ErrRequestTimeout  = errors.New("long polling timeout")
	ErrUnavailable     = errors.New("service is shutting down")
	ErrTooManyRequests = errors.New("Too many requests. Consider increasing `consumer.channel_buffer_size` (https://github.com/mailgun/kafka-pixy/blob/master/default.yaml#L43)")
)

Types

type Event

type Event struct {
	T      eventType
	Offset int64
}

func Ack

func Ack(offset int64) Event

type Message

type Message struct {
	sarama.ConsumerMessage
	HighWaterMark int64
	EventsCh      chan<- Event
}

Message encapsulates a Kafka message returned by the consumer.

type Request

type Request struct {
	Timestamp  time.Time
	Group      string
	Topic      string
	ResponseCh chan Response
}

Request

func NewRequest

func NewRequest(group, topic string) Request

type Response

type Response struct {
	Msg Message
	Err error
}

Response defines responses returned upstream by the children.

type T

type T interface {
	// Consume consumes a message from the specified topic on behalf of the
	// specified consumer group. If there are no more new messages in the topic
	// at the time of the request then it will block for
	// `Config.Consumer.LongPollingTimeout`. If no new message is produced during
	// that time, then `ErrRequestTimeout` is returned.
	//
	// Note that during state transitions topic subscribe<->unsubscribe and
	// consumer group register<->deregister the method may return either
	// `ErrBufferOverflow` or `ErrRequestTimeout` even when there are messages
	// available for consumption. In that case the user should back off a bit
	// and then repeat the request.
	Consume(group, topic string) (Message, error)

	// AsyncConsume is an asynchronous counterpart of Consume function. It
	// sends a response down to a buffered channel of the consumer machinery
	// and returns a channel that a response should be expected from.
	AsyncConsume(group, topic string) <-chan Response

	// Stop sends a shutdown signal to all internal goroutines and blocks until
	// they are stopped. It is guaranteed that all last consumed offsets of all
	// consumer groups/topics are committed to Kafka before Consumer stops.
	Stop()
}