diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go
index 26aa3e50ff..4772ad25a1 100644
--- a/cmd/swarm/main.go
+++ b/cmd/swarm/main.go
@@ -440,7 +440,7 @@ func getPassPhrase(prompt string, i int, passwords []string) string {
return password
}
-func injectBootnodes(srv *p2p.Server, nodes []string) {
+func injectBootnodes(srv p2p.Server, nodes []string) {
for _, url := range nodes {
n, err := discover.ParseNode(url)
if err != nil {
diff --git a/cmd/wnode/main.go b/cmd/wnode/main.go
index f18025dff8..5d5aa77ee8 100644
--- a/cmd/wnode/main.go
+++ b/cmd/wnode/main.go
@@ -51,7 +51,7 @@ const quitCommand = "~Q"
// singletons
var (
- server *p2p.Server
+ server p2p.Server
shh *whisper.Whisper
done chan struct{}
mailServer mailserver.WMailServer
@@ -253,19 +253,17 @@ func initialize() {
maxPeers = 800
}
- server = &p2p.Server{
- Config: p2p.Config{
- PrivateKey: nodeid,
- MaxPeers: maxPeers,
- Name: common.MakeName("wnode", "5.0"),
- Protocols: shh.Protocols(),
- ListenAddr: *argIP,
- NAT: nat.Any(),
- BootstrapNodes: peers,
- StaticNodes: peers,
- TrustedNodes: peers,
- },
- }
+ server = p2p.NewServer(p2p.Config{
+ PrivateKey: nodeid,
+ MaxPeers: maxPeers,
+ Name: common.MakeName("wnode", "5.0"),
+ Protocols: shh.Protocols(),
+ ListenAddr: *argIP,
+ NAT: nat.Any(),
+ BootstrapNodes: peers,
+ StaticNodes: peers,
+ TrustedNodes: peers,
+ })
}
func startServer() {
diff --git a/contracts/release/release.go b/contracts/release/release.go
index 28a35381d4..e0441fd41e 100644
--- a/contracts/release/release.go
+++ b/contracts/release/release.go
@@ -94,7 +94,7 @@ func (r *ReleaseService) Protocols() []p2p.Protocol { return nil }
func (r *ReleaseService) APIs() []rpc.API { return nil }
// Start spawns the periodic version checker goroutine
-func (r *ReleaseService) Start(server *p2p.Server) error {
+func (r *ReleaseService) Start(server p2p.Server) error {
go r.checker()
return nil
}
diff --git a/eth/backend.go b/eth/backend.go
index f864b1d88b..9361cb3064 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -49,7 +49,7 @@ import (
)
type LesServer interface {
- Start(srvr *p2p.Server)
+ Start(srvr p2p.Server)
Stop()
Protocols() []p2p.Protocol
}
@@ -362,7 +362,7 @@ func (s *Ethereum) Protocols() []p2p.Protocol {
// Start implements node.Service, starting all internal goroutines needed by the
// Ethereum protocol implementation.
-func (s *Ethereum) Start(srvr *p2p.Server) error {
+func (s *Ethereum) Start(srvr p2p.Server) error {
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.NetVersion())
s.protocolManager.Start()
diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go
index 8765da8faf..f295fb60d0 100644
--- a/ethstats/ethstats.go
+++ b/ethstats/ethstats.go
@@ -52,7 +52,7 @@ const historyUpdateRange = 50
type Service struct {
stack *node.Node // Temporary workaround, remove when API finalized
- server *p2p.Server // Peer-to-peer server to retrieve networking infos
+ server p2p.Server // Peer-to-peer server to retrieve networking infos
eth *eth.Ethereum // Full Ethereum service if monitoring a full node
les *les.LightEthereum // Light Ethereum service if monitoring a light node
engine consensus.Engine // Consensus engine to retrieve variadic block fields
@@ -101,7 +101,7 @@ func (s *Service) Protocols() []p2p.Protocol { return nil }
func (s *Service) APIs() []rpc.API { return nil }
// Start implements node.Service, starting up the monitoring and reporting daemon.
-func (s *Service) Start(server *p2p.Server) error {
+func (s *Service) Start(server p2p.Server) error {
s.server = server
go s.loop()
diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go
index f9eed87975..146e7f0475 100644
--- a/internal/ethapi/api.go
+++ b/internal/ethapi/api.go
@@ -1434,12 +1434,12 @@ func (api *PrivateDebugAPI) SetHead(number hexutil.Uint64) {
// PublicNetAPI offers network related RPC methods
type PublicNetAPI struct {
- net *p2p.Server
+ net p2p.Server
networkVersion uint64
}
// NewPublicNetAPI creates a new net API instance.
-func NewPublicNetAPI(net *p2p.Server, networkVersion uint64) *PublicNetAPI {
+func NewPublicNetAPI(net p2p.Server, networkVersion uint64) *PublicNetAPI {
return &PublicNetAPI{net, networkVersion}
}
diff --git a/les/backend.go b/les/backend.go
index 646c81a7b1..3dac7f15fc 100644
--- a/les/backend.go
+++ b/les/backend.go
@@ -185,7 +185,7 @@ func (s *LightEthereum) Protocols() []p2p.Protocol {
// Start implements node.Service, starting all internal goroutines needed by the
// Ethereum protocol implementation.
-func (s *LightEthereum) Start(srvr *p2p.Server) error {
+func (s *LightEthereum) Start(srvr p2p.Server) error {
log.Warn("Light client mode is an experimental feature")
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId)
s.protocolManager.Start(srvr)
diff --git a/les/handler.go b/les/handler.go
index 64023af0f5..5af89d4fca 100644
--- a/les/handler.go
+++ b/les/handler.go
@@ -61,6 +61,10 @@ const (
disableClientRemovePeer = false
)
+type discV5Server interface {
+ DiscV5() *discv5.Network
+}
+
// errIncompatibleConfig is returned if the requested protocols and configs are
// not compatible (low protocol version restrictions and high requirements).
var errIncompatibleConfig = errors.New("incompatible configuration")
@@ -256,10 +260,10 @@ func (pm *ProtocolManager) removePeer(id string) {
}
}
-func (pm *ProtocolManager) Start(srvr *p2p.Server) {
+func (pm *ProtocolManager) Start(srvr p2p.Server) {
var topicDisc *discv5.Network
- if srvr != nil {
- topicDisc = srvr.DiscV5
+ if v, ok := srvr.(discV5Server); ok {
+ topicDisc = v.DiscV5()
}
lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8]))
if pm.lightSync {
diff --git a/les/server.go b/les/server.go
index 22fe59b7ac..1be61ef53a 100644
--- a/les/server.go
+++ b/les/server.go
@@ -68,7 +68,7 @@ func (s *LesServer) Protocols() []p2p.Protocol {
}
// Start starts the LES server
-func (s *LesServer) Start(srvr *p2p.Server) {
+func (s *LesServer) Start(srvr p2p.Server) {
s.protocolManager.Start(srvr)
}
diff --git a/les/serverpool.go b/les/serverpool.go
index 64fe991c63..14a7354261 100644
--- a/les/serverpool.go
+++ b/les/serverpool.go
@@ -97,7 +97,7 @@ const (
type serverPool struct {
db ethdb.Database
dbKey []byte
- server *p2p.Server
+ server p2p.Server
quit chan struct{}
wg *sync.WaitGroup
connWg sync.WaitGroup
@@ -118,7 +118,7 @@ type serverPool struct {
}
// newServerPool creates a new serverPool instance
-func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic discv5.Topic, quit chan struct{}, wg *sync.WaitGroup) *serverPool {
+func newServerPool(db ethdb.Database, dbPrefix []byte, server p2p.Server, topic discv5.Topic, quit chan struct{}, wg *sync.WaitGroup) *serverPool {
pool := &serverPool{
db: db,
dbKey: append(dbPrefix, []byte(topic)...),
@@ -139,11 +139,11 @@ func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic
pool.loadNodes()
pool.checkDial()
- if pool.server.DiscV5 != nil {
+ if srv, ok := pool.server.(discV5Server); ok && srv.DiscV5() != nil {
pool.discSetPeriod = make(chan time.Duration, 1)
pool.discNodes = make(chan *discv5.Node, 100)
pool.discLookups = make(chan bool, 100)
- go pool.server.DiscV5.SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups)
+ go srv.DiscV5().SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups)
}
go pool.eventLoop()
diff --git a/node/api.go b/node/api.go
index 570cb9d98e..70aa59d844 100644
--- a/node/api.go
+++ b/node/api.go
@@ -335,7 +335,7 @@ func NewPublicWeb3API(stack *Node) *PublicWeb3API {
// ClientVersion returns the node name
func (s *PublicWeb3API) ClientVersion() string {
- return s.stack.Server().Name
+ return s.stack.serverConfig.Name
}
// Sha3 applies the ethereum sha3 implementation on the input.
diff --git a/node/node.go b/node/node.go
index a89110599f..2c5733cf3e 100644
--- a/node/node.go
+++ b/node/node.go
@@ -56,7 +56,7 @@ type Node struct {
instanceDirLock storage.Storage // prevents concurrent use of instance directory
serverConfig p2p.Config
- server *p2p.Server // Currently running P2P networking layer
+ server p2p.Server // Currently running P2P networking layer
serviceFuncs []ServiceConstructor // Service constructors (in dependency order)
services map[reflect.Type]Service // Currently running services
@@ -165,7 +165,7 @@ func (n *Node) Start() error {
if n.serverConfig.NodeDatabase == "" {
n.serverConfig.NodeDatabase = n.config.NodeDB()
}
- running := &p2p.Server{Config: n.serverConfig}
+ running := p2p.NewServer(n.serverConfig)
log.Info("Starting peer-to-peer node", "instance", n.serverConfig.Name)
// Otherwise copy and specialize the P2P configuration
@@ -194,7 +194,7 @@ func (n *Node) Start() error {
}
// Gather the protocols and start the freshly assembled P2P server
for _, service := range services {
- running.Protocols = append(running.Protocols, service.Protocols()...)
+ n.serverConfig.Protocols = append(n.serverConfig.Protocols, service.Protocols()...)
}
if err := running.Start(); err != nil {
if errno, ok := err.(syscall.Errno); ok && datadirInUseErrnos[uint(errno)] {
@@ -582,7 +582,7 @@ func (n *Node) RPCHandler() (*rpc.Server, error) {
// Server retrieves the currently running P2P network layer. This method is meant
// only to inspect fields of the currently running server, life cycle management
// should be left to this Node entity.
-func (n *Node) Server() *p2p.Server {
+func (n *Node) Server() p2p.Server {
n.lock.RLock()
defer n.lock.RUnlock()
diff --git a/node/node_example_test.go b/node/node_example_test.go
index ee06f4065c..ddf76faea7 100644
--- a/node/node_example_test.go
+++ b/node/node_example_test.go
@@ -37,7 +37,7 @@ type SampleService struct{}
func (s *SampleService) Protocols() []p2p.Protocol { return nil }
func (s *SampleService) APIs() []rpc.API { return nil }
-func (s *SampleService) Start(*p2p.Server) error { fmt.Println("Service starting..."); return nil }
+func (s *SampleService) Start(p2p.Server) error { fmt.Println("Service starting..."); return nil }
func (s *SampleService) Stop() error { fmt.Println("Service stopping..."); return nil }
func ExampleService() {
diff --git a/node/node_test.go b/node/node_test.go
index 2880efa619..a0f13ca570 100644
--- a/node/node_test.go
+++ b/node/node_test.go
@@ -154,7 +154,7 @@ func TestServiceLifeCycle(t *testing.T) {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
- startHook: func(*p2p.Server) { started[id] = true },
+ startHook: func(p2p.Server) { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
@@ -200,7 +200,7 @@ func TestServiceRestarts(t *testing.T) {
running = false
return &InstrumentedService{
- startHook: func(*p2p.Server) {
+ startHook: func(p2p.Server) {
if running {
panic("already running")
}
@@ -250,7 +250,7 @@ func TestServiceConstructionAbortion(t *testing.T) {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
- startHook: func(*p2p.Server) { started[id] = true },
+ startHook: func(p2p.Server) { started[id] = true },
}, nil
}
if err := stack.Register(maker(constructor)); err != nil {
@@ -299,7 +299,7 @@ func TestServiceStartupAbortion(t *testing.T) {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
- startHook: func(*p2p.Server) { started[id] = true },
+ startHook: func(p2p.Server) { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
@@ -352,7 +352,7 @@ func TestServiceTerminationGuarantee(t *testing.T) {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
- startHook: func(*p2p.Server) { started[id] = true },
+ startHook: func(p2p.Server) { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
diff --git a/node/service.go b/node/service.go
index 5e1eb0e645..05320b75ab 100644
--- a/node/service.go
+++ b/node/service.go
@@ -86,7 +86,7 @@ type Service interface {
// Start is called after all services have been constructed and the networking
// layer was also initialized to spawn any goroutines required by the service.
- Start(server *p2p.Server) error
+ Start(server p2p.Server) error
// Stop terminates all goroutines belonging to the service, blocking until they
// are all terminated.
diff --git a/node/utils_test.go b/node/utils_test.go
index 7cdfc2b3aa..e9c999b93c 100644
--- a/node/utils_test.go
+++ b/node/utils_test.go
@@ -31,7 +31,7 @@ type NoopService struct{}
func (s *NoopService) Protocols() []p2p.Protocol { return nil }
func (s *NoopService) APIs() []rpc.API { return nil }
-func (s *NoopService) Start(*p2p.Server) error { return nil }
+func (s *NoopService) Start(p2p.Server) error { return nil }
func (s *NoopService) Stop() error { return nil }
func NewNoopService(*ServiceContext) (Service, error) { return new(NoopService), nil }
@@ -57,7 +57,7 @@ type InstrumentedService struct {
stop error
protocolsHook func()
- startHook func(*p2p.Server)
+ startHook func(p2p.Server)
stopHook func()
}
@@ -74,7 +74,7 @@ func (s *InstrumentedService) APIs() []rpc.API {
return s.apis
}
-func (s *InstrumentedService) Start(server *p2p.Server) error {
+func (s *InstrumentedService) Start(server p2p.Server) error {
if s.startHook != nil {
s.startHook(server)
}
diff --git a/p2p/adapters/exec.go b/p2p/adapters/exec.go
index 5da7ec9488..0348418095 100644
--- a/p2p/adapters/exec.go
+++ b/p2p/adapters/exec.go
@@ -22,9 +22,8 @@ import (
"github.com/ethereum/go-ethereum/rpc"
)
-// serviceFunc returns a node.ServiceConstructor which can be used to boot
-// devp2p nodes
-type serviceFunc func(id *NodeId) node.ServiceConstructor
+// serviceFunc returns a node.Service which can be used to boot devp2p nodes
+type serviceFunc func(id *NodeId) node.Service
// serviceFuncs is a map of registered services which are used to boot devp2p
// nodes
@@ -52,9 +51,9 @@ type ExecNode struct {
Dir string
Config *node.Config
Cmd *exec.Cmd
- Client *rpc.Client
Info *p2p.NodeInfo
+ client *rpc.Client
newCmd func() *exec.Cmd
}
@@ -96,6 +95,10 @@ func (n *ExecNode) Addr() []byte {
return []byte(n.Info.Enode)
}
+func (n *ExecNode) Client() (*rpc.Client, error) {
+ return n.client, nil
+}
+
// Start exec's the node passing the ID and service as command line arguments
// and the node config encoded as JSON in the _P2P_NODE_CONFIG environment
// variable
@@ -131,11 +134,11 @@ func (n *ExecNode) Start() (err error) {
n.Cmd = cmd
// create the RPC client and load the node info
- n.Client = rpc.NewClientWithConn(pipe2)
+ n.client = rpc.NewClientWithConn(pipe2)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
var info p2p.NodeInfo
- if err := n.Client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil {
+ if err := n.client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil {
return fmt.Errorf("error getting node info: %s", err)
}
n.Info = &info
@@ -163,9 +166,9 @@ func (n *ExecNode) Stop() error {
n.Cmd = nil
}()
- if n.Client != nil {
- n.Client.Close()
- n.Client = nil
+ if n.client != nil {
+ n.client.Close()
+ n.client = nil
n.Info = nil
}
@@ -184,24 +187,6 @@ func (n *ExecNode) Stop() error {
}
}
-// Connect connects the node to the given addr by calling the Admin.AddPeer
-// IPC method
-func (n *ExecNode) Connect(addr []byte) error {
- if n.Client == nil {
- return errors.New("node not started")
- }
- return n.Client.Call(nil, "admin_addPeer", string(addr))
-}
-
-// Disconnect disconnects the node from the given addr by calling the
-// Admin.RemovePeer IPC method
-func (n *ExecNode) Disconnect(addr []byte) error {
- if n.Client == nil {
- return errors.New("node not started")
- }
- return n.Client.Call(nil, "admin_removePeer", string(addr))
-}
-
func init() {
// register a reexec function to start a devp2p node when the current
// binary is executed as "p2p-node"
@@ -230,11 +215,12 @@ func execP2PNode() {
log.Crit("error decoding _P2P_NODE_CONFIG", "err", err)
}
- // lookup the service constructor
- service, exists := serviceFuncs[serviceName]
+ // initialize the service
+ serviceFunc, exists := serviceFuncs[serviceName]
if !exists {
log.Crit(fmt.Sprintf("unknown node service %q", serviceName))
}
+ service := serviceFunc(id)
// use explicit IP address in ListenAddr so that Enode URL is usable
if strings.HasPrefix(conf.P2P.ListenAddr, ":") {
@@ -251,15 +237,9 @@ func execP2PNode() {
}
// start the devp2p stack
- stack, err := node.New(&conf)
+ stack, err := startP2PNode(&conf, service)
if err != nil {
- log.Crit("error creating node", "err", err)
- }
- if err := stack.Register(service(id)); err != nil {
- log.Crit("error registering service", "err", err)
- }
- if err := stack.Start(); err != nil {
- log.Crit("error starting node", "err", err)
+ log.Crit("error starting p2p node", "err", err)
}
// use stdin / stdout for RPC to avoid the parent needing to access
@@ -283,6 +263,23 @@ func execP2PNode() {
stack.Wait()
}
+func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) {
+ stack, err := node.New(conf)
+ if err != nil {
+ return nil, err
+ }
+ constructor := func(ctx *node.ServiceContext) (node.Service, error) {
+ return service, nil
+ }
+ if err := stack.Register(constructor); err != nil {
+ return nil, err
+ }
+ if err := stack.Start(); err != nil {
+ return nil, err
+ }
+ return stack, nil
+}
+
// stdioConn wraps os.Stdin / os.Stdout with a nop Close method so we can
// use them to handle RPC messages
type stdioConn struct {
diff --git a/p2p/adapters/inproc.go b/p2p/adapters/inproc.go
index 9877607638..7c96cf21ca 100644
--- a/p2p/adapters/inproc.go
+++ b/p2p/adapters/inproc.go
@@ -17,12 +17,15 @@
package adapters
import (
+ "errors"
"fmt"
"sync"
"github.com/ethereum/go-ethereum/log"
+ "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/rpc"
)
func newPeer(rw *p2p.MsgPipeRW) *Peer {
@@ -53,29 +56,94 @@ type SimNode struct {
lock sync.RWMutex
Id *NodeId
network Network
+ service node.Service
peerMap map[discover.NodeID]int
peers []*Peer
- Run ProtoCall
+ client *rpc.Client
}
-func NewSimNode(id *NodeId, n Network) *SimNode {
+func NewSimNode(id *NodeId, svc node.Service, n Network) *SimNode {
+ // for simplicity, only support single protocol services
+ if len(svc.Protocols()) != 1 {
+ panic("service must have a single protocol")
+ }
+
return &SimNode{
Id: id,
network: n,
+ service: svc,
peerMap: make(map[discover.NodeID]int),
}
}
+// Addr returns the node's address
func (self *SimNode) Addr() []byte {
- return self.Id.Bytes()
+ return []byte(self.Node().String())
}
+func (self *SimNode) Node() *discover.Node {
+ return discover.NewNode(self.Id.NodeID, nil, 0, 0)
+}
+
+func (self *SimNode) Client() (*rpc.Client, error) {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ if self.client == nil {
+ return nil, errors.New("RPC not started")
+ }
+ return self.client, nil
+}
+
+// Start starts the RPC handler and the underlying service
func (self *SimNode) Start() error {
+ if err := self.startRPC(); err != nil {
+ return err
+ }
+ return self.service.Start(self)
+}
+
+// Stop stops the RPC handler and the underlying service
+func (self *SimNode) Stop() error {
+ self.stopRPC()
+ return self.service.Stop()
+}
+
+func (self *SimNode) startRPC() error {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ if self.client != nil {
+ return errors.New("RPC already started")
+ }
+
+ // add SimAdminAPI so that the network can call the AddPeer
+ // and RemovePeer RPC methods
+ apis := append(self.service.APIs(), rpc.API{
+ Namespace: "admin",
+ Version: "1.0",
+ Service: &SimAdminAPI{self},
+ })
+
+ // start the RPC handler
+ handler := rpc.NewServer()
+ for _, api := range apis {
+ if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
+ return fmt.Errorf("error registering RPC: %s", err)
+ }
+ }
+
+ // create an in-process RPC client
+ self.client = rpc.DialInProc(handler)
+
return nil
}
-func (self *SimNode) Stop() error {
- return nil
+func (self *SimNode) stopRPC() {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ if self.client != nil {
+ self.client.Close()
+ self.client = nil
+ }
}
func (self *SimNode) GetPeer(id *NodeId) *Peer {
@@ -112,13 +180,13 @@ func (self *SimNode) setPeer(id *NodeId, rw *p2p.MsgPipeRW) *Peer {
return p
}
-func (self *SimNode) Disconnect(rid []byte) error {
+func (self *SimNode) RemovePeer(node *discover.Node) {
self.lock.Lock()
defer self.lock.Unlock()
- id := NewNodeId(rid)
+ id := &NodeId{node.ID}
peer := self.getPeer(id)
if peer == nil || peer.MsgPipeRW == nil {
- return fmt.Errorf("already disconnected")
+ return
}
peer.MsgPipeRW.Close()
peer.MsgPipeRW = nil
@@ -126,55 +194,85 @@ func (self *SimNode) Disconnect(rid []byte) error {
// peer = na.(*SimNode).GetPeer(self.Id)
// peer.RW = nil
log.Trace(fmt.Sprintf("dropped peer %v", id))
-
- return nil
}
-func (self *SimNode) Connect(rid []byte) error {
+func (self *SimNode) AddPeer(node *discover.Node) {
self.lock.Lock()
defer self.lock.Unlock()
- id := NewNodeId(rid)
+ id := &NodeId{node.ID}
na := self.network.GetNodeAdapter(id)
if na == nil {
- return fmt.Errorf("node adapter for %v is missing", id)
+ panic(fmt.Sprintf("node adapter for %v is missing", id))
}
rw, rrw := p2p.MsgPipe()
// // run protocol on remote node with self as peer
peer := self.getPeer(id)
if peer != nil && peer.MsgPipeRW != nil {
- return fmt.Errorf("already connected %v to peer %v", self.Id, id)
+ return
}
peer = self.setPeer(id, rrw)
close(peer.Connc)
defer close(peer.Readyc)
- err := na.(ProtocolRunner).RunProtocol(self.Id, rrw, rw, peer)
- if err != nil {
- return fmt.Errorf("cannot run protocol (%v -> %v) %v", self.Id, id, err)
- }
+ na.(*SimNode).RunProtocol(self, rrw, rw, peer)
// run protocol on remote node with self as peer
- err = self.RunProtocol(id, rw, rrw, peer)
- if err != nil {
- return fmt.Errorf("cannot run protocol (%v -> %v): %v", id, self.Id, err)
- }
+ self.RunProtocol(na.(*SimNode), rw, rrw, peer)
+}
+
+func (self *SimNode) PeerCount() int {
+ self.lock.Lock()
+ defer self.lock.Unlock()
+ return len(self.peers)
+}
+
+func (self *SimNode) NodeInfo() *p2p.NodeInfo {
+ return &p2p.NodeInfo{ID: self.Id.String()}
+}
+
+func (self *SimNode) PeersInfo() (info []*p2p.PeerInfo) {
return nil
}
-func (self *SimNode) RunProtocol(id *NodeId, rw, rrw p2p.MsgReadWriter, peer *Peer) error {
- if self.Run == nil {
+func (self *SimNode) RunProtocol(node *SimNode, rw, rrw p2p.MsgReadWriter, peer *Peer) {
+ id := node.Id
+ protocol := self.service.Protocols()[0]
+ if protocol.Run == nil {
log.Trace(fmt.Sprintf("no protocol starting on peer %v (connection with %v)", self.Id, id))
- return nil
+ return
}
log.Trace(fmt.Sprintf("protocol starting on peer %v (connection with %v)", self.Id, id))
p := p2p.NewPeer(id.NodeID, id.Label(), []p2p.Cap{})
go func() {
self.network.DidConnect(self.Id, id)
- err := self.Run(p, rw)
+ err := protocol.Run(p, rw)
<-peer.Readyc
- self.Disconnect(id.Bytes())
+ self.RemovePeer(node.Node())
peer.Errc <- err
log.Trace(fmt.Sprintf("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err))
self.network.DidDisconnect(self.Id, id)
}()
- return nil
+}
+
+// SimAdminAPI implements the AddPeer and RemovePeer RPC methods (API
+// compatible with node.PrivateAdminAPI)
+type SimAdminAPI struct {
+ *SimNode
+}
+
+func (api *SimAdminAPI) AddPeer(url string) (bool, error) {
+ node, err := discover.ParseNode(url)
+ if err != nil {
+ return false, fmt.Errorf("invalid enode: %v", err)
+ }
+ api.SimNode.AddPeer(node)
+ return true, nil
+}
+
+func (api *SimAdminAPI) RemovePeer(url string) (bool, error) {
+ node, err := discover.ParseNode(url)
+ if err != nil {
+ return false, fmt.Errorf("invalid enode: %v", err)
+ }
+ api.SimNode.RemovePeer(node)
+ return true, nil
}
diff --git a/p2p/adapters/rlpx.go b/p2p/adapters/rlpx.go
deleted file mode 100644
index b6a57bc519..0000000000
--- a/p2p/adapters/rlpx.go
+++ /dev/null
@@ -1,100 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package adapters
-
-import (
- "fmt"
- "net"
- //"encoding/binary"
-
- "github.com/ethereum/go-ethereum/p2p"
- "github.com/ethereum/go-ethereum/p2p/discover"
-)
-
-// devp2p RLPx underlay support
-
-type RLPx struct {
- id *NodeId
- net *p2p.Server
- addr []byte
- r Reporter
-}
-
-func NewRLPx(addr []byte, srv *p2p.Server) *RLPx {
- return &RLPx{
- net: srv,
- addr: addr,
- }
-}
-
-func NewReportingRLPx(addr []byte, srv *p2p.Server, r Reporter) *RLPx {
- rlpx := NewRLPx(addr, srv)
- rlpx.r = r
- srv.PeerConnHook = func(p *p2p.Peer) {
- r.DidConnect(rlpx.id, &NodeId{p.ID()})
- }
- srv.PeerDisconnHook = func(p *p2p.Peer) {
- r.DidDisconnect(rlpx.id, &NodeId{p.ID()})
- }
- return rlpx
-}
-
-func (self *RLPx) LocalAddr() []byte {
- return self.addr
-}
-
-func (self *RLPx) Connect(enode []byte) error {
- // TCP/UDP node address encoded with enode url scheme
- // @:(?udp=)
- node, err := discover.ParseNode(string(enode))
- if err != nil {
- return fmt.Errorf("invalid node URL: %v", err)
- }
- self.net.AddPeer(node)
- return nil
-}
-
-//func (self *RLPx) Disconnect(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-func (self *RLPx) Disconnect(b []byte) error {
- //p.Disconnect(p2p.DiscSubprotocolError)
- //d, _ := binary.Uvarint(b)
- //p.Disconnect(p2p.DiscReason(d))
- return nil
-}
-
-// ParseAddr take two arguments, advertised in handshake and the one set on the peer struct
-// and constructs the remote address object
-func (self *RLPx) ParseAddr(s []byte, remoteAddr string) ([]byte, error) {
-
- // returns self advertised node connection info (listening address w enodes)
- // IP will get repaired on the other end if missing
- // or resolved via ID by discovery at dialout
- n, err := discover.ParseNode(string(s))
- if err != nil {
- return nil, err
- }
-
- // repair reported address if IP missing
- if n.IP.IsUnspecified() {
- host, _, err := net.SplitHostPort(remoteAddr)
- if err != nil {
- return nil, err
- }
- n.IP = net.ParseIP(host)
- }
- return []byte(n.String()), nil
-}
diff --git a/p2p/adapters/types.go b/p2p/adapters/types.go
index 3618217528..47041f73b9 100644
--- a/p2p/adapters/types.go
+++ b/p2p/adapters/types.go
@@ -20,6 +20,7 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/rpc"
)
const lablen = 4
@@ -65,10 +66,9 @@ func (self *NodeId) Label() string {
type NodeAdapter interface {
Addr() []byte
+ Client() (*rpc.Client, error)
Start() error
Stop() error
- Connect(addr []byte) error
- Disconnect(addr []byte) error
}
type ProtocolRunner interface {
diff --git a/p2p/dial.go b/p2p/dial.go
index b779713963..b832fc233e 100644
--- a/p2p/dial.go
+++ b/p2p/dial.go
@@ -49,7 +49,7 @@ const (
// dialstate schedules dials and discovery lookups.
// it get's a chance to compute new tasks on every iteration
-// of the main loop in Server.run.
+// of the main loop in server.run.
type dialstate struct {
maxDynDials int
ntab discoverTable
@@ -84,7 +84,7 @@ type pastDial struct {
}
type task interface {
- Do(*Server)
+ Do(*server)
}
// A dialTask is generated for each node that is dialed. Its
@@ -104,7 +104,7 @@ type discoverTask struct {
}
// A waitExpireTask is generated if there are no other tasks
-// to keep the loop in Server.run ticking.
+// to keep the loop in server.run ticking.
type waitExpireTask struct {
time.Duration
}
@@ -267,7 +267,7 @@ func (s *dialstate) taskDone(t task, now time.Time) {
}
}
-func (t *dialTask) Do(srv *Server) {
+func (t *dialTask) Do(srv *server) {
if t.dest.Incomplete() {
if !t.resolve(srv) {
return
@@ -288,7 +288,7 @@ func (t *dialTask) Do(srv *Server) {
// Resolve operations are throttled with backoff to avoid flooding the
// discovery network with useless queries for nodes that don't exist.
// The backoff delay resets when the node is found.
-func (t *dialTask) resolve(srv *Server) bool {
+func (t *dialTask) resolve(srv *server) bool {
if srv.ntab == nil {
log.Debug("Can't resolve node", "id", t.dest.ID, "err", "discovery is disabled")
return false
@@ -317,7 +317,7 @@ func (t *dialTask) resolve(srv *Server) bool {
}
// dial performs the actual connection attempt.
-func (t *dialTask) dial(srv *Server, dest *discover.Node) bool {
+func (t *dialTask) dial(srv *server, dest *discover.Node) bool {
addr := &net.TCPAddr{IP: dest.IP, Port: int(dest.TCP)}
fd, err := srv.Dialer.Dial("tcp", addr.String())
if err != nil {
@@ -333,7 +333,7 @@ func (t *dialTask) String() string {
return fmt.Sprintf("%v %x %v:%d", t.flags, t.dest.ID[:8], t.dest.IP, t.dest.TCP)
}
-func (t *discoverTask) Do(srv *Server) {
+func (t *discoverTask) Do(srv *server) {
// newTasks generates a lookup task whenever dynamic dials are
// necessary. Lookups need to take some time, otherwise the
// event loop spins too fast.
@@ -355,7 +355,7 @@ func (t *discoverTask) String() string {
return s
}
-func (t waitExpireTask) Do(*Server) {
+func (t waitExpireTask) Do(*server) {
time.Sleep(t.Duration)
}
func (t waitExpireTask) String() string {
diff --git a/p2p/dial_test.go b/p2p/dial_test.go
index 08e863bae0..55c0154f80 100644
--- a/p2p/dial_test.go
+++ b/p2p/dial_test.go
@@ -598,7 +598,7 @@ func TestDialResolve(t *testing.T) {
// Now run the task, it should resolve the ID once.
config := Config{Dialer: &net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}}
- srv := &Server{ntab: table, Config: config}
+ srv := &server{ntab: table, Config: config}
tasks[0].Do(srv)
if !reflect.DeepEqual(table.resolveCalls, []discover.NodeID{dest.ID}) {
t.Fatalf("wrong resolve calls, got %v", table.resolveCalls)
diff --git a/p2p/server.go b/p2p/server.go
index 0411c76d66..4b3cf1557d 100644
--- a/p2p/server.go
+++ b/p2p/server.go
@@ -136,17 +136,29 @@ type Config struct {
NoDial bool `toml:",omitempty"`
}
+type Server interface {
+ Start() error
+ Stop() error
+ AddPeer(node *discover.Node)
+ RemovePeer(node *discover.Node)
+ PeerCount() int
+ NodeInfo() *NodeInfo
+ PeersInfo() []*PeerInfo
+}
+
+func NewServer(conf Config) Server {
+ return &server{Config: conf}
+}
+
// Server manages all peer connections.
-type Server struct {
+type server struct {
// Config fields may not be modified while the server is running.
Config
// Hooks for testing. These are useful because we can inhibit
// the whole protocol stack.
- newTransport func(net.Conn) transport
- newPeerHook func(*Peer)
- PeerConnHook func(*Peer)
- PeerDisconnHook func(*Peer)
+ newTransport func(net.Conn) transport
+ newPeerHook func(*Peer)
lock sync.Mutex // protects running
running bool
@@ -155,7 +167,7 @@ type Server struct {
listener net.Listener
ourHandshake *protoHandshake
lastLookup time.Time
- DiscV5 *discv5.Network
+ discV5 *discv5.Network
// These are for Peers, PeerCount (and nothing else).
peerOp chan peerOpFunc
@@ -247,7 +259,7 @@ func (c *conn) is(f connFlag) bool {
}
// Peers returns all connected peers.
-func (srv *Server) Peers() []*Peer {
+func (srv *server) Peers() []*Peer {
var ps []*Peer
select {
// Note: We'd love to put this function into a variable but
@@ -265,7 +277,7 @@ func (srv *Server) Peers() []*Peer {
}
// PeerCount returns the number of connected peers.
-func (srv *Server) PeerCount() int {
+func (srv *server) PeerCount() int {
var count int
select {
case srv.peerOp <- func(ps map[discover.NodeID]*Peer) { count = len(ps) }:
@@ -278,7 +290,7 @@ func (srv *Server) PeerCount() int {
// AddPeer connects to the given node and maintains the connection until the
// server is shut down. If the connection fails for any reason, the server will
// attempt to reconnect the peer.
-func (srv *Server) AddPeer(node *discover.Node) {
+func (srv *server) AddPeer(node *discover.Node) {
select {
case srv.addstatic <- node:
case <-srv.quit:
@@ -286,7 +298,7 @@ func (srv *Server) AddPeer(node *discover.Node) {
}
// RemovePeer disconnects from the given node
-func (srv *Server) RemovePeer(node *discover.Node) {
+func (srv *server) RemovePeer(node *discover.Node) {
select {
case srv.removestatic <- node:
case <-srv.quit:
@@ -294,7 +306,7 @@ func (srv *Server) RemovePeer(node *discover.Node) {
}
// Self returns the local node's endpoint information.
-func (srv *Server) Self() *discover.Node {
+func (srv *server) Self() *discover.Node {
srv.lock.Lock()
defer srv.lock.Unlock()
@@ -304,7 +316,7 @@ func (srv *Server) Self() *discover.Node {
return srv.makeSelf(srv.listener, srv.ntab)
}
-func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *discover.Node {
+func (srv *server) makeSelf(listener net.Listener, ntab discoverTable) *discover.Node {
// If the server's not running, return an empty node.
// If the node is running but discovery is off, manually assemble the node infos.
if ntab == nil {
@@ -326,11 +338,11 @@ func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *discover
// Stop terminates the server and all active peer connections.
// It blocks until all active connections have been closed.
-func (srv *Server) Stop() {
+func (srv *server) Stop() error {
srv.lock.Lock()
defer srv.lock.Unlock()
if !srv.running {
- return
+ return nil
}
srv.running = false
if srv.listener != nil {
@@ -339,11 +351,12 @@ func (srv *Server) Stop() {
}
close(srv.quit)
srv.loopWG.Wait()
+ return nil
}
// Start starts running the server.
// Servers can not be re-used after stopping.
-func (srv *Server) Start() (err error) {
+func (srv *server) Start() (err error) {
srv.lock.Lock()
defer srv.lock.Unlock()
if srv.running {
@@ -391,7 +404,7 @@ func (srv *Server) Start() (err error) {
if err := ntab.SetFallbackNodes(srv.BootstrapNodesV5); err != nil {
return err
}
- srv.DiscV5 = ntab
+ srv.discV5 = ntab
}
dynPeers := (srv.MaxPeers + 1) / 2
@@ -421,7 +434,7 @@ func (srv *Server) Start() (err error) {
return nil
}
-func (srv *Server) startListening() error {
+func (srv *server) startListening() error {
// Launch the TCP listener.
listener, err := net.Listen("tcp", srv.ListenAddr)
if err != nil {
@@ -450,7 +463,7 @@ type dialer interface {
removeStatic(*discover.Node)
}
-func (srv *Server) run(dialstate dialer) {
+func (srv *server) run(dialstate dialer) {
defer srv.loopWG.Done()
var (
peers = make(map[discover.NodeID]*Peer)
@@ -569,8 +582,8 @@ running:
if srv.ntab != nil {
srv.ntab.Close()
}
- if srv.DiscV5 != nil {
- srv.DiscV5.Close()
+ if srv.discV5 != nil {
+ srv.discV5.Close()
}
// Disconnect all peers.
for _, p := range peers {
@@ -586,7 +599,7 @@ running:
}
}
-func (srv *Server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
+func (srv *server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
// Drop connections with no matching protocols.
if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 {
return DiscUselessPeer
@@ -596,7 +609,7 @@ func (srv *Server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn
return srv.encHandshakeChecks(peers, c)
}
-func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
+func (srv *server) encHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
switch {
case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
return DiscTooManyPeers
@@ -615,7 +628,7 @@ type tempError interface {
// listenLoop runs in its own goroutine and accepts
// inbound connections.
-func (srv *Server) listenLoop() {
+func (srv *server) listenLoop() {
defer srv.loopWG.Done()
log.Info("RLPx listener up", "self", srv.makeSelf(srv.listener, srv.ntab))
@@ -676,7 +689,7 @@ func (srv *Server) listenLoop() {
// setupConn runs the handshakes and attempts to add the connection
// as a peer. It returns when the connection has been added as a peer
// or the handshakes have failed.
-func (srv *Server) setupConn(fd net.Conn, flags connFlag, dialDest *discover.Node) {
+func (srv *server) setupConn(fd net.Conn, flags connFlag, dialDest *discover.Node) {
// Prevent leftover pending conns from entering the handshake.
srv.lock.Lock()
running := srv.running
@@ -736,7 +749,7 @@ func truncateName(s string) string {
// checkpoint sends the conn to run, which performs the
// post-handshake checks for the stage (posthandshake, addpeer).
-func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error {
+func (srv *server) checkpoint(c *conn, stage chan<- *conn) error {
select {
case stage <- c:
case <-srv.quit:
@@ -753,20 +766,14 @@ func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error {
// runPeer runs in its own goroutine for each peer.
// it waits until the Peer logic returns and removes
// the peer.
-func (srv *Server) runPeer(p *Peer) {
+func (srv *server) runPeer(p *Peer) {
if srv.newPeerHook != nil {
srv.newPeerHook(p)
}
- if srv.PeerConnHook != nil {
- srv.PeerConnHook(p)
- }
remoteRequested, err := p.run()
// Note: run waits for existing peers to be sent on srv.delpeer
// before returning, so this send should not select on srv.quit.
srv.delpeer <- peerDrop{p, err, remoteRequested}
- if srv.PeerDisconnHook != nil {
- srv.PeerDisconnHook(p)
- }
}
// NodeInfo represents a short summary of the information known about the host.
@@ -784,7 +791,7 @@ type NodeInfo struct {
}
// NodeInfo gathers and returns a collection of metadata known about the host.
-func (srv *Server) NodeInfo() *NodeInfo {
+func (srv *server) NodeInfo() *NodeInfo {
node := srv.Self()
// Gather and assemble the generic node infos
@@ -813,7 +820,7 @@ func (srv *Server) NodeInfo() *NodeInfo {
}
// PeersInfo returns an array of metadata objects describing connected peers.
-func (srv *Server) PeersInfo() []*PeerInfo {
+func (srv *server) PeersInfo() []*PeerInfo {
// Gather all the generic and sub-protocol specific infos
infos := make([]*PeerInfo, 0, srv.PeerCount())
for _, peer := range srv.Peers() {
@@ -831,3 +838,7 @@ func (srv *Server) PeersInfo() []*PeerInfo {
}
return infos
}
+
+func (srv *server) DiscV5() *discv5.Network {
+ return srv.discV5
+}
diff --git a/p2p/server_test.go b/p2p/server_test.go
index 971faf002c..ca8f1873ee 100644
--- a/p2p/server_test.go
+++ b/p2p/server_test.go
@@ -72,7 +72,7 @@ func startTestServer(t *testing.T, id discover.NodeID, pf func(*Peer)) *Server {
ListenAddr: "127.0.0.1:0",
PrivateKey: newkey(),
}
- server := &Server{
+ server := &server{
Config: config,
newPeerHook: pf,
newTransport: func(fd net.Conn) transport { return newTestTransport(id, fd) },
@@ -201,7 +201,7 @@ func TestServerTaskScheduling(t *testing.T) {
// The Server in this test isn't actually running
// because we're only interested in what run does.
- srv := &Server{
+ srv := &server{
Config: Config{MaxPeers: 10},
quit: make(chan struct{}),
ntab: fakeTable{},
@@ -246,7 +246,7 @@ func TestServerManyTasks(t *testing.T) {
}
var (
- srv = &Server{quit: make(chan struct{}), ntab: fakeTable{}, running: true}
+ srv = &server{quit: make(chan struct{}), ntab: fakeTable{}, running: true}
done = make(chan *testTask)
start, end = 0, 0
)
@@ -317,7 +317,7 @@ func (t *testTask) Do(srv *Server) {
// at capacity. Trusted connections should still be accepted.
func TestServerAtCap(t *testing.T) {
trustedID := randomID()
- srv := &Server{
+ srv := &server{
Config: Config{
PrivateKey: newkey(),
MaxPeers: 10,
@@ -420,7 +420,7 @@ func TestServerSetupConn(t *testing.T) {
}
for i, test := range tests {
- srv := &Server{
+ srv := &server{
Config: Config{
PrivateKey: srvkey,
MaxPeers: 10,
diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go
index e1254bb97b..a04eb29dcc 100644
--- a/p2p/simulations/network.go
+++ b/p2p/simulations/network.go
@@ -37,6 +37,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/rpc"
)
type NetworkConfig struct {
@@ -582,16 +583,19 @@ func (self *Network) Connect(oneId, otherId *adapters.NodeId) error {
// any other way of connection (like peerpool) will need to call back
// to this method with connect = false to avoid infinite recursion
// this is not relevant for nodes starting up (which can only be externally triggered)
+ var addr []byte
+ var client *rpc.Client
if rev {
- err = conn.other.na.Connect(conn.one.na.Addr())
+ addr = conn.one.na.Addr()
+ client, err = conn.other.na.Client()
} else {
- err = conn.one.na.Connect(conn.other.na.Addr())
+ addr = conn.other.na.Addr()
+ client, err = conn.one.na.Client()
}
if err != nil {
return err
}
- return nil
- // return self.DidConnect(oneId, otherId)
+ return client.Call(nil, "admin_addPeer", string(addr))
}
// Disconnect(i, j) attempts to disconnect nodes i and j (args given as nodeId)
@@ -611,11 +615,20 @@ func (self *Network) Disconnect(oneId, otherId *adapters.NodeId) error {
if conn.One.NodeID != oneId.NodeID {
rev = true
}
+ var addr []byte
+ var client *rpc.Client
+ var err error
if rev {
- return conn.other.na.Disconnect(oneId.Bytes())
+ addr = conn.one.na.Addr()
+ client, err = conn.other.na.Client()
+ } else {
+ addr = conn.other.na.Addr()
+ client, err = conn.one.na.Client()
}
- return conn.one.na.Disconnect(otherId.Bytes())
- // return self.DidDisconnect(oneId, otherId)
+ if err != nil {
+ return err
+ }
+ return client.Call(nil, "admin_removePeer", string(addr))
}
func (self *Network) DidConnect(one, other *adapters.NodeId) error {
diff --git a/p2p/testing/protocolsession.go b/p2p/testing/protocolsession.go
index 208268ca81..723154e4f9 100644
--- a/p2p/testing/protocolsession.go
+++ b/p2p/testing/protocolsession.go
@@ -22,7 +22,6 @@ type TestMessenger interface {
type TestNodeAdapter interface {
GetPeer(id *adapters.NodeId) *adapters.Peer
- Connect([]byte) error
}
// exchanges are the basic units of protocol tests
@@ -57,9 +56,9 @@ type Disconnect struct {
Error error // disconnect reason
}
-func NewProtocolSession(na adapters.NodeAdapter, ids []*adapters.NodeId) *ProtocolSession {
+func NewProtocolSession(na TestNodeAdapter, ids []*adapters.NodeId) *ProtocolSession {
ps := &ProtocolSession{
- TestNodeAdapter: na.(TestNodeAdapter),
+ TestNodeAdapter: na,
Ids: ids,
}
return ps
diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go
index 393425f6fd..e900fd6672 100644
--- a/p2p/testing/protocoltester.go
+++ b/p2p/testing/protocoltester.go
@@ -5,79 +5,84 @@ import (
"testing"
"github.com/ethereum/go-ethereum/log"
+ "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/p2p/simulations"
+ "github.com/ethereum/go-ethereum/rpc"
)
type ProtocolTester struct {
*ProtocolSession
network *simulations.Network
- na adapters.NodeAdapter
}
-func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run adapters.ProtoCall) *ProtocolTester {
-
+func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
net := simulations.NewNetwork(&simulations.NetworkConfig{})
naf := func(conf *simulations.NodeConfig) adapters.NodeAdapter {
- na := adapters.NewSimNode(conf.Id, net)
+ node := &testNode{}
if conf.Id.NodeID == id.NodeID {
log.Trace(fmt.Sprintf("adapter run function set to protocol for node %v (=%v)", conf.Id, id))
- na.Run = run
+ node.run = run
}
- return na
+ return adapters.NewSimNode(conf.Id, node, net)
}
net.SetNaf(naf)
- err := net.NewNode(&simulations.NodeConfig{Id: id})
- if err != nil {
+
+ if err := net.NewNode(&simulations.NodeConfig{Id: id}); err != nil {
+ panic(err.Error())
+ }
+ if err := net.Start(id); err != nil {
panic(err.Error())
}
- //na := net.GetNode(id).Adapter()
- na := net.GetNodeAdapter(id)
-
+ node := net.GetNodeAdapter(id).(*adapters.SimNode)
ids := adapters.RandomNodeIds(n)
-
- ps := NewProtocolSession(na, ids)
+ ps := NewProtocolSession(node, ids)
self := &ProtocolTester{
ProtocolSession: ps,
network: net,
- na: na,
}
- self.Connect(ids...)
+ self.Connect(id, ids...)
return self
}
-func (self *ProtocolTester) Start(id *adapters.NodeId) error {
- err := self.network.NewNode(&simulations.NodeConfig{Id: id})
- if err != nil {
- return err
- }
- node := self.network.GetNode(id)
- if node == nil {
- log.Trace(fmt.Sprintf("node for peer %v not found", id))
- return nil
- }
- if node.Adapter() == nil {
- log.Trace(fmt.Sprintf("node adapter for peer %v not found", id))
- return nil
+func (self *ProtocolTester) Connect(selfId *adapters.NodeId, ids ...*adapters.NodeId) {
+ for _, id := range ids {
+ log.Trace(fmt.Sprintf("start node %v", id))
+ if err := self.network.NewNode(&simulations.NodeConfig{Id: id}); err != nil {
+ panic(fmt.Sprintf("error starting peer %v: %v", id, err))
+ }
+ if err := self.network.Start(id); err != nil {
+ panic(fmt.Sprintf("error starting peer %v: %v", id, err))
+ }
+ log.Trace(fmt.Sprintf("connect to %v", id))
+ if err := self.network.Connect(selfId, id); err != nil {
+ panic(fmt.Sprintf("error connecting to peer %v: %v", id, err))
+ }
}
+
+}
+
+// testNode wraps a protocol run function and implements the node.Service
+// interface
+type testNode struct {
+ run func(*p2p.Peer, p2p.MsgReadWriter) error
+}
+
+func (t *testNode) Protocols() []p2p.Protocol {
+ return []p2p.Protocol{{Run: t.run}}
+}
+
+func (t *testNode) APIs() []rpc.API {
return nil
}
-func (self *ProtocolTester) Connect(ids ...*adapters.NodeId) {
- for _, id := range ids {
- log.Trace(fmt.Sprintf("start node %v", id))
- err := self.Start(id)
- if err != nil {
- log.Trace(fmt.Sprintf("error starting peer %v: %v", id, err))
- }
- log.Trace(fmt.Sprintf("connect to %v", id))
- err = self.na.Connect(id.Bytes())
- if err != nil {
- log.Trace(fmt.Sprintf("error connecting to peer %v: %v", id, err))
- }
- }
-
+func (t *testNode) Start(server p2p.Server) error {
+ return nil
+}
+
+func (t *testNode) Stop() error {
+ return nil
}
diff --git a/swarm/network/pss_test.go b/swarm/network/pss_test.go
index d7dbfb0c35..497e6e8f59 100644
--- a/swarm/network/pss_test.go
+++ b/swarm/network/pss_test.go
@@ -5,9 +5,7 @@ import (
"testing"
"time"
- "github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/p2p/protocols"
- "github.com/ethereum/go-ethereum/p2p/simulations"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
@@ -173,13 +171,6 @@ func newPssBaseTester(t *testing.T, addr *peerAddr, n int) *pssTester {
to := NewKademlia(addr.OverlayAddr(), kp)
pp := NewHive(NewHiveParams(), to)
ps := NewPss(to, addr.OverlayAddr())
- net := simulations.NewNetwork(&simulations.NetworkConfig{})
- naf := func(conf *simulations.NodeConfig) adapters.NodeAdapter {
- na := adapters.NewSimNode(conf.Id, net)
- return na
- }
- net.SetNaf(naf)
-
srv := func(p Peer) error {
p.Register(&PssMsg{}, ps.HandlePssMsg)
pp.Add(p)
diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go
index 2db46f84fb..37087958df 100644
--- a/swarm/network/simulations/discovery/discovery_test.go
+++ b/swarm/network/simulations/discovery/discovery_test.go
@@ -26,7 +26,9 @@ const serviceName = "discovery"
func init() {
// register the discovery service which will run as a devp2p
// protocol when using the exec adapter
- adapters.RegisterService(serviceName, discoveryService)
+ adapters.RegisterService(serviceName, func(id *adapters.NodeId) p2pnode.Service {
+ return newNode(id)
+ })
// log.Root().SetHandler(log.LvlFilterHandler(log.LvlError, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
@@ -98,7 +100,9 @@ func TestDiscoverySimulationExecAdapter(t *testing.T) {
func TestDiscoverySimulationSimAdapter(t *testing.T) {
setup := func(net *simulations.Network, trigger chan *adapters.NodeId) {
net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter {
- return newSimNode(conf.Id, net, trigger)
+ node := newNode(conf.Id)
+ node.trigger = trigger
+ return adapters.NewSimNode(conf.Id, node, net)
})
}
@@ -188,25 +192,9 @@ type node struct {
*network.Hive
*adapters.SimNode
- id *adapters.NodeId
- trigger chan *adapters.NodeId
- protocol *p2p.Protocol
- connectPeer func(string) error
-}
-
-func newSimNode(id *adapters.NodeId, net *simulations.Network, trigger chan *adapters.NodeId) *node {
- node := newNode(id)
-
- node.SimNode = adapters.NewSimNode(id, net)
- node.Run = node.protocol.Run
-
- node.trigger = trigger
-
- node.connectPeer = func(s string) error {
- return node.Connect(adapters.NewNodeIdFromHex(s).Bytes())
- }
-
- return node
+ id *adapters.NodeId
+ trigger chan *adapters.NodeId
+ protocol *p2p.Protocol
}
func newNode(id *adapters.NodeId) *node {
@@ -249,8 +237,24 @@ func newHive(kademlia *network.Kademlia) *network.Hive {
return network.NewHive(params, kademlia)
}
-func (n *node) Start() error {
- return n.Hive.Start(n.connectPeer, n.hiveKeepAlive)
+func (n *node) Protocols() []p2p.Protocol {
+ return []p2p.Protocol{*n.protocol}
+}
+
+func (n *node) APIs() []rpc.API {
+ return nil
+}
+
+func (n *node) Start(server p2p.Server) error {
+ connectPeer := func(url string) error {
+ node, err := discover.ParseNode(url)
+ if err != nil {
+ return fmt.Errorf("invalid node URL: %v", err)
+ }
+ server.AddPeer(node)
+ return nil
+ }
+ return n.Hive.Start(connectPeer, n.hiveKeepAlive)
}
func (n *node) Stop() error {
@@ -277,38 +281,3 @@ func (n *node) triggerCheck() {
// TODO: rate limit the trigger?
go func() { n.trigger <- n.id }()
}
-
-func discoveryService(id *adapters.NodeId) p2pnode.ServiceConstructor {
- return func(ctx *p2pnode.ServiceContext) (p2pnode.Service, error) {
- node := newNode(id)
- return &p2pService{node}, nil
- }
-}
-
-type p2pService struct {
- node *node
-}
-
-func (s *p2pService) Protocols() []p2p.Protocol {
- return []p2p.Protocol{*s.node.protocol}
-}
-
-func (s *p2pService) APIs() []rpc.API {
- return nil
-}
-
-func (s *p2pService) Start(server *p2p.Server) error {
- s.node.connectPeer = func(url string) error {
- node, err := discover.ParseNode(url)
- if err != nil {
- return fmt.Errorf("invalid node URL: %v", err)
- }
- server.AddPeer(node)
- return nil
- }
- return s.node.Start()
-}
-
-func (s *p2pService) Stop() error {
- return s.node.Stop()
-}
diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go
index fb3c211f0e..fc4fae45af 100644
--- a/swarm/network/simulations/overlay.go
+++ b/swarm/network/simulations/overlay.go
@@ -16,21 +16,30 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/adapters"
+ "github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
+ "github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
)
// Network extends simulations.Network with hives for each node.
type Network struct {
*simulations.Network
- hives []*network.Hive
+ hives map[discover.NodeID]*network.Hive
}
// SimNode is the adapter used by Swarm simulations.
type SimNode struct {
- connect func(s string) error
- hive *network.Hive
- adapters.NodeAdapter
+ hive *network.Hive
+ protocol *p2p.Protocol
+}
+
+func (s *SimNode) Protocols() []p2p.Protocol {
+ return []p2p.Protocol{*s.protocol}
+}
+
+func (s *SimNode) APIs() []rpc.API {
+ return nil
}
// the hive update ticker for hive
@@ -39,26 +48,29 @@ func af() <-chan time.Time {
}
// Start() starts up the hive
-// makes SimNode implement *NodeAdapter
-func (self *SimNode) Start() error {
- return self.hive.Start(self.connect, af)
+// makes SimNode implement node.Service
+func (self *SimNode) Start(server p2p.Server) error {
+ connectPeer := func(url string) error {
+ node, err := discover.ParseNode(url)
+ if err != nil {
+ return fmt.Errorf("invalid node URL: %v", err)
+ }
+ server.AddPeer(node)
+ return nil
+ }
+ return self.hive.Start(connectPeer, af)
}
// Stop() shuts down the hive
-// makes SimNode implement *NodeAdapter
+// makes SimNode implement node.Service
func (self *SimNode) Stop() error {
self.hive.Stop()
return nil
}
-func (self *SimNode) RunProtocol(id *adapters.NodeId, rw, rrw p2p.MsgReadWriter, peer *adapters.Peer) error {
- return self.NodeAdapter.(adapters.ProtocolRunner).RunProtocol(id, rw, rrw, peer)
-}
-
// NewSimNode creates adapters for nodes in the simulation.
func (self *Network) NewSimNode(conf *simulations.NodeConfig) adapters.NodeAdapter {
id := conf.Id
- na := adapters.NewSimNode(id, self.Network)
addr := network.NewPeerAddrFromNodeId(id)
kp := network.NewKadParams()
@@ -70,12 +82,10 @@ func (self *Network) NewSimNode(conf *simulations.NodeConfig) adapters.NodeAdapt
kp.RetryInterval = 1000000
to := network.NewKademlia(addr.OverlayAddr(), kp) // overlay topology driver
- // to := network.NewTestOverlay(addr.OverlayAddr()) // overlay topology driver
hp := network.NewHiveParams()
hp.CallInterval = 5000
- pp := network.NewHive(hp, to) // hive
- self.hives = append(self.hives, pp) // remember hive
- // bzz protocol Run function. messaging through SimPipe
+ pp := network.NewHive(hp, to) // hive
+ self.hives[id.NodeID] = pp // remember hive
services := func(p network.Peer) error {
dp := network.NewDiscovery(p, to)
@@ -88,20 +98,19 @@ func (self *Network) NewSimNode(conf *simulations.NodeConfig) adapters.NodeAdapt
}
ct := network.BzzCodeMap(network.DiscoveryMsgs...) // bzz protocol code map
- na.Run = network.Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, services, nil, nil).Run
- connect := func(s string) error {
- return self.Connect(id, adapters.NewNodeIdFromHex(s))
- }
- return &SimNode{
- connect: connect,
- hive: pp,
- NodeAdapter: na,
+
+ node := &SimNode{
+ hive: pp,
+ protocol: network.Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, services, nil, nil),
}
+ return adapters.NewSimNode(id, node, self.Network)
+
}
-func NewNetwork(network *simulations.Network) *Network {
+func NewNetwork(net *simulations.Network) *Network {
n := &Network{
- Network: network,
+ Network: net,
+ hives: make(map[discover.NodeID]*network.Hive),
}
n.SetNaf(n.NewSimNode)
return n
@@ -128,7 +137,7 @@ func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simu
} else {
peerId = ids[i-1]
}
- err := net.hives[i].Register(network.NewPeerAddrFromNodeId(peerId))
+ err := net.hives[id.NodeID].Register(network.NewPeerAddrFromNodeId(peerId))
if err != nil {
panic(err.Error())
}
@@ -171,7 +180,7 @@ func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simu
Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) {
var results []string
for _, id := range ids {
- pp := net.GetNode(id).Adapter().(*SimNode).hive
+ pp := net.hives[id.NodeID]
results = append(results, pp.String())
}
return results, nil
@@ -185,7 +194,7 @@ func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simu
//GET //nodes/ -- returns 's kademlia table
Retrieve: &simulations.ResourceHandler{
Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) {
- pp := net.GetNode(id).Adapter().(*SimNode).hive
+ pp := net.hives[id.NodeID]
if pp != nil {
return pp.String(), nil
}
diff --git a/swarm/swarm.go b/swarm/swarm.go
index 442e68d517..00d92ea450 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -166,7 +166,7 @@ Start is called when the stack is started
* TODO: start subservices like sword, swear, swarmdns
*/
// implements the node.Service interface
-func (self *Swarm) Start(net *p2p.Server) error {
+func (self *Swarm) Start(net p2p.Server) error {
connectPeer := func(url string) error {
node, err := discover.ParseNode(url)
if err != nil {
diff --git a/whisper/whisperv2/main.go b/whisper/whisperv2/main.go
index be41604890..081e4fe438 100644
--- a/whisper/whisperv2/main.go
+++ b/whisper/whisperv2/main.go
@@ -48,14 +48,14 @@ func main() {
shh := whisper.New()
// Create an Ethereum peer to communicate through
- server := p2p.Server{
+ server := p2p.NewServer(p2p.Config{
PrivateKey: key,
MaxPeers: 10,
Name: name,
Protocols: []p2p.Protocol{shh.Protocol()},
ListenAddr: ":30300",
NAT: nat.Any(),
- }
+ })
fmt.Println("Starting Ethereum peer...")
if err := server.Start(); err != nil {
fmt.Printf("Failed to start Ethereum peer: %v.\n", err)
diff --git a/whisper/whisperv2/whisper.go b/whisper/whisperv2/whisper.go
index 1d7c21bd12..6d460bdee5 100644
--- a/whisper/whisperv2/whisper.go
+++ b/whisper/whisperv2/whisper.go
@@ -172,7 +172,7 @@ func (self *Whisper) Send(envelope *Envelope) error {
// Start implements node.Service, starting the background data propagation thread
// of the Whisper protocol.
-func (self *Whisper) Start(*p2p.Server) error {
+func (self *Whisper) Start(p2p.Server) error {
log.Info(fmt.Sprint("Whisper started"))
go self.update()
return nil
diff --git a/whisper/whisperv5/peer_test.go b/whisper/whisperv5/peer_test.go
index d3cd63b0b2..9e3b66c63d 100644
--- a/whisper/whisperv5/peer_test.go
+++ b/whisper/whisperv5/peer_test.go
@@ -78,7 +78,7 @@ type TestData struct {
type TestNode struct {
shh *Whisper
id *ecdsa.PrivateKey
- server *p2p.Server
+ server p2p.Server
filerId string
}
@@ -140,19 +140,17 @@ func initialize(t *testing.T) {
peers = append(peers, peer)
}
- node.server = &p2p.Server{
- Config: p2p.Config{
- PrivateKey: node.id,
- MaxPeers: NumNodes/2 + 1,
- Name: name,
- Protocols: node.shh.Protocols(),
- ListenAddr: addr,
- NAT: nat.Any(),
- BootstrapNodes: peers,
- StaticNodes: peers,
- TrustedNodes: peers,
- },
- }
+ node.server = p2p.NewServer(p2p.Config{
+ PrivateKey: node.id,
+ MaxPeers: NumNodes/2 + 1,
+ Name: name,
+ Protocols: node.shh.Protocols(),
+ ListenAddr: addr,
+ NAT: nat.Any(),
+ BootstrapNodes: peers,
+ StaticNodes: peers,
+ TrustedNodes: peers,
+ })
err = node.server.Start()
if err != nil {
diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go
index f2aad08efb..77ca20a58c 100644
--- a/whisper/whisperv5/whisper.go
+++ b/whisper/whisperv5/whisper.go
@@ -396,7 +396,7 @@ func (w *Whisper) Send(envelope *Envelope) error {
// Start implements node.Service, starting the background data propagation thread
// of the Whisper protocol.
-func (w *Whisper) Start(*p2p.Server) error {
+func (w *Whisper) Start(p2p.Server) error {
log.Info("started whisper v." + ProtocolVersionStr)
go w.update()