progress

Documentation

Index

• func CompareBatchTransactions(a *ordered_map.OrderedMap, b *ordered_map.OrderedMap) bool
• func UpdateTransactions(msg *marshaller.MarshalledMessage, transactions *ordered_map.OrderedMap)
• type Ledger
  • func NewLedger() Ledger
• type LedgerEntry
• type Progress
• type ProgressTracker
  • func New(shutdownHandler shutdown.ShutdownHandler, txnSeenChan <-chan *Seen, ...) ProgressTracker
  • func (p *ProgressTracker) Start(tickerDuration time.Duration)
• type Seen
• type Written

Functions

func CompareBatchTransactions

func CompareBatchTransactions(a *ordered_map.OrderedMap, b *ordered_map.OrderedMap) bool

CompareBatchTransactions compares two BatchTransactions for equality of their fields.

func UpdateTransactions

func UpdateTransactions(msg *marshaller.MarshalledMessage, transactions *ordered_map.OrderedMap)

Types

type Ledger

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

Ledger keeps track of ledger entries in order. When any given ledger entry count is 0 and it has a non-nil CommitWalStart, the WalsStart is emitted and the entry is cleared.

func NewLedger

func NewLedger() Ledger

type LedgerEntry

type LedgerEntry struct {
	Transaction    string // transaction id of the transaction
	TimeBasedKey   string // time based key of the transaction (composite key of txn id and time)
	CommitWalStart uint64 // WalStart of the commit
	Count          int    // number of messages written (number of messages that were in the batch for this transaction)
	TotalMsgs      int    // total number of messages in transaction
}

LedgerEntry are used by ledger to keep track of the current count of wal messages per transaction that have been both seen and written. It uses WalTuple to back-reference to the transactionToTimeBasedKey.

type Progress

type Progress struct {
	WorkerNum int    // worker id
	WalStart  uint64 // wal offset of the latest _contiguous_ record processed by worker
}

type ProgressTracker

type ProgressTracker struct {
	OutputChan chan uint64 // channel to send overall (youngest) progress on
	// contains filtered or unexported fields
}

func New

func New(shutdownHandler shutdown.ShutdownHandler,
	txnSeenChan <-chan *Seen,
	txnsWritten <-chan *ordered_map.OrderedMap,
	statsChan chan stats.Stat) ProgressTracker

New creates a progress "table" based on the number of workers (inputChans) and returns a ProgressTracker txnsSeen MUST be an unbuffered channel

func (*ProgressTracker) Start

func (p *ProgressTracker) Start(tickerDuration time.Duration)

Start begins progress tracking which emits progress on a timer or otherwise updates the ledger when transactions ares seen from the batcher or successfully written out by the transporter

type Seen

type Seen struct {
	Transaction    string // transaction id of the transaction
	TimeBasedKey   string // time based key of the transaction (composite key of txn id and time)
	TotalMsgs      int    // total number of messages in transaction
	CommitWalStart uint64 // WalStart of the commit
}

type Written

type Written struct {
	Transaction  string // transaction id of the transaction
	TimeBasedKey string // time based key of the transaction (composite key of txn id and time)
	Count        int    // number of messages written (number of messages that were in the batch for this transaction)
}