source

Documentation

Index

• type CommitMode
• type Opt
  • func WithBufferSize(size int) Opt
• type Opts
  • func DefaultOpts() *Opts
  • func (o *Opts) Configure(opts ...Opt)
• type Source
  • func WithContext[K, V any](ctx context.Context, r *kgo.Reader, key codec.Decoder[K], ...) *Source[K, V]
  • func (s *Source[K, V]) Commit(msgs ...msg.Message[K, V]) error
  • func (s *Source[K, V]) Error() error
  • func (s *Source[K, V]) Messages() chan msg.Message[K, V]

Types

type CommitMode

type CommitMode int

CommitMode is a Kafka source commit mode.

const (
	// CommitAuto commits messages automatically.
	CommitAuto CommitMode = iota

	// CommitManual commits messages manually.
	CommitManual
)

type Opt

type Opt func(o *Opts)

Opt is a Kafka source option.

func WithBufferSize

func WithBufferSize(size int) Opt

WithBufferSize configures the buffer size for a kafka source.

type Opts

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

Opts is a set of options for a kafka source.

func DefaultOpts

func DefaultOpts() *Opts

DefaultOpts is the default options for a kafka source.

func (*Opts) Configure

func (o *Opts) Configure(opts ...Opt)

Configure is a function that configures a kafka source.

type Source

type Source[K, V any] struct {
	// contains filtered or unexported fields
}

Source is a Kafka source.

func WithContext

func WithContext[K, V any](ctx context.Context, r *kgo.Reader, key codec.Decoder[K], value codec.Decoder[V], keyEncoder codec.Encoder[K], opts ...Opt) *Source[K, V]

WithContext is a constructor for a kafka source with a cancellation context.

func (*Source[K, V]) Commit

func (s *Source[K, V]) Commit(msgs ...msg.Message[K, V]) error

Commit is a Kafka source commit.

func (*Source[K, V]) Error

func (s *Source[K, V]) Error() error

Error is a Kafka source error.

func (*Source[K, V]) Messages

func (s *Source[K, V]) Messages() chan msg.Message[K, V]

Message is a Kafka source message.