kafka

README

Working with string encoder

Documentation

Index

• Variables
• func BrokerConfig(c *sarama.Config) broker.BrokerOption
• func ClusterConfig(c *sarama.Config) broker.BrokerOption
• func CreateKafkaBroker(cfg *KafkaBrokerConfig, opts ...broker.BrokerOption) (broker.Broker, error)
• func NewKafkaBroker(opts ...broker.BrokerOption) broker.Broker
• func SubscribeConfig(c *sarama.Config) broker.SubscribeOption
• func SubscribeContext(ctx context.Context) broker.SubscribeOption
• type KafkaBrokerConfig
• 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)

Variables

var (
	DefaultKafkaBroker = "127.0.0.1:9092"
	DefaultLogger      = logrus.NewLogrusLogger()
)

var (
	DefaultBrokerConfig  = sarama.NewConfig()
	DefaultClusterConfig = sarama.NewConfig()
)

var (
	SHA256 scram.HashGeneratorFcn = sha256.New
	SHA512 scram.HashGeneratorFcn = sha512.New
)

Functions

func BrokerConfig

func BrokerConfig(c *sarama.Config) broker.BrokerOption

func ClusterConfig

func ClusterConfig(c *sarama.Config) broker.BrokerOption

func CreateKafkaBroker

func CreateKafkaBroker(cfg *KafkaBrokerConfig, opts ...broker.BrokerOption) (broker.Broker, error)

func NewKafkaBroker

func NewKafkaBroker(opts ...broker.BrokerOption) broker.Broker

func SubscribeConfig

func SubscribeConfig(c *sarama.Config) broker.SubscribeOption

func SubscribeContext

func SubscribeContext(ctx context.Context) broker.SubscribeOption

SubscribeContext set the context for broker.SubscribeOption

Types

type KafkaBrokerConfig

type KafkaBrokerConfig struct {
	Addresses     []string
	ConsumerGroup string

	SASLEnabled   bool
	SASLUser      string
	SASLPassword  string
	SASLHandshake bool
	SASLAlgorithm string

	TLSEnabled        bool
	TLSSkipVerify     bool
	TLSClientCertFile string
	TLSClientKeyFile  string
	TLSCaCertFile     string
}

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)