p2p

Documentation

Overview

Package p2p provides a networking api for creating p2p protocols by enabling sending direct messages to a set of provided neighbors or broadcasting a message to all of them. the discovery, connectivity and encryption is completely transparent to consumers. NOTE: gossip protocols must take care of their own message validation.

Index

• Constants
• Variables
• func ExtractData(pm *Payload) (service.Data, error)
• func StringIdentifiers(boot ...*Switch) []string
• type IntegrationTestSuite
  • func (its *IntegrationTestSuite) ForAll(f func(idx int, s NodeTestInstance) error, filter []int) []error
  • func (its *IntegrationTestSuite) ForAllAsync(ctx context.Context, f func(idx int, s NodeTestInstance) error) ([]error, error)
  • func (its *IntegrationTestSuite) SetupSuite()
  • func (its *IntegrationTestSuite) TearDownSuite()
  • func (its *IntegrationTestSuite) WaitForGossip(ctx context.Context) error
• type Lookuper
• type NodeTestInstance
• type Payload
  • func CreatePayload(data service.Data) (*Payload, error)
• type ProtocolMessage
• type ProtocolMessageMetadata
• type Service
• type Switch
  • func New(ctx context.Context, config config.Config, logger log.Log, path string) (*Switch, error)
  • func (s *Switch) Broadcast(ctx context.Context, protocol string, payload []byte) error
  • func (s *Switch) Disconnect(peer p2pcrypto.PublicKey)
  • func (s *Switch) GossipReady() <-chan struct{}
  • func (s *Switch) LocalNode() node.LocalNode
  • func (s *Switch) ProcessDirectProtocolMessage(ctx context.Context, sender p2pcrypto.PublicKey, protocol string, ...) error
  • func (s *Switch) ProcessGossipProtocolMessage(ctx context.Context, sender p2pcrypto.PublicKey, ownMessage bool, ...) error
  • func (s *Switch) RegisterDirectProtocol(protocol string) chan service.DirectMessage
  • func (s *Switch) RegisterDirectProtocolWithChannel(protocol string, ingressChannel chan service.DirectMessage) chan service.DirectMessage
  • func (s *Switch) RegisterGossipProtocol(protocol string, prio priorityq.Priority) chan service.GossipMessage
  • func (s *Switch) SendMessage(ctx context.Context, peerPubkey p2pcrypto.PublicKey, protocol string, ...) error
  • func (s *Switch) SendWrappedMessage(ctx context.Context, nodeID p2pcrypto.PublicKey, protocol string, ...) error
  • func (s *Switch) Shutdown()
  • func (s *Switch) Start(ctx context.Context) error
  • func (s *Switch) SubscribePeerEvents() (conn, disc chan p2pcrypto.PublicKey)
• type UDPMux
  • func NewUDPMux(ctx, shutdownCtx context.Context, localNode node.LocalNode, lookuper Lookuper, ...) *UDPMux
  • func (mux *UDPMux) ProcessDirectProtocolMessage(sender p2pcrypto.PublicKey, protocol string, data service.Data, ...) error
  • func (mux *UDPMux) RegisterDirectProtocolWithChannel(name string, c chan service.DirectMessage) chan service.DirectMessage
  • func (mux *UDPMux) SendMessage(ctx context.Context, peerPubkey p2pcrypto.PublicKey, protocol string, ...) error
  • func (mux *UDPMux) SendWrappedMessage(ctx context.Context, nodeID p2pcrypto.PublicKey, protocol string, ...) error
  • func (mux *UDPMux) Shutdown()
  • func (mux *UDPMux) Start() error

Constants

const ConnectingTimeout = 20 * time.Second // todo: add to the config

ConnectingTimeout is the timeout we wait when trying to connect a neighborhood

const NoResultsInterval = 1 * time.Second

NoResultsInterval is the timeout we wait between requesting more peers repeatedly

const UPNPRetries = 20

UPNPRetries is the number of times to retry obtaining a port due to a UPnP failure

Variables

var (
	// ErrBadFormat1 could'nt deserialize the payload
	ErrBadFormat1 = errors.New("bad msg format, couldn't deserialize 1")
	// ErrBadFormat2 could'nt deserialize the protocol message payload
	ErrBadFormat2 = errors.New("bad msg format, couldn't deserialize 2")
	// ErrOutOfSync is returned when message timestamp was out of sync
	ErrOutOfSync = errors.New("received out of sync msg")
	// ErrFailDecrypt session cant decrypt
	ErrFailDecrypt = errors.New("can't decrypt message payload with session key")
	// ErrNoProtocol we don't have the protocol message
	ErrNoProtocol = errors.New("received msg to an unsupported protocol")
	// ErrNoSession we don't have this session
	ErrNoSession = errors.New("connection is missing a session")
)

