p2p

Documentation

Index

• Constants
• Variables
• func EventId(channel channels.Channel, payload []byte) (hash.Hash, error)
• func MessageType(decodedPayload interface{}) string
• type Connector
• type IDTranslator
• type Network
  • func NewNetwork(param *NetworkParameters) (*Network, error)
  • func (n *Network) Identities() flow.IdentityList
  • func (n *Network) Identity(pid peer.ID) (*flow.Identity, bool)
  • func (n *Network) MulticastOnChannel(channel channels.Channel, message interface{}, num uint, ...) error
  • func (n *Network) PublishOnChannel(channel channels.Channel, message interface{}, targetIDs ...flow.Identifier) error
  • func (n *Network) Receive(nodeID flow.Identifier, msg *message.Message, decodedMsgPayload interface{}) error
  • func (n *Network) Register(channel channels.Channel, messageProcessor network.MessageProcessor) (network.Conduit, error)
  • func (n *Network) RegisterBlobService(channel channels.Channel, ds datastore.Batching, ...) (network.BlobService, error)
  • func (n *Network) RegisterPingService(pingProtocol protocol.ID, provider network.PingInfoProvider) (network.PingService, error)
  • func (n *Network) Topology() flow.IdentityList
  • func (n *Network) UnRegisterChannel(channel channels.Channel) error
  • func (n *Network) UnicastOnChannel(channel channels.Channel, message interface{}, targetID flow.Identifier) error
• type NetworkOptFunction
  • func WithConduitFactory(f network.ConduitFactory) NetworkOptFunction
• type NetworkParameters
• type PeerFilter
• type PeerManager
• type PeerManagerFactoryFunc
• type PeersProvider
• type SubscriptionProvider
• type TopicProvider

Constants

const (
	// DefaultReceiveCacheSize represents size of receive cache that keeps hash of incoming messages
	// for sake of deduplication.
	DefaultReceiveCacheSize = 10e4
)

Variables

var (
	// ErrInvalidId indicates that the given ID (either `peer.ID` or `flow.Identifier`) has an invalid format.
	ErrInvalidId = errors.New("empty peer ID")

	// ErrUnknownId indicates that the given ID (either `peer.ID` or `flow.Identifier`) is unknown.
	ErrUnknownId = errors.New("unknown ID")
)

var ErrNetworkShutdown = errors.New("network has already shutdown")

var NotEjectedFilter = filter.Not(filter.Ejected)

NotEjectedFilter is an identity filter that, when applied to the identity table at a given snapshot, returns all nodes that we should communicate with over the networking layer.

NOTE: The protocol state includes nodes from the previous/next epoch that should be included in network communication. We omit any nodes that have been ejected.

Functions

func EventId

func EventId(channel channels.Channel, payload []byte) (hash.Hash, error)

func MessageType

func MessageType(decodedPayload interface{}) string

Types

type Connector

type Connector interface {
	// UpdatePeers connects to the given peer.IDs. It also disconnects from any other peers with which it may have
	// previously established connection.
	// UpdatePeers implementation should be idempotent such that multiple calls to connect to the same peer should not
	// create multiple connections
	UpdatePeers(ctx context.Context, peerIDs peer.IDSlice)
}

Connector connects to peer and disconnects from peer using the underlying networking library

type IDTranslator

type IDTranslator interface {
	// GetPeerID returns the peer ID for the given `flow.Identifier`.
	// During normal operations, the following error returns are expected
	//  * ErrUnknownId if the given Identifier is unknown
	//  * ErrInvalidId if the given Identifier has an invalid format.
	//    This error return is only possible, when the Identifier is generated from some
	//    other input (e.g. the node's public key). While the Flow protocol itself makes
	//    no assumptions about the structure of the `Identifier`, protocol extensions
	//    (e.g. for Observers) might impose a strict relationship between both flow
	//    Identifier and peer ID, in which case they can use this error.
	// TODO: implementations do not fully adhere to this convention on error returns
	GetPeerID(flow.Identifier) (peer.ID, error)

	// GetFlowID returns the `flow.Identifier` for the given `peer.ID`.
	// During normal operations, the following error returns are expected
	//  * ErrUnknownId if the given Identifier is unknown
	//  * ErrInvalidId if the given Identifier has an invalid format.
	//    This error return is only possible, when the Identifier is generated from some
	//    other input (e.g. the node's public key). While the Flow protocol itself makes
	//    no assumptions about the structure of the `Identifier`, protocol extensions
	//    (e.g. for Observers) might impose a strict relationship between both flow
	//    Identifier and peer ID, in which case they can use this error.
	// TODO: implementations do not fully adhere to this convention on error returns
	GetFlowID(peer.ID) (flow.Identifier, error)
}

IDTranslator provides an interface for converting from Flow ID's to LibP2P peer ID's and vice versa.

type Network

type Network struct {
	sync.RWMutex
	*component.ComponentManager
	// contains filtered or unexported fields
}

Network represents the overlay network of our peer-to-peer network, including the protocols for handshakes, authentication, gossiping and heartbeats.

func NewNetwork

func NewNetwork(param *NetworkParameters) (*Network, error)

NewNetwork creates a new naive overlay network, using the given middleware to communicate to direct peers, using the given codec for serialization, and using the given state & cache interfaces to track volatile information. csize determines the size of the cache dedicated to keep track of received messages

