peers

README

Peer pool signals

Peer pool sends 3 types of signals.

Discovery started signal will be sent once discovery server is started. And every time node will have to re-start discovery server because peer number dropped too low.

{
  "type": "discovery.started",
  "event": null
}

Discovery stopped signal will be sent once discovery found max limit of peers for every registered topic.

{
  "type": "discovery.stopped",
  "event": null
}

Discovery summary signal will be sent every time new peer is added or removed from a cluster. It will contain a map with capability as a key and total numbers of peers with that capability as a value.

{
  "type": "discovery.summary",
  "event": [
    {
      "id": "339c84c816b5f17a622c8d7ab9498f9998e942a274f70794af934bf5d3d02e14db8ddca2170e4edccede29ea6d409b154c141c34c01006e76c95e17672a27454",
      "name": "peer-0/v1.0/darwin/go1.10.1",
      "caps": [
        "shh/6"
      ],
      "network": {
        "localAddress": "127.0.0.1:61049",
        "remoteAddress": "127.0.0.1:33732",
        "inbound": false,
        "trusted": false,
        "static": true
      },
      "protocols": {
        "shh": "unknown"
      }
    }
  ]
}

Or if we don't have any peers:

{
  "type": "discovery.summary",
  "event": []
}

Documentation

Index

• Constants
• Variables
• func SendDiscoverySummary(peers []*p2p.PeerInfo)
• type Cache
  • func NewCache(db *leveldb.DB) *Cache
  • func (d *Cache) AddPeer(peer *discv5.Node, topic discv5.Topic) error
  • func (d *Cache) GetPeersRange(topic discv5.Topic, limit int) (nodes []*discv5.Node)
  • func (d *Cache) RemovePeer(nodeID enode.ID, topic discv5.Topic) error
• type Options
  • func NewDefaultOptions() *Options
• type PeerPool
  • func NewPeerPool(discovery discovery.Discovery, config map[discv5.Topic]params.Limits, ...) *PeerPool
  • func (p *PeerPool) Start(server *p2p.Server, rpcClient contracts.RPCClient) error
  • func (p *PeerPool) Stop()
  • func (p *PeerPool) UpdateTopic(topic string, limits params.Limits) error
• type PoolEvent
• type Register
  • func NewRegister(discovery discovery.Discovery, topics ...discv5.Topic) *Register
  • func (r *Register) Start() error
  • func (r *Register) Stop()
• type TopicPool
  • func (t *TopicPool) AddPeerFromTable(server *p2p.Server) *discv5.Node
  • func (t *TopicPool) BelowMin() bool
  • func (t *TopicPool) ConfirmAdded(server *p2p.Server, nodeID enode.ID)
  • func (t *TopicPool) ConfirmDropped(server *p2p.Server, nodeID enode.ID) bool
  • func (t *TopicPool) MaxReached() bool
  • func (t *TopicPool) SearchRunning() bool
  • func (t *TopicPool) SetLimits(limits params.Limits)
  • func (t *TopicPool) StartSearch(server *p2p.Server) error
  • func (t *TopicPool) StopSearch(server *p2p.Server)
  • func (t *TopicPool) Topic() discv5.Topic
• type TopicPoolInterface
• type Verifier

Constants

const (

	// DefaultFastSync is a recommended value for aggressive peers search.
	DefaultFastSync = 3 * time.Second
	// DefaultSlowSync is a recommended value for slow (background) peers search.
	DefaultSlowSync = 30 * time.Second
	// DefaultDiscV5Timeout is a timeout after which Discv5 is stopped.
	DefaultDiscV5Timeout = 3 * time.Minute
	// DefaultTopicFastModeTimeout is a timeout after which sync mode is switched to slow mode.
	DefaultTopicFastModeTimeout = 30 * time.Second
	// DefaultTopicStopSearchDelay is the default delay when stopping a topic search.
	DefaultTopicStopSearchDelay = 10 * time.Second
)

const MailServerDiscoveryTopic = "whispermail"

MailServerDiscoveryTopic topic name for mailserver discovery.

Variables

var (
	// ErrDiscv5NotRunning returned when pool is started but discover v5 is not running or not enabled.
	ErrDiscv5NotRunning = errors.New("Discovery v5 is not running")
)

var MailServerDiscoveryLimits = params.Limits{Min: 3, Max: 3}

MailServerDiscoveryLimits default mailserver discovery limits.

Functions

func SendDiscoverySummary

func SendDiscoverySummary(peers []*p2p.PeerInfo)

SendDiscoverySummary sends discovery.summary signal.

Types

type Cache

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

Cache maintains list of peers that were discovered.

func NewCache

func NewCache(db *leveldb.DB) *Cache

NewCache returns instance of PeersDatabase

func (*Cache) AddPeer

func (d *Cache) AddPeer(peer *discv5.Node, topic discv5.Topic) error

AddPeer stores peer with a following key: <topic><peer ID>

func (*Cache) GetPeersRange

func (d *Cache) GetPeersRange(topic discv5.Topic, limit int) (nodes []*discv5.Node)

GetPeersRange returns peers for a given topic with a limit.

func (*Cache) RemovePeer

