api

Documentation

Index

• Variables
• type ConsumerTracer
• type ProduceTracer
• type Reader
  • func NewReader(ctx context.Context, log log.Log, r *kafka.Reader, bufferSize int, ...) *Reader
  • func (k *Reader) Close(ctx context.Context) error
  • func (k *Reader) Commit(ctx context.Context) error
  • func (k *Reader) StoreMessage(ctx context.Context, msg *kafka.Message) error
• type Writer
  • func NewWriter(ctx context.Context, w *kafka.Writer, bufferLen int, log log.Log, ...) *Writer
  • func (w *Writer) Close(ctx context.Context) error
  • func (w *Writer) Flush(ctx context.Context) error
  • func (w *Writer) Send(ctx context.Context, msg *kafka.Message) error

Variables

var ErrReaderBufferFull = fmt.Errorf("Reader.StoreMessage: Buffer full")

var ErrWriterBufferFull = fmt.Errorf("Reader.Send: Buffer full")

Types

type ConsumerTracer

type ConsumerTracer interface {
	span.SpanOp
}

type ProduceTracer

type ProduceTracer interface {
	KafkaInject(ctx context.Context, msg *kafka.Message)
	span.SpanOp
}

type Reader

type Reader struct {
	*kafka.Reader
	// contains filtered or unexported fields
}

func NewReader

func NewReader(ctx context.Context, log log.Log, r *kafka.Reader, bufferSize int, tr ConsumerTracer) *Reader

func (*Reader) Close

func (k *Reader) Close(ctx context.Context) error

func (*Reader) Commit

func (k *Reader) Commit(ctx context.Context) error

func (*Reader) StoreMessage

func (k *Reader) StoreMessage(ctx context.Context, msg *kafka.Message) error

type Writer

type Writer struct {
	*kafka.Writer
	// contains filtered or unexported fields
}

func NewWriter

func NewWriter(ctx context.Context, w *kafka.Writer, bufferLen int, log log.Log, tr ProduceTracer) *Writer

func (*Writer) Close

func (w *Writer) Close(ctx context.Context) error

func (*Writer) Flush

func (w *Writer) Flush(ctx context.Context) error

func (*Writer) Send

func (w *Writer) Send(ctx context.Context, msg *kafka.Message) error