kafkatransport

Documentation

Index

• type Consumer
  • func NewConsumer(channel string, endpoint endpoint.Endpoint, client sarama.ConsumerGroup, ...) *Consumer
  • func (c *Consumer) Consume(ctx context.Context)
• type ConsumerHook
• type ConsumerOption
  • func ConsumerBefore(before ...ConsumerRequestFunc) ConsumerOption
  • func WithCleanupHook(hook ConsumerHook) ConsumerOption
  • func WithErrorHandler(errorHandler transport.ErrorHandler) ConsumerOption
  • func WithSetupHook(hook ConsumerHook) ConsumerOption
• type ConsumerRequestFunc
• type DecodeMessageFunc
• type EncodeMessageFunc
• type Producer
  • func NewProducer(topic string, client sarama.Client, enc EncodeMessageFunc, ...) *Producer
  • func (p Producer) Endpoint() endpoint.Endpoint
• type ProducerOption
  • func ProducerAfter(after ...ProducerResponseFunc) ProducerOption
  • func ProducerBefore(before ...ProducerRequestFunc) ProducerOption
• type ProducerRequestFunc
• type ProducerResponseFunc

Types

type Consumer

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

func NewConsumer

func NewConsumer(
	channel string,
	endpoint endpoint.Endpoint,
	client sarama.ConsumerGroup,
	dec DecodeMessageFunc,
	options ...ConsumerOption) *Consumer

func (*Consumer) Consume

func (c *Consumer) Consume(ctx context.Context)

type ConsumerHook

type ConsumerHook func(sarama.ConsumerGroupSession) error

type ConsumerOption

type ConsumerOption func(*Consumer)

func ConsumerBefore

func ConsumerBefore(before ...ConsumerRequestFunc) ConsumerOption

func WithCleanupHook

func WithCleanupHook(hook ConsumerHook) ConsumerOption

func WithErrorHandler

func WithErrorHandler(errorHandler transport.ErrorHandler) ConsumerOption

func WithSetupHook

func WithSetupHook(hook ConsumerHook) ConsumerOption

type ConsumerRequestFunc

type ConsumerRequestFunc func(context.Context) context.Context

type DecodeMessageFunc

type DecodeMessageFunc func(ctx context.Context, msg *sarama.ConsumerMessage) (request interface{}, err error)

type EncodeMessageFunc

type EncodeMessageFunc func(ctx context.Context, request interface{}) (msg *sarama.ProducerMessage, err error)

type Producer

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

func NewProducer

func NewProducer(
	topic string,
	client sarama.Client,
	enc EncodeMessageFunc,
	options ...ProducerOption,
) *Producer

func (Producer) Endpoint

func (p Producer) Endpoint() endpoint.Endpoint

type ProducerOption

type ProducerOption func(*Producer)

func ProducerAfter

func ProducerAfter(after ...ProducerResponseFunc) ProducerOption

func ProducerBefore

func ProducerBefore(before ...ProducerRequestFunc) ProducerOption

type ProducerRequestFunc

type ProducerRequestFunc func(context.Context) context.Context

type ProducerResponseFunc

type ProducerResponseFunc func(partition int32, offset int64, msg *sarama.ProducerMessage)