sarama

Documentation

Index

• func Finalize(c *Config) error
• type Client
  • func NewClient(servers []string, config *Config) (*Client, error)
• type Config
  • func NewConfig() *Config
• type ConsumerGroup
  • func NewConsumerGroup(logger logging.Logger, servers []string, groupID string, config *Config, ...) (*ConsumerGroup, error)
  • func (c *ConsumerGroup) Close() error
  • func (c *ConsumerGroup) Consume(ctx context.Context, topics []string) error
• type SamplerHandler
  • func NewSamplerHandler(logger logging.Logger, samplerOpts []sampler.Option) *SamplerHandler
  • func (h *SamplerHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (h *SamplerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *SamplerHandler) Setup(sess sarama.ConsumerGroupSession) error
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Functions

func Finalize

func Finalize(c *Config) error

Types

type Client

type Client struct {
	sarama.Client
	// contains filtered or unexported fields
}

func NewClient

func NewClient(servers []string, config *Config) (*Client, error)

type Config

type Config = sarama.Config

func NewConfig

func NewConfig() *Config

type ConsumerGroup

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

func NewConsumerGroup

func NewConsumerGroup(logger logging.Logger, servers []string, groupID string, config *Config, samplerOpts []sampler.Option) (*ConsumerGroup, error)

func (*ConsumerGroup) Close

func (c *ConsumerGroup) Close() error

func (*ConsumerGroup) Consume

func (c *ConsumerGroup) Consume(ctx context.Context, topics []string) error

type SamplerHandler

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

Handler represents a Sarama consumer group consumer

func NewSamplerHandler

func NewSamplerHandler(logger logging.Logger, samplerOpts []sampler.Option) *SamplerHandler

func (*SamplerHandler) Cleanup

func (h *SamplerHandler) Cleanup(sarama.ConsumerGroupSession) error

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

func (*SamplerHandler) ConsumeClaim

func (h *SamplerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

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

func (*SamplerHandler) Setup

func (h *SamplerHandler) Setup(sess sarama.ConsumerGroupSession) error

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

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)