client

Documentation

Index

• Constants
• func Consume(cfg *Config, tranceConfig TranceConsumerConfigFunc, doCommit bool, ...) (<-chan string, error)
• func DefaultConfig() *sarama.Config
• func ParseBalanceStrategy(strategy BalanceStrategy) sarama.BalanceStrategy
• func ParseKafkaVersion(version string) sarama.KafkaVersion
• func ParsePartitionStrategy(strategy PartitionStrategy) sarama.PartitionerConstructor
• func Produce(cfg *Config, tranceConfig TranceProducerConfigFunc, hook EventHook) (chan<- string, error)
• type BalanceStrategy
• type Config
  • func (c *Config) CheckParams(defaults map[string]string, requires []string) error
  • func (c *Config) Get(key, dftValue string) string
  • func (c *Config) SetKey(key, value string)
• type ConfigSlice
  • func (s ConfigSlice) FindConfig(name MessageType) *Config
• type EventHook
• type EventType
• type Message
• type MessageType
• type MissingConfigError
  • func (m MissingConfigError) Error() string
• type PartitionStrategy
• type TranceConsumerConfigFunc
• type TranceProducerConfigFunc

Constants

const (
	EventTypeInitialError EventType = iota
	EventTypeTranceMessageError

	EventTypeProduceMsgSuccess
	EventTypeProduceMsgError

	EventTypeConsumeMsgSuccess
	EventTypeConsumeMsgError
	EventTypeConsumerGroupError

	BalanceStrategySticky     BalanceStrategy = "sticky"
	BalanceStrategyRoundRobin BalanceStrategy = "round_robin"
	BalanceStrategyRange      BalanceStrategy = "range"

	PartitionStrategyRoundRobin PartitionStrategy = "round_robin"
	PartitionStrategyHash       PartitionStrategy = "hash"
	PartitionStrategyManual     PartitionStrategy = "manual"
	PartitionStrategyRandom     PartitionStrategy = "random"
)

Functions

func Consume

func Consume(cfg *Config, tranceConfig TranceConsumerConfigFunc, doCommit bool, hook EventHook) (<-chan string, error)

func DefaultConfig

func DefaultConfig() *sarama.Config

func ParseBalanceStrategy

func ParseBalanceStrategy(strategy BalanceStrategy) sarama.BalanceStrategy

func ParseKafkaVersion

func ParseKafkaVersion(version string) sarama.KafkaVersion

func ParsePartitionStrategy

func ParsePartitionStrategy(strategy PartitionStrategy) sarama.PartitionerConstructor

func Produce

func Produce(cfg *Config, tranceConfig TranceProducerConfigFunc, hook EventHook) (chan<- string, error)

Types

type BalanceStrategy

type BalanceStrategy string

type Config

type Config struct {
	Name   MessageType       `json:"name"`
	Params map[string]string `json:"params"`
}

func (*Config) CheckParams

func (c *Config) CheckParams(defaults map[string]string, requires []string) error

func (*Config) Get

func (c *Config) Get(key, dftValue string) string

func (*Config) SetKey

func (c *Config) SetKey(key, value string)

type ConfigSlice

type ConfigSlice []*Config

func (ConfigSlice) FindConfig

func (s ConfigSlice) FindConfig(name MessageType) *Config

type EventHook

type EventHook func(EventType, interface{}, error)

type EventType

type EventType int

type Message

type Message map[string]interface{}

type MessageType

type MessageType string

type MissingConfigError

type MissingConfigError struct {
	Name string
}

func (MissingConfigError) Error

func (m MissingConfigError) Error() string

type PartitionStrategy

type PartitionStrategy string

type TranceConsumerConfigFunc

type TranceConsumerConfigFunc func(configs *Config) (string, string, []string, *sarama.Config, error)

return groupId,topic, servers, kafkaConfig, error

type TranceProducerConfigFunc

type TranceProducerConfigFunc func(configs *Config) (string, []string, *sarama.Config, error)

return topic, servers, kafkaConfig, error