kafka

Documentation

Index

• type Config
• type Consumer
  • func (consumer *Consumer) Cleanup(group sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error
• type HandlerFunc
• type IServer
  • func Server(sig chan os.Signal, cfg Config, opts ...Option) (IServer, error)
• type Option
  • func RegisterRoute(topic string, handler HandlerFunc) Option
  • func RegisterRoutes(handlers map[string]HandlerFunc) Option

Types

type Config

type Config struct {
	// Brokers required, kafka brokers separated by comma.
	Brokers string

	// Group required, kafka consumer group.
	Group string

	// SaramaConfig required, kafka sarama config.
	SaramaConfig *sarama.Config

	// Name optional, kafka server name to be exposed.
	Name string
}

Config http config server.

type Consumer

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

Consumer represents a Sarama consumer group consumer

func (*Consumer) Cleanup

func (consumer *Consumer) Cleanup(group sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Consumer) ConsumeClaim

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*Consumer) Setup

func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type HandlerFunc

type HandlerFunc func(ctx context.Context, message *sarama.ConsumerMessage) error

type IServer

type IServer interface {
	Run() error
	Shutdown() error
}

func Server

func Server(sig chan os.Signal, cfg Config, opts ...Option) (IServer, error)

Server functions to initialize http server.

type Option

type Option interface {
	Apply(o *options)
}

func RegisterRoute

func RegisterRoute(topic string, handler HandlerFunc) Option

RegisterRoute function to add route to server.

func RegisterRoutes

func RegisterRoutes(handlers map[string]HandlerFunc) Option

RegisterRoutes function to add multi route to server.