ingest

type LedgerReporter interface {
	// OnNewLedger is called when the session begins processing
	// a ledger at the given sequence number.
	OnNewLedger(sequence uint32)
	// OnLedgerEntry is called when the session processes a transaction
	// from the io.LedgerReader
	OnLedgerTransaction()
	// OnEndLedger is called when the session finishes processing
	// a ledger.
	// if err is not nil it means that the session stoped processing the
	// ledger because of an error.
	// if shutdown is true it means the session stopped processing the
	// ledger because it received a shutdown signal
	OnEndLedger(err error, shutdown bool)
}

type LiveSession struct {
	Archive           historyarchive.ArchiveInterface
	LedgerBackend     ledgerbackend.LedgerBackend
	StellarCoreClient *stellarcore.Client
	// StellarCoreCursor defines cursor name used in `setcursor` command of
	// stellar-core. If you run multiple sessions against a single stellar-core
	// instance the cursor name needs to be different in each session.
	StellarCoreCursor string
	StatePipeline     *pipeline.StatePipeline
	StateReporter     StateReporter
	LedgerPipeline    *pipeline.LedgerPipeline
	LedgerReporter    LedgerReporter
	// TempSet is a store used to hold temporary objects generated during
	// state processing. If nil, defaults to io.MemoryTempSet.
	TempSet io.TempSet
	// contains filtered or unexported fields
}

type Session interface {
	// Run start the session and works as long as the session is active. If
	// you want to resume a session use Resume().
	Run() error
	// Resume resumes the session at ledger with a given sequence. It's up to
	// Session user to determine what was the last ledger processed by a
	// Session as it's stateless (or if Run() should be called first).
	Resume(ledgerSequence uint32) error
	// GetLatestProcessedLedger returns the latest ledger sequence processed in
	// the session. Return 0 if no ledgers were processed yet.
	// Please note that this value is not synchronized with pipeline hooks.
	GetLatestProcessedLedger() uint32
	// Shutdown gracefully stops running session and stops all internal
	// objects.
	// Calling Shutdown() does not trigger post processing hooks.
	Shutdown()
	// QueryLock locks the session for sending a query. This ensures a consistent
	// view when data is saved to multiple stores (ex. postgres/redis/memory).
	// TODO this is fine for a demo but we need to check if it works for systems
	// that don't provide strong consistency. This may also slow down readers
	// if commiting data to stores take longer time. This probably doesn't work
	// in distributed apps.
	QueryLock()
	// QueryUnlock unlocks read lock of the session.
	QueryUnlock()
	// UpdateLock locks the session for updating data. This ensures a consistent
	// view when data is saved to multiple stores (ex. postgres/redis/memory).
	// TODO this is fine for a demo but we need to check if it works for systems
	// that don't provide strong consistency. This may also slow down readers
	// if commiting data to stores take longer time. This probably doesn't work
	// in distributed apps.
	UpdateLock()
	// UpdateUnlock unlocks write lock of the session.
	UpdateUnlock()
}

type SingleLedgerSession struct {
	Archive        *historyarchive.Archive
	LedgerSequence uint32
	StatePipeline  *pipeline.StatePipeline
	StateReporter  StateReporter
	// TempSet is a store used to hold temporary objects generated during
	// state processing. If nil, defaults to io.MemoryTempSet.
	TempSet io.TempSet
	// contains filtered or unexported fields
}

type StateReporter interface {
	// OnStartState is called when the session begins processing
	// a history archive snapshot at the given sequence number.
	OnStartState(sequence uint32)
	// OnStateEntry is called when the session processes an entry
	// from the io.StateReader
	OnStateEntry()
	// OnEndState is called when the session finishes processing
	// a history archive snapshot.
	// if err is not nil it means that the session stoped processing the
	// history archive snapshot because of an error.
	// if shutdown is true it means the session stopped processing the
	// history archive snapshot because it received a shutdown signal
	OnEndState(err error, shutdown bool)
}