log

Documentation

Index

• func NewBroadcaster(orm ORM, ethClient eth.Client, config Config, logger *logger.Logger, ...) *broadcaster
• func NewORM(db *gorm.DB) *orm
• type Broadcast
  • func NewLogBroadcast(rawLog types.Log, decodedLog interface{}) Broadcast
• type Broadcaster
• type BroadcasterInTest
• type Config
• type Listener
• type ListenerOpts
• type LogBroadcast
  • func (b LogBroadcast) AsKey() LogBroadcastAsKey
• type LogBroadcastAsKey
  • func NewLogBroadcastAsKey(log types.Log, listener Listener) LogBroadcastAsKey
• type NullBroadcaster
  • func (n *NullBroadcaster) AddDependents(int)
  • func (n *NullBroadcaster) AwaitDependents() <-chan struct{}
  • func (n *NullBroadcaster) BackfillBlockNumber() null.Int64
  • func (n *NullBroadcaster) Close() error
  • func (n *NullBroadcaster) DependentReady()
  • func (n *NullBroadcaster) Healthy() error
  • func (n *NullBroadcaster) IsConnected() bool
  • func (n *NullBroadcaster) MarkConsumed(db *gorm.DB, lb Broadcast) error
  • func (n *NullBroadcaster) OnNewLongestChain(context.Context, models.Head)
  • func (n *NullBroadcaster) Ready() error
  • func (n *NullBroadcaster) Register(listener Listener, opts ListenerOpts) (unsubscribe func())
  • func (n *NullBroadcaster) ReplayFromBlock(number int64)
  • func (n *NullBroadcaster) Start() error
  • func (n *NullBroadcaster) TrackedAddressesCount() uint32
  • func (n *NullBroadcaster) WasAlreadyConsumed(db *gorm.DB, lb Broadcast) (bool, error)
• type ORM
• type ParseLogFunc
• type Topic
• type Uint64
  • func (a Uint64) Compare(b heaps.Item) int

Functions

func NewBroadcaster

func NewBroadcaster(orm ORM, ethClient eth.Client, config Config, logger *logger.Logger, highestSavedHead *models.Head) *broadcaster

NewBroadcaster creates a new instance of the broadcaster

func NewORM

func NewORM(db *gorm.DB) *orm

Types

type Broadcast

type Broadcast interface {
	DecodedLog() interface{}
	RawLog() types.Log
	String() string
	LatestBlockNumber() uint64
	LatestBlockHash() common.Hash
	JobID() int32
}

The Broadcast type wraps a types.Log but provides additional functionality for determining whether or not the log has been consumed and for marking the log as consumed

func NewLogBroadcast

func NewLogBroadcast(rawLog types.Log, decodedLog interface{}) Broadcast

type Broadcaster

type Broadcaster interface {
	utils.DependentAwaiter
	service.Service
	httypes.HeadTrackable
	ReplayFromBlock(number int64)

	IsConnected() bool
	Register(listener Listener, opts ListenerOpts) (unsubscribe func())

	WasAlreadyConsumed(db *gorm.DB, lb Broadcast) (bool, error)
	MarkConsumed(db *gorm.DB, lb Broadcast) error
}

The Broadcaster manages log subscription requests for the Chainlink node. Instead of creating a new subscription for each request, it multiplexes all subscriptions to all of the relevant contracts over a single connection and forwards the logs to the relevant subscribers.

In case of node crash and/or restart, the logs will be backfilled from the latest head from DB, for subscribers that are added before all dependents of LogBroadcaster are done.

If a subscriber is added after the LogBroadcaster does the initial backfill, then it's possible/likely that the backfill fill only have depth: 1 (from latest head)

Of course, these backfilled logs + any new logs will only be sent after the NumConfirmations for given subscriber.

type BroadcasterInTest

type BroadcasterInTest interface {
	Broadcaster
	BackfillBlockNumber() null.Int64
	TrackedAddressesCount() uint32
}

type Config