Functions

func ExtractData

func ExtractData(pm *Payload) (service.Data, error)

ExtractData is a helper function to extract the payload data from a message payload.

func StringIdentifiers

func StringIdentifiers(boot ...*Switch) []string

StringIdentifiers turns Switch into string representation node for use as bootnodes.

Types

type IntegrationTestSuite

type IntegrationTestSuite struct {
	suite.Suite

	BootstrapNodesCount   int
	BootstrappedNodeCount int
	NeighborsCount        int

	BeforeHook func(idx int, s NodeTestInstance)
	AfterHook  func(idx int, s NodeTestInstance)

	Instances []*Switch
	// contains filtered or unexported fields
}

IntegrationTestSuite is a suite which bootstraps a network according to the given params and lets you run actions on this network. You must set the params before running the suite.

func (*IntegrationTestSuite) ForAll

func (its *IntegrationTestSuite) ForAll(f func(idx int, s NodeTestInstance) error, filter []int) []error

ForAll executes f on all the node and returns error if it failed

func (*IntegrationTestSuite) ForAllAsync

func (its *IntegrationTestSuite) ForAllAsync(ctx context.Context, f func(idx int, s NodeTestInstance) error) ([]error, error)

ForAllAsync executes f on all the nodes concurrently, it stops if ctx is cancelled.

func (*IntegrationTestSuite) SetupSuite

func (its *IntegrationTestSuite) SetupSuite()

SetupSuite setups the configured nodes, bootstraps and connects them.

func (*IntegrationTestSuite) TearDownSuite

func (its *IntegrationTestSuite) TearDownSuite()

TearDownSuite shutdowns all nodes.

func (*IntegrationTestSuite) WaitForGossip

func (its *IntegrationTestSuite) WaitForGossip(ctx context.Context) error

WaitForGossip waits that all nodes initialized gossip connections

type Lookuper

type Lookuper func(key p2pcrypto.PublicKey) (*node.Info, error)

Lookuper is a service used to lookup for nodes we know already

type NodeTestInstance

type NodeTestInstance interface {
	Service
	LocalNode() node.LocalNode // this holds the keys
}

NodeTestInstance is an instance of a p2p node for testing

type Payload

type Payload struct {
	Payload []byte
	Wrapped *service.DataMsgWrapper
}

Payload holds either a byte array or a wrapped req-res message.

func CreatePayload

func CreatePayload(data service.Data) (*Payload, error)

CreatePayload is a helper function to format a payload for sending.

type ProtocolMessage

type ProtocolMessage struct {
	Metadata *ProtocolMessageMetadata
	Payload  *Payload
}

ProtocolMessage is a pair of metadata and a a payload.

type ProtocolMessageMetadata

type ProtocolMessageMetadata struct {
	NextProtocol  string
	ClientVersion string
	Timestamp     int64
	AuthPubkey    []byte
	NetworkID     int32
}

ProtocolMessageMetadata is a general p2p message wrapper

type Service

type Service service.Service

Service is a wrapper for service.Service to expose the Service interface to `p2p` package clients

type Switch

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

Switch is the heart of the p2p package. it runs and orchestrates all services within it. It provides the external interface for protocols to access peers or receive incoming messages.

func New

func New(ctx context.Context, config config.Config, logger log.Log, path string) (*Switch, error)

New creates a new P2P service a.k.a `Switch` it loads existing node information from the disk or creates a new one.

func (*Switch) Broadcast

func (s *Switch) Broadcast(ctx context.Context, protocol string, payload []byte) error

Broadcast creates a gossip message signs it and disseminate it to neighbors. this message must be validated by our own node first just as any other message.

func (*Switch) Disconnect

func (s *Switch) Disconnect(peer p2pcrypto.PublicKey)

Disconnect removes a peer from the neighborhood. It requests more peers if our outbound peer count is less than configured

func (*Switch) GossipReady

func (s *Switch) GossipReady() <-chan struct{}

GossipReady is a chan which is closed when we established initial min connections with peers.

func (*Switch) LocalNode

func (s *Switch) LocalNode() node.LocalNode

LocalNode is the local p2p identity.

func (*Switch) ProcessDirectProtocolMessage

