kf

Documentation

Index

• Variables
• type Client
  • func (k *Client) Connect(ctx context.Context, conf *conf.GlobalConfiguration, topic string, ...) error
  • func (k *Client) Consumer(ctx context.Context, reader *kafka.Reader, action func(kafka.Message))
  • func (k *Client) NewReader(groupID, topic string, partition int) (*kafka.Reader, error)
  • func (k *Client) TopicList() (map[string]struct{}, error)

Variables

var (
	// Database var
	Database kf = &Client{}
)

Types

type Client

type Client struct {
	Config   *kafka.Conn
	Messages chan *sarama.ConsumerMessage
	Errors   chan *sarama.ConsumerError

	Count int
	// contains filtered or unexported fields
}

Client struct

func (*Client) Connect

func (k *Client) Connect(ctx context.Context, conf *conf.GlobalConfiguration, topic string, partition int) error

Connect is basic connection to kafka

func (*Client) Consumer

func (k *Client) Consumer(ctx context.Context, reader *kafka.Reader, action func(kafka.Message))

Consumer for a topic

func (*Client) NewReader

func (k *Client) NewReader(groupID, topic string, partition int) (*kafka.Reader, error)

NewReader func

func (*Client) TopicList

func (k *Client) TopicList() (map[string]struct{}, error)

TopicList get the list of topics for a broker