type Config interface {
	BlockBackfillDepth() uint64
	BlockBackfillSkip() bool
	EvmFinalityDepth() uint
	EvmLogBackfillBatchSize() uint32
}

type Listener

type Listener interface {
	HandleLog(b Broadcast)
	JobID() int32
}

The Listener responds to log events through HandleLog.

type ListenerOpts

type ListenerOpts struct {
	Contract common.Address

	// Event types to receive, with value filter for each field in the event
	// No filter or an empty filter for a given field position mean: all values allowed
	// the key should be a result of AbigenLog.Topic() call
	LogsWithTopics map[common.Hash][][]Topic

	ParseLog ParseLogFunc

	// Minimum number of block confirmations before the log is received
	NumConfirmations uint64
}

type LogBroadcast

type LogBroadcast struct {
	BlockHash common.Hash
	LogIndex  uint
	JobID     int32
}

LogBroadcast - gorm-compatible receive data from log_broadcasts table columns

func (LogBroadcast) AsKey

func (b LogBroadcast) AsKey() LogBroadcastAsKey

type LogBroadcastAsKey

type LogBroadcastAsKey struct {
	BlockHash common.Hash
	LogIndex  uint
	JobId     int32
}

LogBroadcastAsKey - used as key in a map to filter out already consumed logs

func NewLogBroadcastAsKey

func NewLogBroadcastAsKey(log types.Log, listener Listener) LogBroadcastAsKey

type NullBroadcaster

type NullBroadcaster struct{ ErrMsg string }

func (*NullBroadcaster) AddDependents

func (n *NullBroadcaster) AddDependents(int)

func (*NullBroadcaster) AwaitDependents

func (n *NullBroadcaster) AwaitDependents() <-chan struct{}

func (*NullBroadcaster) BackfillBlockNumber

func (n *NullBroadcaster) BackfillBlockNumber() null.Int64

func (*NullBroadcaster) Close

func (n *NullBroadcaster) Close() error

func (*NullBroadcaster) DependentReady

func (n *NullBroadcaster) DependentReady()

func (*NullBroadcaster) Healthy

func (n *NullBroadcaster) Healthy() error

func (*NullBroadcaster) IsConnected

func (n *NullBroadcaster) IsConnected() bool

func (*NullBroadcaster) MarkConsumed

func (n *NullBroadcaster) MarkConsumed(db *gorm.DB, lb Broadcast) error

func (*NullBroadcaster) OnNewLongestChain

func (n *NullBroadcaster) OnNewLongestChain(context.Context, models.Head)

func (*NullBroadcaster) Ready

func (n *NullBroadcaster) Ready() error

func (*NullBroadcaster) Register

func (n *NullBroadcaster) Register(listener Listener, opts ListenerOpts) (unsubscribe func())

func (*NullBroadcaster) ReplayFromBlock

func (n *NullBroadcaster) ReplayFromBlock(number int64)

func (*NullBroadcaster) Start

func (n *NullBroadcaster) Start() error

func (*NullBroadcaster) TrackedAddressesCount

func (n *NullBroadcaster) TrackedAddressesCount() uint32

func (*NullBroadcaster) WasAlreadyConsumed

func (n *NullBroadcaster) WasAlreadyConsumed(db *gorm.DB, lb Broadcast) (bool, error)

type ORM

type ORM interface {
	FindConsumedLogs(fromBlockNum int64, toBlockNum int64) ([]LogBroadcast, error)
	WasBroadcastConsumed(tx *gorm.DB, blockHash common.Hash, logIndex uint, jobID int32) (bool, error)
	MarkBroadcastConsumed(tx *gorm.DB, blockHash common.Hash, blockNumber uint64, logIndex uint, jobID int32) error
}

type ParseLogFunc

type ParseLogFunc func(log types.Log) (generated.AbigenLog, error)

type Topic

type Topic common.Hash

type Uint64

type Uint64 int

func (Uint64) Compare

func (a Uint64) Compare(b heaps.Item) int