transport

Documentation

Index

• func EncodeJSONRequest(_ context.Context, msg *kafka.Message, request interface{}) error
• type Consumer
  • func NewConsumer(e endpoint.Endpoint, dec DecodeRequestFunc, opts ...ConsumerOption) *Consumer
  • func (c Consumer) Handle(ctx context.Context, msg kafka.Message) (err error)
• type ConsumerFinalizerFunc
• type ConsumerOption
  • func ConsumerAfter(after ...ConsumerResponseFunc) ConsumerOption
  • func ConsumerBefore(before ...RequestFunc) ConsumerOption
  • func ConsumerErrorHandler(errorHandler transport.ErrorHandler) ConsumerOption
  • func ConsumerErrorLogger(logger log.Logger) ConsumerOption
  • func ConsumerFinalizer(f ...ConsumerFinalizerFunc) ConsumerOption
• type ConsumerResponseFunc
• type DecodeRequestFunc
• type EncodeRequestFunc
• type EncodeResponseFunc
• type Handlers
• type Producer
  • func NewProducer(handler kafka.Handler, topic string, enc EncodeRequestFunc, ...) *Producer
  • func (p Producer) Endpoint() endpoint.Endpoint
• type ProducerFinalizerFunc
• type ProducerOption
  • func ProducerAfter(after ...ProducerResponseFunc) ProducerOption
  • func ProducerBefore(before ...RequestFunc) ProducerOption
  • func ProducerFinalizer(f ...ProducerFinalizerFunc) ProducerOption
  • func ProducerResponse(response interface{}) ProducerOption
• type ProducerResponseFunc
• type RequestFunc
• type Router
  • func NewRouter(opts ...RouterOption) *Router
  • func (r *Router) AddHandler(topic string, handler kafka.Handler) *Router
  • func (r Router) Handle(ctx context.Context, msg kafka.Message) error
  • func (r Router) Handlers() Handlers
• type RouterOption
  • func RouterWithHandler(topic string, handler kafka.Handler) RouterOption
  • func RouterWithHandlers(handlers Handlers) RouterOption

Functions

func EncodeJSONRequest

func EncodeJSONRequest(_ context.Context, msg *kafka.Message, request interface{}) error

Types

type Consumer

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

func NewConsumer

func NewConsumer(
	e endpoint.Endpoint,
	dec DecodeRequestFunc,
	opts ...ConsumerOption,
) *Consumer

func (Consumer) Handle

func (c Consumer) Handle(ctx context.Context, msg kafka.Message) (err error)

type ConsumerFinalizerFunc

type ConsumerFinalizerFunc func(ctx context.Context, msg *kafka.Message, err error)

type ConsumerOption

type ConsumerOption func(consumer *Consumer)

func ConsumerAfter

func ConsumerAfter(after ...ConsumerResponseFunc) ConsumerOption

func ConsumerBefore

func ConsumerBefore(before ...RequestFunc) ConsumerOption

func ConsumerErrorHandler

func ConsumerErrorHandler(errorHandler transport.ErrorHandler) ConsumerOption

func ConsumerErrorLogger

func ConsumerErrorLogger(logger log.Logger) ConsumerOption

func ConsumerFinalizer

func ConsumerFinalizer(f ...ConsumerFinalizerFunc) ConsumerOption

type ConsumerResponseFunc

type ConsumerResponseFunc func(ctx context.Context, response interface{}) context.Context

type DecodeRequestFunc

type DecodeRequestFunc func(ctx context.Context, msg kafka.Message) (request interface{}, err error)

type EncodeRequestFunc

type EncodeRequestFunc func(context.Context, *kafka.Message, interface{}) error

type EncodeResponseFunc

type EncodeResponseFunc func(context.Context, interface{}) error

type Handlers

type Handlers map[string][]kafka.Handler

Handlers represents Topic -> []Handler mapping

type Producer

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

func NewProducer

func NewProducer(
	handler kafka.Handler,
	topic string,
	enc EncodeRequestFunc,
	options ...ProducerOption,
) *Producer

func (Producer) Endpoint

func (p Producer) Endpoint() endpoint.Endpoint

type ProducerFinalizerFunc

type ProducerFinalizerFunc func(ctx context.Context, err error)

type ProducerOption

type ProducerOption func(consumer *Producer)

func ProducerAfter

func ProducerAfter(after ...ProducerResponseFunc) ProducerOption

func ProducerBefore

func ProducerBefore(before ...RequestFunc) ProducerOption

func ProducerFinalizer

func ProducerFinalizer(f ...ProducerFinalizerFunc) ProducerOption

func ProducerResponse

func ProducerResponse(response interface{}) ProducerOption

type ProducerResponseFunc

type ProducerResponseFunc func(ctx context.Context) context.Context

type RequestFunc

type RequestFunc func(ctx context.Context, msg *kafka.Message) context.Context

type Router

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

func NewRouter

func NewRouter(opts ...RouterOption) *Router

func (*Router) AddHandler

func (r *Router) AddHandler(topic string, handler kafka.Handler) *Router

func (Router) Handle

func (r Router) Handle(ctx context.Context, msg kafka.Message) error

func (Router) Handlers

func (r Router) Handlers() Handlers

type RouterOption

type RouterOption func(*Router)

func RouterWithHandler

func RouterWithHandler(topic string, handler kafka.Handler) RouterOption

func RouterWithHandlers

func RouterWithHandlers(handlers Handlers) RouterOption