kafka

Documentation

Index

• type Client
• type Config
  • func NewConfig() *Config
  • func (c *Config) Finalize() error
• type ConsumerGroup
• type ConsumerManager
  • func NewConsumerManager(ctx context.Context, logger logging.Logger, config *Config, ...) (*ConsumerManager, error)
  • func (m *ConsumerManager) Reconcile() error
  • func (m *ConsumerManager) Topics() []string
• type TopicsConfig

Types

type Client

type Client interface {
	// Topics returns the set of available topics as retrieved from cluster metadata.
	Topics() ([]string, error)

	// RefreshMetadata takes a list of topics and queries the cluster to refresh the
	// available metadata for those topics. If no topics are provided, it will refresh
	// metadata for all topics.
	RefreshMetadata(topics ...string) error

	Close() error
}

type Config

type Config struct {
	Servers       []string
	ConsumerGroup string
	Sarama        sarama.Config
	Topics        TopicsConfig
}

func NewConfig

func NewConfig() *Config

func (*Config) Finalize

func (c *Config) Finalize() error

type ConsumerGroup

type ConsumerGroup interface {
	Consume(ctx context.Context, topics []string) error
	Close() error
}

type ConsumerManager

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

func NewConsumerManager

func NewConsumerManager(ctx context.Context, logger logging.Logger, config *Config, neblicConfig *neblic.Config) (*ConsumerManager, error)

func (*ConsumerManager) Reconcile

func (m *ConsumerManager) Reconcile() error

func (*ConsumerManager) Topics

func (m *ConsumerManager) Topics() []string

type TopicsConfig

type TopicsConfig struct {
	Max           int
	RefreshPeriod time.Duration
	Filter        filter.Config
}