ksgo

README

ksgo

Kafka Stream implemented in Golang

Documentation

Index

• func NewBadgerKVStore(opt badger.Options) *badgerKVStore
• type Iterator
• type KVStore
• type Processor
• type ProcessorContext
• type ReadOnlyKVStore
• type StateStore
• type Topology
  • func (t *Topology) AddGlobalStore(name string) *Topology
  • func (t *Topology) AddProcessor(name string) *Topology
  • func (t *Topology) AddSink(name string) *Topology
  • func (t *Topology) AddSource(name string) *Topology
  • func (t *Topology) AddStateStore(name string) *Topology

Functions

func NewBadgerKVStore

func NewBadgerKVStore(opt badger.Options) *badgerKVStore

Types

type Iterator

type Iterator interface {
	Err() error
	Key() []byte
	Next() bool
	Close()
	Seek(key []byte) bool
	Value() ([]byte, error)
}

type KVStore

type KVStore interface {
	ReadOnlyKVStore
	Set(string, []byte) error
	Delete(string) error
}

type Processor

type Processor func(ctx ProcessorContext, val interface{})

type ProcessorContext

type ProcessorContext interface {
	Topic() string
	Partition() int32
	Key() interface{}
	Offset() int64
	Forward(key string, val interface{})
	GetStateStore(name string) StateStore
}

type ReadOnlyKVStore

type ReadOnlyKVStore interface {
	StateStore
	Has(string) (bool, error)
	Get(string) ([]byte, error)
	GetOffset(defValue int64) (int64, error)
	Iterator() (Iterator, error)
	IteratorWithRange(start, limit []byte) (Iterator, error)
}

type StateStore

type StateStore interface {
	// Name() string
	// Open() error
	Close() error
}

type Topology

type Topology struct {
}

func (*Topology) AddGlobalStore

func (t *Topology) AddGlobalStore(name string) *Topology

func (*Topology) AddProcessor

func (t *Topology) AddProcessor(name string) *Topology

func (*Topology) AddSink

func (t *Topology) AddSink(name string) *Topology

func (*Topology) AddSource

func (t *Topology) AddSource(name string) *Topology

func (*Topology) AddStateStore

func (t *Topology) AddStateStore(name string) *Topology