kafka

Jul 11, 2019

Changes in this version

+ const BrokersKey

+ const ConnectionTimeoutKey

+ const GroupKey

+ const InitialOffsetKey

+ const OffsetNewest

+ const OffsetOldest

+ const VersionKey

+ var DefaultConnectionTimeout = 5 * time.Second

+ var ErrNotOpened = errors.New("error Kafka dialect not opened")

+ func ValidateConnectionKeyVal(values ConnectionMap) error

+ type Config struct

+ Brokers []string

+ ConnectionTimeout time.Duration

+ Group string

+ InitialOffset int64

+ Version sarama.KafkaVersion

+ func NewConfig(values ConnectionMap) (Config, error)

+ type ConnectionMap map[string]string

+ func ParseConnectionstring(connectionstring string) ConnectionMap

+ type Dialect struct

+ Config *sarama.Config

+ Connection Config

+ Topics []types.Topic

+ func NewDialect(connectionstring string) (*Dialect, error)

+ func (dialect *Dialect) Assigned(topic types.Topic)

+ func (dialect *Dialect) Close() error

+ func (dialect *Dialect) Consumer() types.Consumer

+ func (dialect *Dialect) Healthy() bool

+ func (dialect *Dialect) Open() error

+ func (dialect *Dialect) Producer() types.Producer