processor

Documentation

Index

• Constants
• Variables
• func BroadcastCaller(handler interface{}, params ...interface{})
• func TransactionProcessedCaller(handler interface{}, params ...interface{})
• type CachedWorkUnit
  • func (c *CachedWorkUnit) WorkUnit() *WorkUnit
• type Events
• type Options
• type Processor
  • func New(requestQueue rqueue.Queue, peerManager *peering.Manager, opts *Options) *Processor
  • func (proc *Processor) CompressAndEmit(tx *transaction.Transaction, txTrits trinary.Trits) error
  • func (proc *Processor) Process(p *peer.Peer, msgType message.Type, data []byte)
  • func (proc *Processor) Run(shutdownSignal <-chan struct{})
  • func (proc *Processor) ValidateTimestamp(hornetTx *hornet.Transaction) (valid, broadcast bool)
  • func (proc *Processor) ValidateTransactionTrytesAndEmit(txTrytes trinary.Trytes) error
  • func (proc *Processor) WorkUnitsSize() int
• type WorkUnit
  • func (wu *WorkUnit) Is(state WorkUnitState) bool
  • func (wu *WorkUnit) ObjectStorageKey() []byte
  • func (wu *WorkUnit) ObjectStorageValue() []byte
  • func (wu *WorkUnit) UnmarshalObjectStorageValue(_ []byte) (consumedBytes int, err error)
  • func (wu *WorkUnit) Update(_ objectstorage.StorableObject)
  • func (wu *WorkUnit) UpdateState(state WorkUnitState)
• type WorkUnitState

Constants

const (
	WorkerQueueSize = 50000
)

Variables

var (
	ErrInvalidTimestamp = errors.New("invalid timestamp")
)

Functions

func BroadcastCaller

func BroadcastCaller(handler interface{}, params ...interface{})

func TransactionProcessedCaller

func TransactionProcessedCaller(handler interface{}, params ...interface{})

Types

type CachedWorkUnit

type CachedWorkUnit struct {
	objectstorage.CachedObject
}

CachedWorkUnit represents a cached WorkUnit.

func (*CachedWorkUnit) WorkUnit

func (c *CachedWorkUnit) WorkUnit() *WorkUnit

WorkUnit gets the underlying WorkUnit.

type Events

type Events struct {
	// Fired when a transaction was fully processed.
	TransactionProcessed *events.Event
	// Fired when a transaction is meant to be broadcasted.
	BroadcastTransaction *events.Event
}

Events are the events fired by the Processor.

type Options

type Options struct {
	ValidMWM          uint64
	WorkUnitCacheOpts profile.CacheOpts
}

The Options for the Processor.

type Processor

type Processor struct {
	Events Events
	// contains filtered or unexported fields
}

Processor processes submitted messages in parallel and fires appropriate completion events.

func New

func New(requestQueue rqueue.Queue, peerManager *peering.Manager, opts *Options) *Processor

New creates a new processor which parses messages.

func (*Processor) CompressAndEmit

func (proc *Processor) CompressAndEmit(tx *transaction.Transaction, txTrits trinary.Trits) error

CompressAndEmit compresses the given transaction and emits TransactionProcessed and BroadcastTransaction events. This function does not run within the Processor's worker pool.

func (*Processor) Process

func (proc *Processor) Process(p *peer.Peer, msgType message.Type, data []byte)

Process submits the given message to the processor for processing.

func (*Processor) Run

func (proc *Processor) Run(shutdownSignal <-chan struct{})

Run runs the processor and blocks until the shutdown signal is triggered.

func (*Processor) ValidateTimestamp

func (proc *Processor) ValidateTimestamp(hornetTx *hornet.Transaction) (valid, broadcast bool)

checks whether the given transaction's timestamp is valid. the timestamp is automatically valid if the transaction is a solid entry point. the timestamp should be in the range of +/- 10 minutes to current time.

func (*Processor) ValidateTransactionTrytesAndEmit

func (proc *Processor) ValidateTransactionTrytesAndEmit(txTrytes trinary.Trytes) error

ValidateTransactionTrytesAndEmit validates the given transaction trytes which were not received via gossip but through some other mechanism. This function does not run within the Processor's worker pool. Emits a TransactionProcessed and BroadcastTransaction event if the transaction was processed.

func (*Processor) WorkUnitsSize

func (proc *Processor) WorkUnitsSize() int

WorkUnitSize returns the size of WorkUnits currently cached.

type WorkUnit

type WorkUnit struct {
	objectstorage.StorableObjectFlags
	// contains filtered or unexported fields
}

WorkUnit defines the work around processing a received transaction and its associated requests from peers. There is at most one WorkUnit active per same transaction bytes.

func (*WorkUnit) Is

func (wu *WorkUnit) Is(state WorkUnitState) bool

Is tells whether the WorkUnit has the given state.

func (*WorkUnit) ObjectStorageKey

func (wu *WorkUnit) ObjectStorageKey() []byte

func (*WorkUnit) ObjectStorageValue

func (wu *WorkUnit) ObjectStorageValue() []byte

func (*WorkUnit) UnmarshalObjectStorageValue

func (wu *WorkUnit) UnmarshalObjectStorageValue(_ []byte) (consumedBytes int, err error)

func (*WorkUnit) Update

func (wu *WorkUnit) Update(_ objectstorage.StorableObject)

func (*WorkUnit) UpdateState

func (wu *WorkUnit) UpdateState(state WorkUnitState)

UpdateState updates the WorkUnit's state.

type WorkUnitState

type WorkUnitState byte

WorkUnitState defines the state which a WorkUnit is in.

const (
	Hashing WorkUnitState = 1 << 0
	Invalid WorkUnitState = 1 << 1
	Hashed  WorkUnitState = 1 << 2
)