func (s *Switch) ProcessDirectProtocolMessage(ctx context.Context, sender p2pcrypto.PublicKey, protocol string, data service.Data, metadata service.P2PMetadata) error

ProcessDirectProtocolMessage passes an already decrypted message to a protocol. if protocol does not exist, return and error.

func (*Switch) ProcessGossipProtocolMessage

func (s *Switch) ProcessGossipProtocolMessage(ctx context.Context, sender p2pcrypto.PublicKey, ownMessage bool, protocol string, data service.Data, validationCompletedChan chan service.MessageValidation) error

ProcessGossipProtocolMessage passes an already decrypted message to a protocol. It is expected that the protocol will send the message syntactic validation result on the validationCompletedChan ASAP

func (*Switch) RegisterDirectProtocol

func (s *Switch) RegisterDirectProtocol(protocol string) chan service.DirectMessage

RegisterDirectProtocol registers an handler for a direct messaging based protocol.

func (*Switch) RegisterDirectProtocolWithChannel

func (s *Switch) RegisterDirectProtocolWithChannel(protocol string, ingressChannel chan service.DirectMessage) chan service.DirectMessage

RegisterDirectProtocolWithChannel registers a direct protocol with a given channel. NOTE: eventually should replace RegisterDirectProtocol

func (*Switch) RegisterGossipProtocol

func (s *Switch) RegisterGossipProtocol(protocol string, prio priorityq.Priority) chan service.GossipMessage

RegisterGossipProtocol registers an handler for a gossip based protocol. priority must be provided.

func (*Switch) SendMessage

func (s *Switch) SendMessage(ctx context.Context, peerPubkey p2pcrypto.PublicKey, protocol string, payload []byte) error

SendMessage sends a p2p message to a peer using its public key. The provided public key must belong to one of our connected neighbors, otherwise an error will be returned.

func (*Switch) SendWrappedMessage

func (s *Switch) SendWrappedMessage(ctx context.Context, nodeID p2pcrypto.PublicKey, protocol string, payload *service.DataMsgWrapper) error

SendWrappedMessage sends a wrapped message in order to differentiate between request response and sub protocol messages. It is used by `MessageServer`.

func (*Switch) Shutdown

func (s *Switch) Shutdown()

Shutdown sends a shutdown signal to all running services of Switch and then runs an internal shutdown to cleanup.

func (*Switch) Start

func (s *Switch) Start(ctx context.Context) error

Start starts the p2p service. if configured, bootstrap is started in the background. returns error if the Switch is already running or there was an error starting one of the needed services.

func (*Switch) SubscribePeerEvents

func (s *Switch) SubscribePeerEvents() (conn, disc chan p2pcrypto.PublicKey)

SubscribePeerEvents lets clients listen on events inside the Switch about peers. first chan is new peers, second is deleted peers.

type UDPMux

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

UDPMux is a server for receiving and sending udp messages through protocols.

func NewUDPMux

func NewUDPMux(ctx, shutdownCtx context.Context, localNode node.LocalNode, lookuper Lookuper, udpNet udpNetwork, networkid uint32, logger log.Log) *UDPMux

NewUDPMux creates a new udp protocol server

func (*UDPMux) ProcessDirectProtocolMessage

func (mux *UDPMux) ProcessDirectProtocolMessage(sender p2pcrypto.PublicKey, protocol string, data service.Data, metadata service.P2PMetadata) error

ProcessDirectProtocolMessage passes a message to the protocol.

func (*UDPMux) RegisterDirectProtocolWithChannel

func (mux *UDPMux) RegisterDirectProtocolWithChannel(name string, c chan service.DirectMessage) chan service.DirectMessage

RegisterDirectProtocolWithChannel registers a protocol on a channel, should be done before `Start` was called. not thread-safe

func (*UDPMux) SendMessage

func (mux *UDPMux) SendMessage(ctx context.Context, peerPubkey p2pcrypto.PublicKey, protocol string, payload []byte) error

SendMessage is a proxy method to the sendMessageImpl.

func (*UDPMux) SendWrappedMessage

func (mux *UDPMux) SendWrappedMessage(ctx context.Context, nodeID p2pcrypto.PublicKey, protocol string, payload *service.DataMsgWrapper) error

SendWrappedMessage is a proxy method to the sendMessageImpl. it sends a wrapped message and used within MessageServer

func (*UDPMux) Shutdown

func (mux *UDPMux) Shutdown()

Shutdown closes the server

func (*UDPMux) Start

func (mux *UDPMux) Start() error

Start starts the UDPMux