scanner

Documentation

Index

• Constants
• Variables
• type BlockNotification
• type ConsumeRecordFn
• type DatasetNotification
• type IndexBlockConsumer
• type IndexSubscription
  • func NewIndexSubscription(indexID string, consumer IndexBlockConsumer) *IndexSubscription
  • func (w *IndexSubscription) AddLockers(lockers ...LockerEntry) error
  • func (w *IndexSubscription) ConsumeBlock(ctx context.Context, n BlockNotification) error
  • func (w *IndexSubscription) InactiveSince() int64
  • func (w *IndexSubscription) IndexID() string
  • func (w *IndexSubscription) LockerConfigs() []*LockerConfig
  • func (w *IndexSubscription) NotifyScanCompleted(topBlock int64) error
  • func (w *IndexSubscription) SetStatus(status int)
  • func (w *IndexSubscription) Status() int
• type LockerConfig
  • func (lc *LockerConfig) Hydrate() error
• type LockerEntry
• type MutableSubscription
• type PartyLookup
• type Scanner
  • func NewScanner(ledgerAPI model.Ledger) *Scanner
  • func (isu *Scanner) AddSubscription(sub Subscription) error
  • func (isu *Scanner) RemoveSubscription(indexID string) error
  • func (isu *Scanner) Scan() (bool, error)
  • func (isu *Scanner) ScanLedger(scannerList []*LockerConfig, startBlockNumber int64, endBlockNumber int64, ...) (int64, bool, error)
  • func (isu *Scanner) Subscriptions() map[string]Subscription
• type Subscription

Constants

const (
	ScanStatusActive = 0
	ScanStatusPaused = 1
	ScanStatusError  = 2
)

Variables

var (
	// ErrIndexResultPending is a signal used by ConsumeRecordFn functions to tell the scanner
	// that the given record will be processed outside of the scanner's loop and processing
	// of the underlying index (a set of locker participant keys that belongs to the same entity)
	// should be paused until the processing is completed.
	ErrIndexResultPending = errors.New("index result pending")
)

Types

type BlockNotification

type BlockNotification struct {
	Block    int64                 `json:"b"`
	Datasets []DatasetNotification `json:"ds"`
}

type ConsumeRecordFn

type ConsumeRecordFn func(blockNumber int64, keyID int, key []byte, idx uint32, r *model.Record) error

type DatasetNotification

type DatasetNotification struct {
	KeyID     int          `json:"kid"`
	RecordID  string       `json:"r"`
	Operation model.OpType `json:"t"`

	// The fields below are passed in order to optimise downstream processing
	Key    []byte        `json:"-"`
	Record *model.Record `json:"-"`
}

type IndexBlockConsumer

type IndexBlockConsumer interface {
	SetSubscription(sub Subscription)
	ConsumeBlock(ctx context.Context, indexID string, partyLookup PartyLookup, n BlockNotification) error
	NotifyScanCompleted(block int64) error
}

type IndexSubscription

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

func NewIndexSubscription

func NewIndexSubscription(indexID string, consumer IndexBlockConsumer) *IndexSubscription

func (*IndexSubscription) AddLockers

func (w *IndexSubscription) AddLockers(lockers ...LockerEntry) error

func (*IndexSubscription) ConsumeBlock

func (w *IndexSubscription) ConsumeBlock(ctx context.Context, n BlockNotification) error

func (*IndexSubscription) InactiveSince

func (w *IndexSubscription) InactiveSince() int64

func (*IndexSubscription) IndexID

func (w *IndexSubscription) IndexID() string

func (*IndexSubscription) LockerConfigs

func (w *IndexSubscription) LockerConfigs() []*LockerConfig

func (*IndexSubscription) NotifyScanCompleted

func (w *IndexSubscription) NotifyScanCompleted(topBlock int64) error

func (*IndexSubscription) SetStatus

func (w *IndexSubscription) SetStatus(status int)

func (*IndexSubscription) Status

func (w *IndexSubscription) Status() int

type LockerConfig

type LockerConfig struct {
	KeyID        int    `json:"key"`
	LastBlock    int64  `json:"last"`
	PublicKeyStr string `json:"pubk"`

	Subscription Subscription `json:"-"`
	// contains filtered or unexported fields
}

func (*LockerConfig) Hydrate

func (lc *LockerConfig) Hydrate() error

type LockerEntry

type LockerEntry struct {
	Locker    *model.Locker
	LastBlock int64
}

type MutableSubscription

type MutableSubscription interface {
	AddLockers(lockers ...LockerEntry) error
}

type PartyLookup

type PartyLookup func(keyID int) (string, string, string, int64)

type Scanner

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

func NewScanner

func NewScanner(ledgerAPI model.Ledger) *Scanner

func (*Scanner) AddSubscription

func (isu *Scanner) AddSubscription(sub Subscription) error

func (*Scanner) RemoveSubscription

func (isu *Scanner) RemoveSubscription(indexID string) error

func (*Scanner) Scan

func (isu *Scanner) Scan() (bool, error)

func (*Scanner) ScanLedger

func (isu *Scanner) ScanLedger(scannerList []*LockerConfig, startBlockNumber int64, endBlockNumber int64, blockBatchSize int) (int64, bool, error)

func (*Scanner) Subscriptions

func (isu *Scanner) Subscriptions() map[string]Subscription

type Subscription

type Subscription interface {
	IndexID() string
	LockerConfigs() []*LockerConfig
	ConsumeBlock(ctx context.Context, n BlockNotification) error
	NotifyScanCompleted(topBlock int64) error
	InactiveSince() int64
	Status() int
	SetStatus(status int)
}