kafclient

Jul 7, 2023

Changes in this version

+ func RandStringBytes(n int) string

type Client

+ func (k *Client) CreateTopic(topic string, numPart int)

+ func (k *Client) ListTopics2()

Jun 28, 2023

Changes in this version

type IClient

+ IsReaderConnected func() bool

+ IsWriters func() bool

+ Listen func(ctx context.Context, cMgs chan *Message) error

+ ListenWithAutoCommit func(ctx context.Context, cMgs chan *Message) error

+ NewConsumer func(consumerGroup string, topics []string)

+ NewPublisher func() error

+ SetAddrs func(addrs []string)

Jun 26, 2023

Changes in this version

type Client

+ func (k *Client) HealthCheckBroker()

+ func (k *Client) IsReaderConnected() bool

+ func (k *Client) IsWriters() bool

+ func (k *Client) Listen(ctx context.Context, cMgs chan *Message) error

+ func (k *Client) ListenWithAutoCommit(ctx context.Context, cMgs chan *Message) error

+ func (k *Client) NewConsumer(consumerGroup string, topics []string)

+ func (k *Client) NewPublisher() error

+ func (k *Client) SetAddrs(addrs []string)

type Message

+ Key string

Jun 26, 2023

Changes in this version

+ var NUM_PARTITION = 3

+ var REPLICATION_FACTOR = 1

+ func BodyParse(bin []byte, p interface{}) error

+ func ToInt32(in *int32) int32

+ func ToPInt32(in int32) *int32

+ type Client struct

+ func (p *Client) AsyncPublish(topic string, messages ...interface{}) error

+ func (p *Client) Publish(topic string, messages ...interface{}) error

+ func (p *Client) PublishWithConfig(topic *Topic, config *SenderConfig, messages ...interface{}) error

+ func (ps *Client) Close() error

+ func (ps *Client) InitConsumer(brokerURLs ...string) error

+ func (ps *Client) InitConsumerGroup(consumerGroup string, brokerURLs ...string) error

+ func (ps *Client) InitPublisher(brokerURLs ...string)

+ func (ps *Client) ListTopics(brokers ...string) ([]string, error)

+ func (ps *Client) OnAsyncSubscribe(topics []*Topic, numberPuller int, buf chan Message) error

+ func (ps *Client) OnScanMessages(topics []string, bufMessage chan Message) error

+ type ConsumerGroupHandle struct

+ func (consumer *ConsumerGroupHandle) Cleanup(ss sarama.ConsumerGroupSession) error

+ func (consumer *ConsumerGroupHandle) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

+ func (consumer *ConsumerGroupHandle) Setup(ss sarama.ConsumerGroupSession) error

+ type IClient interface

+ Close func() error

+ InitConsumer func(brokerURLs ...string) error

+ InitConsumerGroup func(consumerGroup string, brokerURLs ...string) error

+ InitPublisher func(brokerURLs ...string)

+ ListTopics func(brokers ...string) ([]string, error)

+ OnAsyncSubscribe func(topics []*Topic, numberPuller int, buf chan Message) error

+ OnScanMessages func(topics []string, bufMessage chan Message) error

+ Publish func(topic string, messages ...interface{}) error

+ PublishWithConfig func(topic *Topic, config *SenderConfig, messages ...interface{}) error

+ type Message struct

+ Body []byte

+ Commit func()

+ ConsumerGroup string

+ Headers map[string]string

+ Offset int64

+ Partition int

+ Timestamp int64

+ Topic string

+ type SenderConfig struct

+ Headers map[string]string

+ Metadata interface{}

+ type Topic struct

+ AutoCommit bool

+ IsNeedManualCreateTopic bool

+ Name string

+ Partition *int32