topic

Documentation

Index

• Variables
• type Partition
  • func (t *Partition) BatchPutMessages(msgs []*sgproto.Message) error
  • func (p *Partition) Close() error
  • func (p *Partition) ForRange(min, max sandflake.ID, fn func(msg *sgproto.Message) error) error
  • func (s *Partition) GetMessage(offset sandflake.ID, k, suffix []byte) (*sgproto.Message, error)
  • func (t *Partition) HasKey(key, clusterKey []byte) (bool, error)
  • func (t *Partition) InitStore(basePath string) error
  • func (p *Partition) Iter() storage.MessageIterator
  • func (p *Partition) LastMessage() (*sgproto.Message, error)
  • func (p *Partition) LastWALEntry() []byte
  • func (t *Partition) NextID() sandflake.ID
  • func (t *Partition) PutMessage(msg *sgproto.Message) error
  • func (p *Partition) RangeFromWAL(min []byte, fn func(*sgproto.Message) error) error
  • func (t *Partition) String() string
• type Topic
  • func (t *Topic) BatchPutMessages(msgs []*sgproto.Message) error
  • func (t *Topic) ChoosePartition(msg *sgproto.Message) *Partition
  • func (t *Topic) ChoosePartitionForKey(key []byte) *Partition
  • func (t *Topic) ChooseRandomPartition() *Partition
  • func (t *Topic) Close() error
  • func (t *Topic) ForEach(fn func(msg *sgproto.Message) error) error
  • func (t *Topic) ForRange(min, max sandflake.ID, fn func(msg *sgproto.Message) error) error
  • func (t *Topic) GetPartition(id string) *Partition
  • func (t *Topic) InitStore(basePath string) error
  • func (t *Topic) ListPartitions() []*Partition
  • func (t *Topic) PutMessage(partition string, msg *sgproto.Message) error
  • func (t *Topic) Validate() error

Variables

var (
	ErrNoKeySet = errors.New("ErrNoKeySet")
)

Types

type Partition

type Partition struct {
	Id       string
	Replicas []string
	// contains filtered or unexported fields
}

func (*Partition) BatchPutMessages

func (t *Partition) BatchPutMessages(msgs []*sgproto.Message) error

func (*Partition) Close

func (p *Partition) Close() error

func (*Partition) ForRange

func (p *Partition) ForRange(min, max sandflake.ID, fn func(msg *sgproto.Message) error) error

func (*Partition) GetMessage

func (s *Partition) GetMessage(offset sandflake.ID, k, suffix []byte) (*sgproto.Message, error)

func (*Partition) HasKey

func (t *Partition) HasKey(key, clusterKey []byte) (bool, error)

func (*Partition) InitStore

func (t *Partition) InitStore(basePath string) error

func (*Partition) Iter

func (p *Partition) Iter() storage.MessageIterator

func (*Partition) LastMessage

func (p *Partition) LastMessage() (*sgproto.Message, error)

func (*Partition) LastWALEntry

func (p *Partition) LastWALEntry() []byte

func (*Partition) NextID

func (t *Partition) NextID() sandflake.ID

func (*Partition) PutMessage

func (t *Partition) PutMessage(msg *sgproto.Message) error

func (*Partition) RangeFromWAL

func (p *Partition) RangeFromWAL(min []byte, fn func(*sgproto.Message) error) error

func (*Partition) String

func (t *Partition) String() string

type Topic

type Topic struct {
	Name              string
	Kind              sgproto.TopicKind
	ReplicationFactor int
	NumPartitions     int
	Partitions        []*Partition
	StorageDriver     sgproto.StorageDriver
	// contains filtered or unexported fields
}

func (*Topic) BatchPutMessages

func (t *Topic) BatchPutMessages(msgs []*sgproto.Message) error

func (*Topic) ChoosePartition

func (t *Topic) ChoosePartition(msg *sgproto.Message) *Partition

func (*Topic) ChoosePartitionForKey

func (t *Topic) ChoosePartitionForKey(key []byte) *Partition

func (*Topic) ChooseRandomPartition

func (t *Topic) ChooseRandomPartition() *Partition

func (*Topic) Close

func (t *Topic) Close() error

func (*Topic) ForEach

func (t *Topic) ForEach(fn func(msg *sgproto.Message) error) error

func (*Topic) ForRange

func (t *Topic) ForRange(min, max sandflake.ID, fn func(msg *sgproto.Message) error) error

func (*Topic) GetPartition

func (t *Topic) GetPartition(id string) *Partition

func (*Topic) InitStore

func (t *Topic) InitStore(basePath string) error

func (*Topic) ListPartitions

func (t *Topic) ListPartitions() []*Partition

func (*Topic) PutMessage

func (t *Topic) PutMessage(partition string, msg *sgproto.Message) error

func (*Topic) Validate

func (t *Topic) Validate() error