cursor

Documentation

Index

• type Checkpoint
  • func ToCheckpoint(msg pulsar.Message) (cp Checkpoint, err error)
  • func (cp *Checkpoint) After(cp2 Checkpoint) bool
  • func (cp *Checkpoint) Equal(cp2 Checkpoint) bool
  • func (cp *Checkpoint) FromKey(str string) error
  • func (cp *Checkpoint) ToKey() string
• type PulsarSubscriptionTracker
  • func NewPulsarSubscriptionTracker(client pulsar.Client, topic string, commitInterval time.Duration, ...) (*PulsarSubscriptionTracker, error)
  • func (p *PulsarSubscriptionTracker) Close()
  • func (p *PulsarSubscriptionTracker) Commit(_ Checkpoint, mid pulsar.MessageID) error
  • func (p *PulsarSubscriptionTracker) Last() (Checkpoint, error)
  • func (p *PulsarSubscriptionTracker) Start()
• type PulsarTracker
  • func NewPulsarTracker(client pulsar.Client, topic string) (*PulsarTracker, error)
  • func (p *PulsarTracker) Close()
  • func (p *PulsarTracker) Commit(_ Checkpoint, _ pulsar.MessageID) error
  • func (p *PulsarTracker) Last() (last Checkpoint, err error)
  • func (p *PulsarTracker) Start()
• type Tracker

Types

type Checkpoint

type Checkpoint struct {
	LSN        uint64
	Seq        uint32
	Data       []byte
	ServerTime time.Time
}

func ToCheckpoint

func ToCheckpoint(msg pulsar.Message) (cp Checkpoint, err error)

func (*Checkpoint) After

func (cp *Checkpoint) After(cp2 Checkpoint) bool

func (*Checkpoint) Equal

func (cp *Checkpoint) Equal(cp2 Checkpoint) bool

func (*Checkpoint) FromKey

func (cp *Checkpoint) FromKey(str string) error

func (*Checkpoint) ToKey

func (cp *Checkpoint) ToKey() string

type PulsarSubscriptionTracker

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

func NewPulsarSubscriptionTracker

func NewPulsarSubscriptionTracker(client pulsar.Client, topic string, commitInterval time.Duration, replicateState bool) (*PulsarSubscriptionTracker, error)

func (*PulsarSubscriptionTracker) Close

func (p *PulsarSubscriptionTracker) Close()

func (*PulsarSubscriptionTracker) Commit

func (p *PulsarSubscriptionTracker) Commit(_ Checkpoint, mid pulsar.MessageID) error

func (*PulsarSubscriptionTracker) Last

func (p *PulsarSubscriptionTracker) Last() (Checkpoint, error)

func (*PulsarSubscriptionTracker) Start

func (p *PulsarSubscriptionTracker) Start()

type PulsarTracker

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

func NewPulsarTracker

func NewPulsarTracker(client pulsar.Client, topic string) (*PulsarTracker, error)

func (*PulsarTracker) Close

func (p *PulsarTracker) Close()

func (*PulsarTracker) Commit

func (p *PulsarTracker) Commit(_ Checkpoint, _ pulsar.MessageID) error

func (*PulsarTracker) Last

func (p *PulsarTracker) Last() (last Checkpoint, err error)

func (*PulsarTracker) Start

func (p *PulsarTracker) Start()

type Tracker

type Tracker interface {
	Last() (cp Checkpoint, err error)
	Start()
	Commit(cp Checkpoint, mid pulsar.MessageID) error
	Close()
}