processor

Documentation

Index

• type EntityOption
  • func WithSeparateOTBuckets(separate bool) EntityOption
• type ProcessorEntitier
• type ProcessorEntity
  • func NewProcessorEntity(name string, publishKeyspace string, inputOpts ...EntityOption) *ProcessorEntity
  • func (p *ProcessorEntity) BuildOTWatcherKey(watermark Watermark) string
  • func (p *ProcessorEntity) GetBucketName() string
  • func (p *ProcessorEntity) GetID() string
  • func (p *ProcessorEntity) GetPublishKeyspace() string
  • func (p *ProcessorEntity) IsSharedBucket() bool
  • func (p *ProcessorEntity) ParseOTWatcherKey(key string) (epoch int64, skip bool, err error)
• type Watermark
  • func (w Watermark) String() string

Examples

• Watermark.String

Types

type EntityOption

type EntityOption func(*entityOptions)

EntityOption set options for FromVertex.

func WithSeparateOTBuckets

func WithSeparateOTBuckets(separate bool) EntityOption

WithSeparateOTBuckets creates a different bucket for maintaining each processor offset-timeline.

type ProcessorEntitier

type ProcessorEntitier interface {
	GetID() string
	GetBucketName() string
	BuildOTWatcherKey(Watermark) string
	ParseOTWatcherKey(string) (int64, bool, error)
	IsSharedBucket() bool
	GetPublishKeyspace() string
}

ProcessorEntitier defines what can be a processor. The Processor is the smallest unit where the watermark will monotonically increase.

type ProcessorEntity

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

ProcessorEntity implements ProcessorEntitier.

func NewProcessorEntity

func NewProcessorEntity(name string, publishKeyspace string, inputOpts ...EntityOption) *ProcessorEntity

NewProcessorEntity returns a new `ProcessorEntity`

func (*ProcessorEntity) BuildOTWatcherKey

func (p *ProcessorEntity) BuildOTWatcherKey(watermark Watermark) string

BuildOTWatcherKey builds the offset-timeline key name

func (*ProcessorEntity) GetBucketName

func (p *ProcessorEntity) GetBucketName() string

GetBucketName returns the offset-timeline for the entity.

func (*ProcessorEntity) GetID

func (p *ProcessorEntity) GetID() string

GetID returns the ID of the processor.

func (*ProcessorEntity) GetPublishKeyspace

func (p *ProcessorEntity) GetPublishKeyspace() string

GetPublishKeyspace returns the publishKeyspace of the entity

func (*ProcessorEntity) IsSharedBucket

func (p *ProcessorEntity) IsSharedBucket() bool

IsSharedBucket returns true if the bucket is shared.

func (*ProcessorEntity) ParseOTWatcherKey

func (p *ProcessorEntity) ParseOTWatcherKey(key string) (epoch int64, skip bool, err error)

ParseOTWatcherKey parses the key of the KeyValue OT watcher and returns the epoch, a boolean to indicate whether the record can be skipped and error if any. NOTE: _defaultKeySeparator has constraints, please make sure we will not end up with multiple values

type Watermark

type Watermark time.Time

Watermark is the monotonically increasing watermark. It is tightly coupled with ProcessorEntity as the processor is responsible for monotonically increasing Watermark for that processor. NOTE: today we support only second progression of watermark, we need to support millisecond too.

func (Watermark) String

func (w Watermark) String() string

Example

location, _ := time.LoadLocation("UTC")
wm := Watermark(time.Unix(1651129200, 0).In(location))
fmt.Println(wm)

Output:

2022-04-28T07:00:00Z