adapter

Documentation

Index

• type Header
  • func NewHeader(h *kafka.Header) *Header
  • func (h *Header) Key() string
  • func (h *Header) Value() []byte
• type Message
  • func NewMessage(msg *kafka.Message) *Message
  • func (m *Message) Headers() []kitkafka.Header
  • func (m *Message) Key() []byte
  • func (m *Message) Offset() int64
  • func (m *Message) Partition() int32
  • func (m *Message) Timestamp() time.Time
  • func (m *Message) Topic() string
  • func (m *Message) Value() []byte
• type Reader
  • func NewReader(brokers []string, groupID string, topics []string, opts ...ReaderOption) (*Reader, error)
  • func (r *Reader) Close() error
  • func (r *Reader) ReadMessage(_ context.Context) (kitkafka.Message, error)
• type ReaderOption
  • func ReaderReadTimeout(timeout time.Duration) ReaderOption

Types

type Header

type Header struct {
	// contains filtered or unexported fields
}

func NewHeader

func NewHeader(h *kafka.Header) *Header

func (*Header) Key

func (h *Header) Key() string

func (*Header) Value

func (h *Header) Value() []byte

type Message

type Message struct {
	// contains filtered or unexported fields
}

func NewMessage

func NewMessage(msg *kafka.Message) *Message

func (*Message) Headers

func (m *Message) Headers() []kitkafka.Header

func (*Message) Key

func (m *Message) Key() []byte

func (*Message) Offset

func (m *Message) Offset() int64

func (*Message) Partition

func (m *Message) Partition() int32

func (*Message) Timestamp

func (m *Message) Timestamp() time.Time

func (*Message) Topic

func (m *Message) Topic() string

func (*Message) Value

func (m *Message) Value() []byte

type Reader

type Reader struct {
	// contains filtered or unexported fields
}

func NewReader

func NewReader(brokers []string, groupID string, topics []string, opts ...ReaderOption) (*Reader, error)

func (*Reader) Close

func (r *Reader) Close() error

func (*Reader) ReadMessage

func (r *Reader) ReadMessage(_ context.Context) (kitkafka.Message, error)

type ReaderOption

type ReaderOption func(*Reader)

func ReaderReadTimeout

func ReaderReadTimeout(timeout time.Duration) ReaderOption