func (d *Cache) RemovePeer(nodeID enode.ID, topic discv5.Topic) error

RemovePeer deletes a peer from database.

type Options

type Options struct {
	FastSync time.Duration
	SlowSync time.Duration
	// After this time, Discovery is stopped even if max peers is not reached.
	DiscServerTimeout time.Duration
	// AllowStop allows stopping Discovery when reaching max peers or after timeout.
	AllowStop bool
	// TopicStopSearchDelay time stopSearch will be waiting for max cached peers to be
	// filled before really stopping the search.
	TopicStopSearchDelay time.Duration
	// TrustedMailServers is a list of trusted nodes.
	TrustedMailServers []enode.ID
	// MailServerRegistryAddress is the MailServerRegistry contract address
	MailServerRegistryAddress string
}

Options is a struct with PeerPool configuration.

func NewDefaultOptions

func NewDefaultOptions() *Options

NewDefaultOptions returns a struct with default Options.

type PeerPool

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

PeerPool manages discovered peers and connects them to p2p server

func NewPeerPool

func NewPeerPool(discovery discovery.Discovery, config map[discv5.Topic]params.Limits, cache *Cache, options *Options) *PeerPool

NewPeerPool creates instance of PeerPool

func (*PeerPool) Start

func (p *PeerPool) Start(server *p2p.Server, rpcClient contracts.RPCClient) error

Start creates topic pool for each topic in config and subscribes to server events.

func (*PeerPool) Stop

func (p *PeerPool) Stop()

Stop closes pool quit channel and all channels that are watched by search queries and waits till all goroutines will exit.

func (*PeerPool) UpdateTopic

func (p *PeerPool) UpdateTopic(topic string, limits params.Limits) error

UpdateTopic updates the pre-existing TopicPool limits.

type PoolEvent

type PoolEvent string

PoolEvent is a type used to for peer pool events.

type Register

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

Register manages register topic queries

func NewRegister

func NewRegister(discovery discovery.Discovery, topics ...discv5.Topic) *Register

NewRegister creates instance of topic register

func (*Register) Start

func (r *Register) Start() error

Start topic register query for every topic

func (*Register) Stop

func (r *Register) Stop()

Stop all register topic queries and waits for them to exit

type TopicPool

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

TopicPool manages peers for topic.

func (*TopicPool) AddPeerFromTable

func (t *TopicPool) AddPeerFromTable(server *p2p.Server) *discv5.Node

AddPeerFromTable checks if there is a valid peer in local table and adds it to a server.

func (*TopicPool) BelowMin

func (t *TopicPool) BelowMin() bool

BelowMin returns true if current number of peers is below min limit.

func (*TopicPool) ConfirmAdded

func (t *TopicPool) ConfirmAdded(server *p2p.Server, nodeID enode.ID)

ConfirmAdded called when peer was added by p2p Server.

1. Skip a peer if it not in our peer table
2. Add a peer to a cache.
3. Disconnect a peer if it was connected after we reached max limit of peers. (we can't know in advance if peer will be connected, thats why we allow to overflow for short duration)
4. Switch search to slow mode if it is running.

func (*TopicPool) ConfirmDropped

func (t *TopicPool) ConfirmDropped(server *p2p.Server, nodeID enode.ID) bool

ConfirmDropped called when server receives drop event. 1. Skip peer if it is not in our peer table. 2. If disconnect request - we could drop that peer ourselves. 3. If connected number will drop below min limit - switch to fast mode. 4. Delete a peer from cache and peer table. Returns false if peer is not in our table or we requested removal of this peer. Otherwise peer is removed and true is returned.

func (*TopicPool) MaxReached

func (t *TopicPool) MaxReached() bool

MaxReached returns true if we connected with max number of peers.

func (*TopicPool) SearchRunning

func (t *TopicPool) SearchRunning() bool

SearchRunning returns true if search is running

func (*TopicPool) SetLimits

func (t *TopicPool) SetLimits(limits params.Limits)

SetLimits set the limits for the current TopicPool.

func (*TopicPool) StartSearch

func (t *TopicPool) StartSearch(server *p2p.Server) error

StartSearch creates discv5 queries and runs a loop to consume found peers.

func (*TopicPool) StopSearch

func (t *TopicPool) StopSearch(server *p2p.Server)

StopSearch stops the closes stop

func (*TopicPool) Topic

func (t *TopicPool) Topic() discv5.Topic

Topic exposes the internal discovery topic.

type TopicPoolInterface

type TopicPoolInterface interface {
	StopSearch(server *p2p.Server)
	BelowMin() bool
	SearchRunning() bool
	StartSearch(server *p2p.Server) error
	ConfirmDropped(server *p2p.Server, nodeID enode.ID) bool
	AddPeerFromTable(server *p2p.Server) *discv5.Node
	MaxReached() bool
	ConfirmAdded(server *p2p.Server, nodeID enode.ID)

	Topic() discv5.Topic
	SetLimits(limits params.Limits)
	// contains filtered or unexported methods
}

TopicPoolInterface the TopicPool interface.

type Verifier

type Verifier interface {
	VerifyNode(context.Context, enode.ID) bool
}

Verifier verifies if a give node is trusted.