func (*Network) Identities

func (n *Network) Identities() flow.IdentityList

func (*Network) Identity

func (n *Network) Identity(pid peer.ID) (*flow.Identity, bool)

func (*Network) MulticastOnChannel

func (n *Network) MulticastOnChannel(channel channels.Channel, message interface{}, num uint, targetIDs ...flow.Identifier) error

MulticastOnChannel unreliably sends the specified event over the channel to randomly selected 'num' number of recipients selected from the specified targetIDs.

func (*Network) PublishOnChannel

func (n *Network) PublishOnChannel(channel channels.Channel, message interface{}, targetIDs ...flow.Identifier) error

PublishOnChannel sends the message in an unreliable way to the given recipients. In this context, unreliable means that the message is published over a libp2p pub-sub channel and can be read by any node subscribed to that channel. The selector could be used to optimize or restrict delivery.

func (*Network) Receive

func (n *Network) Receive(nodeID flow.Identifier, msg *message.Message, decodedMsgPayload interface{}) error

func (*Network) Register

func (n *Network) Register(channel channels.Channel, messageProcessor network.MessageProcessor) (network.Conduit, error)

Register will register the given engine with the given unique engine engineID, returning a conduit to directly submit messages to the message bus of the engine.

func (*Network) RegisterBlobService

func (n *Network) RegisterBlobService(channel channels.Channel, ds datastore.Batching, opts ...network.BlobServiceOption) (network.BlobService, error)

RegisterBlobService registers a BlobService on the given channel. The returned BlobService can be used to request blobs from the network.

func (*Network) RegisterPingService

func (n *Network) RegisterPingService(pingProtocol protocol.ID, provider network.PingInfoProvider) (network.PingService, error)

func (*Network) Topology

func (n *Network) Topology() flow.IdentityList

func (*Network) UnRegisterChannel

func (n *Network) UnRegisterChannel(channel channels.Channel) error

UnRegisterChannel unregisters the engine for the specified channel. The engine will no longer be able to send or receive messages from that channel.

func (*Network) UnicastOnChannel

func (n *Network) UnicastOnChannel(channel channels.Channel, message interface{}, targetID flow.Identifier) error

UnicastOnChannel sends the message in a reliable way to the given recipient. It uses 1-1 direct messaging over the underlying network to deliver the message. It returns an error if unicasting fails.

type NetworkOptFunction

type NetworkOptFunction func(*Network)

func WithConduitFactory

func WithConduitFactory(f network.ConduitFactory) NetworkOptFunction

type NetworkParameters

type NetworkParameters struct {
	Logger              zerolog.Logger
	Codec               network.Codec
	Me                  module.Local
	MiddlewareFactory   func() (network.Middleware, error)
	Topology            network.Topology
	SubscriptionManager network.SubscriptionManager
	Metrics             module.NetworkMetrics
	IdentityProvider    module.IdentityProvider
	ReceiveCache        *netcache.ReceiveCache
	Options             []NetworkOptFunction
}

type PeerFilter

type PeerFilter func(peer.ID) error

type PeerManager

type PeerManager interface {
	component.Component

	// RequestPeerUpdate requests an update to the peer connections of this node.
	// If a peer update has already been requested (either as a periodic request or an on-demand
	// request) and is outstanding, then this call is a no-op.
	RequestPeerUpdate()

	// ForceUpdatePeers initiates an update to the peer connections of this node immediately
	ForceUpdatePeers(context.Context)

	// SetPeersProvider sets the peer managers's peers provider and may be called at most once
	SetPeersProvider(PeersProvider)
}

PeerManager adds and removes connections to peers periodically and on request

type PeerManagerFactoryFunc

type PeerManagerFactoryFunc func(host host.Host, peersProvider PeersProvider, logger zerolog.Logger) (PeerManager, error)

PeerManagerFactoryFunc is a factory function type for generating a PeerManager instance using the given host, peersProvider and logger

type PeersProvider

type PeersProvider func() peer.IDSlice

type SubscriptionProvider

type SubscriptionProvider interface {
	// GetSubscribedTopics returns all the subscriptions of a peer within the pubsub network.
	// Note that the current peer must be subscribed to the topic for it to the same topics in order
	// to query for other peers, e.g., if current peer has subscribed to topics A and B, and peer1
	// has subscribed to topics A, B, and C, then GetSubscribedTopics(peer1) will return A and B. Since this peer
	// has not subscribed to topic C, it will not be able to query for other peers subscribed to topic C.
	GetSubscribedTopics(pid peer.ID) []string
}

SubscriptionProvider provides a list of topics a peer is subscribed to.

type TopicProvider

type TopicProvider interface {
	// GetTopics returns all the topics within the pubsub network that the current peer has subscribed to.
	GetTopics() []string

	// ListPeers returns all the peers subscribed to a topic.
	// Note that the current peer must be subscribed to the topic for it to query for other peers.
	// If the current peer is not subscribed to the topic, an empty list is returned.
	// For example, if current peer has subscribed to topics A and B, then ListPeers only return
	// subscribed peers for topics A and B, and querying for topic C will return an empty list.
	ListPeers(topic string) []peer.ID
}

TopicProvider provides a low-level abstraction for pubsub to perform topic-related queries. This abstraction is provided to encapsulate the pubsub implementation details from the rest of the codebase.