dispatcher

Documentation

Index

• Variables
• type Config
• type Dispatcher
  • func NewDispatcher(cfg Config, verificationFunc VerificationFunc) (Dispatcher, error)
• type IotxDispatcher
  • func (d *IotxDispatcher) AddSubscriber(chainID uint32, subscriber Subscriber)
  • func (d *IotxDispatcher) EventAudit() map[iotexrpc.MessageType]int
  • func (d *IotxDispatcher) EventQueueSize() map[string]int
  • func (d *IotxDispatcher) HandleBroadcast(ctx context.Context, chainID uint32, peer string, msgProto proto.Message)
  • func (d *IotxDispatcher) HandleTell(ctx context.Context, chainID uint32, peer peer.AddrInfo, ...)
  • func (d *IotxDispatcher) Start(ctx context.Context) error
  • func (d *IotxDispatcher) Stop(ctx context.Context) error
  • func (d *IotxDispatcher) ValidateMessage(pMsg proto.Message) (bool, error)
• type RateLimiter
  • func NewRateLimiter(s int, r rate.Limit, b int) *RateLimiter
  • func (rl *RateLimiter) Remainings(key string) int
  • func (rl *RateLimiter) Wait(key string)
• type Subscriber
• type VerificationFunc

Variables

var (
	// DefaultConfig is the default config
	DefaultConfig = Config{
		ActionChanSize:    5000,
		BlockChanSize:     1000,
		BlockSyncChanSize: 400,
		ConsensusChanSize: 1000,
		MiscChanSize:      1000,
		AccountRateLimit:  100,

		ProcessSyncRequestInterval: 0 * time.Second,
	}
)

Types

type Config

type Config struct {
	ActionChanSize             uint          `yaml:"actionChanSize"`
	BlockChanSize              uint          `yaml:"blockChanSize"`
	BlockSyncChanSize          uint          `yaml:"blockSyncChanSize"`
	ConsensusChanSize          uint          `yaml:"consensusChanSize"`
	MiscChanSize               uint          `yaml:"miscChanSize"`
	ProcessSyncRequestInterval time.Duration `yaml:"processSyncRequestInterval"`
	AccountRateLimit           uint          `yaml:"accountRateLimit"`
}

Config is the config for dispatcher

type Dispatcher

type Dispatcher interface {
	lifecycle.StartStopper

	// AddSubscriber adds to dispatcher
	AddSubscriber(uint32, Subscriber)
	// ValidateMessage validates the message
	ValidateMessage(proto.Message) (bool, error)
	// HandleBroadcast handles the incoming broadcast message. The transportation layer semantics is at least once.
	// That said, the handler is likely to receive duplicate messages.
	HandleBroadcast(context.Context, uint32, string, proto.Message)
	// HandleTell handles the incoming tell message. The transportation layer semantics is exact once. The sender is
	// given for the sake of replying the message
	HandleTell(context.Context, uint32, peer.AddrInfo, proto.Message)
}

Dispatcher is used by peers, handles incoming block and header notifications and relays announcements of new blocks.

func NewDispatcher

func NewDispatcher(cfg Config, verificationFunc VerificationFunc) (Dispatcher, error)

NewDispatcher creates a new Dispatcher

type IotxDispatcher

type IotxDispatcher struct {
	lifecycle.Readiness
	lifecycle.Lifecycle
	// contains filtered or unexported fields
}

IotxDispatcher is the request and event dispatcher for iotx node.

func (*IotxDispatcher) AddSubscriber

func (d *IotxDispatcher) AddSubscriber(
	chainID uint32,
	subscriber Subscriber,
)

AddSubscriber adds a subscriber to dispatcher

func (*IotxDispatcher) EventAudit

func (d *IotxDispatcher) EventAudit() map[iotexrpc.MessageType]int

EventAudit returns the event audit map

func (*IotxDispatcher) EventQueueSize

func (d *IotxDispatcher) EventQueueSize() map[string]int

EventQueueSize returns the event queue size

func (*IotxDispatcher) HandleBroadcast

func (d *IotxDispatcher) HandleBroadcast(ctx context.Context, chainID uint32, peer string, msgProto proto.Message)

HandleBroadcast handles incoming broadcast message

func (*IotxDispatcher) HandleTell

func (d *IotxDispatcher) HandleTell(ctx context.Context, chainID uint32, peer peer.AddrInfo, msgProto proto.Message)

HandleTell handles incoming unicast message

func (*IotxDispatcher) Start

func (d *IotxDispatcher) Start(ctx context.Context) error

Start starts the dispatcher.

func (*IotxDispatcher) Stop

func (d *IotxDispatcher) Stop(ctx context.Context) error

Stop gracefully shuts down the dispatcher by stopping all handlers and waiting for them to finish.

func (*IotxDispatcher) ValidateMessage

func (d *IotxDispatcher) ValidateMessage(pMsg proto.Message) (bool, error)

ValidateMessage validates the message

type RateLimiter

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

RateLimiter is a struct that manages a threadsafe map of rate limiters.

func NewRateLimiter

func NewRateLimiter(s int, r rate.Limit, b int) *RateLimiter

NewRateLimiter creates a new RateLimiter.

func (*RateLimiter) Remainings

func (rl *RateLimiter) Remainings(key string) int

Remainings returns the number of remaining tokens for the given key.

func (*RateLimiter) Wait

func (rl *RateLimiter) Wait(key string)

Wait waits for 1 token to become available for the given key, up to the given duration.

type Subscriber

type Subscriber interface {
	ReportFullness(context.Context, iotexrpc.MessageType, float32)
	HandleAction(context.Context, *iotextypes.Action) error
	HandleBlock(context.Context, string, *iotextypes.Block) error
	HandleSyncRequest(context.Context, peer.AddrInfo, *iotexrpc.BlockSync) error
	HandleConsensusMsg(*iotextypes.ConsensusMessage) error
	HandleNodeInfoRequest(context.Context, peer.AddrInfo, *iotextypes.NodeInfoRequest) error
	HandleNodeInfo(context.Context, string, *iotextypes.NodeInfo) error
	HandleActionRequest(ctx context.Context, peer peer.AddrInfo, actHash hash.Hash256) error
	HandleActionHash(ctx context.Context, actHash hash.Hash256, from string) error
}

Subscriber is the dispatcher subscriber interface

type VerificationFunc

type VerificationFunc func(pm proto.Message) (string, error)