topics

Documentation

Index

• Constants
• Variables
• type Controller
  • func NewPubSub(ctx context.Context, logger *zap.Logger, cfg *PubSubConfig, metrics Metrics) (*pubsub.PubSub, Controller, error)
  • func NewTopicsController(ctx context.Context, logger *zap.Logger, msgHandler PubsubMessageHandler, ...) Controller
• type Metrics
• type MsgIDHandler
  • func NewMsgIDHandler(ctx context.Context, ttl time.Duration, ...) MsgIDHandler
• type MsgPeersResolver
• type PubSubConfig
• type PubsubBundle
• type PubsubMessageHandler
• type ScoringConfig
  • func DefaultScoringConfig() *ScoringConfig
• type SubFilter
• type Whitelist

Constants

const (
	// MsgIDEmptyMessage is the msg_id for empty messages
	MsgIDEmptyMessage = "invalid:empty"
	// MsgIDBadEncodedMessage is the msg_id for messages with invalid encoding
	MsgIDBadEncodedMessage = "invalid:encoding"
	// MsgIDError is the msg_id for messages that we can't create their msg_id
	MsgIDError = "invalid:msg_id_error"
	// MsgIDBadPeerID is the msg_id for messages w/o a valid sender
	MsgIDBadPeerID = "invalid:peer_id_error"
)

Variables

var (
	// ErrTopicNotReady happens when trying to access a topic which is not ready yet
	ErrTopicNotReady = errors.New("topic is not ready")
)

Types

type Controller

type Controller interface {
	// Subscribe subscribes to the given topic
	Subscribe(logger *zap.Logger, name string) error
	// Unsubscribe unsubscribes from the given topic
	Unsubscribe(logger *zap.Logger, topicName string, hard bool) error
	// Peers returns the peers subscribed to the given topic
	Peers(topicName string) ([]peer.ID, error)
	// Topics lists all the available topics
	Topics() []string
	// Broadcast publishes the message on the given topic
	Broadcast(topicName string, data []byte, timeout time.Duration) error

	io.Closer
}

Controller is an interface for managing pubsub topics

func NewPubSub

func NewPubSub(ctx context.Context, logger *zap.Logger, cfg *PubSubConfig, metrics Metrics) (*pubsub.PubSub, Controller, error)

NewPubSub creates a new pubsub router and the necessary components

func NewTopicsController

func NewTopicsController(
	ctx context.Context,
	logger *zap.Logger,
	msgHandler PubsubMessageHandler,
	msgValidator messageValidator,
	subFilter SubFilter,
	pubSub *pubsub.PubSub,
	scoreParams func(string) *pubsub.TopicScoreParams,
) Controller

NewTopicsController creates an instance of Controller

type Metrics

type Metrics interface {
	PeerScore(peer.ID, float64)
	PeerP4Score(peer.ID, float64)
	ResetPeerScores()
}

type MsgIDHandler

type MsgIDHandler interface {
	MsgPeersResolver
	MsgID(logger *zap.Logger) func(pmsg *ps_pb.Message) string

	Start()
	GC()
}

MsgIDHandler stores msgIDs and the corresponding sender peer.ID it works in memory as this store is expected to be invoked a lot, adding msgID and peerID pairs for every message this uses to identify msg senders after validation

func NewMsgIDHandler

func NewMsgIDHandler(ctx context.Context, ttl time.Duration, networkConfig networkconfig.NetworkConfig) MsgIDHandler

NewMsgIDHandler creates a new MsgIDHandler

type MsgPeersResolver

type MsgPeersResolver interface {
	GetPeers(msg []byte) []peer.ID
}

MsgPeersResolver will resolve the sending peers of the given message

type PubSubConfig

type PubSubConfig struct {
	Host        host.Host
	TraceLog    bool
	StaticPeers []peer.AddrInfo
	MsgHandler  PubsubMessageHandler
	// MsgValidator accepts the topic name and returns the corresponding msg validator
	// in case we need different validators for specific topics,
	// this should be the place to map a validator to topic
	MsgValidator messageValidator
	ScoreIndex   peers.ScoreIndex
	Scoring      *ScoringConfig
	MsgIDHandler MsgIDHandler
	Discovery    discovery.Discovery

	ValidateThrottle    int
	ValidationQueueSize int
	OutboundQueueSize   int
	MsgIDCacheTTL       time.Duration

	GetValidatorStats      network.GetValidatorStats
	ScoreInspector         pubsub.ExtendedPeerScoreInspectFn
	ScoreInspectorInterval time.Duration
}

PubSubConfig is the needed config to instantiate pubsub

type PubsubBundle

type PubsubBundle struct {
	PS         *pubsub.PubSub
	TopicsCtrl Controller
	Resolver   MsgPeersResolver
}

PubsubBundle includes the pubsub router, plus involved components

type PubsubMessageHandler

type PubsubMessageHandler func(context.Context, string, *pubsub.Message) error

PubsubMessageHandler handles incoming messages

type ScoringConfig

type ScoringConfig struct {
	IPWhilelist        []*net.IPNet
	IPColocationWeight float64
	OneEpochDuration   time.Duration
}

ScoringConfig is the configuration for peer scoring

func DefaultScoringConfig

func DefaultScoringConfig() *ScoringConfig

DefaultScoringConfig returns the default scoring config

type SubFilter

type SubFilter interface {
	// SubscriptionFilter allows controlling what topics the node will subscribe to
	// otherwise it might subscribe to irrelevant topics that were suggested by other peers
	pubsub.SubscriptionFilter
}

SubFilter is a wrapper on top of pubsub.SubscriptionFilter,

type Whitelist

type Whitelist interface {
	// Register adds the given name to the whitelist
	Register(name string)
	// Deregister removes the given name from the whitelist
	Deregister(name string)
	// Whitelisted checks if the given name was whitelisted
	Whitelisted(name string) bool
}

Whitelist is an interface to maintain dynamic whitelists