kafka

Documentation

Index

• type Config
• type Payload
• type Reader
  • func NewReader(brokers []string, topic string, groupId string) *Reader
  • func (k *Reader) CommitMessages(ctx context.Context, commits <-chan kafka.Message) error
  • func (k *Reader) FetchMessage(ctx context.Context, messages chan<- kafka.Message) error
• type TopicParams
• type Writer
  • func NewWriter(broker string, params []TopicParams, batchSize int) *Writer
  • func (w *Writer) WriteMessages(ctx context.Context, payload []byte, topic string) error

Types

type Config

type Config struct {
	Broker string        `json:"broker"`
	Params []TopicParams `json:"params"`
}

type Payload

type Payload struct {
	Status  string       `json:"status"`
	Block   common.Block `json:"block"`
	Version int          `json:"version"`
}

type Reader

type Reader struct {
	Reader *kafka.Reader
}

func NewReader

func NewReader(brokers []string, topic string, groupId string) *Reader

func (*Reader) CommitMessages

func (k *Reader) CommitMessages(ctx context.Context, commits <-chan kafka.Message) error

func (*Reader) FetchMessage

func (k *Reader) FetchMessage(ctx context.Context, messages chan<- kafka.Message) error

type TopicParams

type TopicParams struct {
	Topic     string   `json:"topic"`
	Addresses []string `json:"addresses"`
	Topics    []string `json:"topics"`
}

type Writer

type Writer struct {
	Writer *kafka.Writer
	Params *[]TopicParams
}

func NewWriter

func NewWriter(broker string, params []TopicParams, batchSize int) *Writer

func (*Writer) WriteMessages

func (w *Writer) WriteMessages(ctx context.Context, payload []byte, topic string) error