p2p/adapters: Use RPC with SimNodes

Signed-off-by: Lewis Marshall <lewis@lmars.net>

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 {

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,8 +253,7 @@ func initialize() {
 		maxPeers = 800
 	}
 
-	server = &p2p.Server{
+	server = p2p.NewServer(p2p.Config{
-		Config: p2p.Config{
 		PrivateKey:     nodeid,
 		MaxPeers:       maxPeers,
 		Name:           common.MakeName("wnode", "5.0"),
@@ -264,8 +263,7 @@ func initialize() {
 		BootstrapNodes: peers,
 		StaticNodes:    peers,
 		TrustedNodes:   peers,
-		},
+	})
-	}
 }
 
 func startServer() {

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
 }

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()

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()
 

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}
 }
 

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)

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 {
+	if v, ok := srvr.(discV5Server); ok {
-		topicDisc = srvr.DiscV5
+		topicDisc = v.DiscV5()
 	}
 	lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8]))
 	if pm.lightSync {

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)
 }
 

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()

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.

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()
 

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() {

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
 		}

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.

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)
 	}

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
+// serviceFunc returns a node.Service which can be used to boot devp2p nodes
-// devp2p nodes
+type serviceFunc func(id *NodeId) node.Service
-type serviceFunc func(id *NodeId) node.ServiceConstructor
 
 // 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 {
+	if n.client != nil {
-		n.Client.Close()
+		n.client.Close()
-		n.Client = nil
+		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
+	// initialize the service
-	service, exists := serviceFuncs[serviceName]
+	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)
+		log.Crit("error starting p2p 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)
 	}
 
 	// 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 {

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 {
+func (self *SimNode) stopRPC() {
-	return nil
+	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)
+	na.(*SimNode).RunProtocol(self, rrw, rw, peer)
-	if err != nil {
-		return fmt.Errorf("cannot run protocol (%v -> %v) %v", self.Id, id, err)
-	}
 
 	// run protocol on remote node with self as peer
-	err = self.RunProtocol(id, rw, rrw, peer)
+	self.RunProtocol(na.(*SimNode), rw, rrw, peer)
-	if err != nil {
+}
-		return fmt.Errorf("cannot run protocol (%v -> %v): %v", id, self.Id, err)
+
-	}
+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 {
+func (self *SimNode) RunProtocol(node *SimNode, rw, rrw p2p.MsgReadWriter, peer *Peer) {
-	if self.Run == nil {
+	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
 }

p2p/adapters/rlpx.go

@@ -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 <http://www.gnu.org/licenses/>.
-
-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
-	// <node-id>@<ip-address>:<tcp-port>(?udp=<udp-port>)
-	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
-}

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 {

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 {

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)

p2p/server.go

@@ -136,8 +136,22 @@ 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
 
@@ -145,8 +159,6 @@ type Server struct {
 	// the whole protocol stack.
 	newTransport func(net.Conn) transport
 	newPeerHook  func(*Peer)
-	PeerConnHook    func(*Peer)
-	PeerDisconnHook 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 {
+	if srv.discV5 != nil {
-		srv.DiscV5.Close()
+		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
+}

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,

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 client.Call(nil, "admin_addPeer", string(addr))
-	// return self.DidConnect(oneId, otherId)
 }
 
 // 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())
+	if err != nil {
-	// return self.DidDisconnect(oneId, otherId)
+		return err
+	}
+	return client.Call(nil, "admin_removePeer", string(addr))
 }
 
 func (self *Network) DidConnect(one, other *adapters.NodeId) error {

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

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()
+	node := net.GetNodeAdapter(id).(*adapters.SimNode)
-	na := net.GetNodeAdapter(id)
-
 	ids := adapters.RandomNodeIds(n)
-
+	ps := NewProtocolSession(node, ids)
-	ps := NewProtocolSession(na, 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 {
+func (self *ProtocolTester) Connect(selfId *adapters.NodeId, ids ...*adapters.NodeId) {
-	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
-	}
-	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 := self.network.NewNode(&simulations.NodeConfig{Id: id}); err != nil {
-		if err != nil {
+			panic(fmt.Sprintf("error starting peer %v: %v", id, err))
-			log.Trace(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))
-		err = self.na.Connect(id.Bytes())
+		if err := self.network.Connect(selfId, id); err != nil {
-		if err != nil {
+			panic(fmt.Sprintf("error connecting to peer %v: %v", id, err))
-			log.Trace(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 (t *testNode) Start(server p2p.Server) error {
+	return nil
+}
+
+func (t *testNode) Stop() error {
+	return nil
+}

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)

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)
 		})
 	}
 
@@ -191,22 +195,6 @@ type node struct {
 	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
 }
 
 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 {
+func (n *node) Protocols() []p2p.Protocol {
-	return n.Hive.Start(n.connectPeer, n.hiveKeepAlive)
+	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()
-}

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
+	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
+// makes SimNode implement node.Service
-func (self *SimNode) Start() error {
+func (self *SimNode) Start(server p2p.Server) error {
-	return self.hive.Start(self.connect, af)
+	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
+	self.hives[id.NodeID] = pp    // remember hive
-	// bzz protocol Run function. messaging through SimPipe
 
 	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 {
+	node := &SimNode{
-		return self.Connect(id, adapters.NewNodeIdFromHex(s))
-	}
-	return &SimNode{
-		connect:     connect,
 		hive:     pp,
-		NodeAdapter: na,
+		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 /<networkId>/nodes/<nodeId>  -- returns <nodeId>'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
 						}

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 {

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)

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

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,8 +140,7 @@ func initialize(t *testing.T) {
 			peers = append(peers, peer)
 		}
 
-		node.server = &p2p.Server{
+		node.server = p2p.NewServer(p2p.Config{
-			Config: p2p.Config{
 			PrivateKey:     node.id,
 			MaxPeers:       NumNodes/2 + 1,
 			Name:           name,
@@ -151,8 +150,7 @@ func initialize(t *testing.T) {
 			BootstrapNodes: peers,
 			StaticNodes:    peers,
 			TrustedNodes:   peers,
-			},
+		})
-		}
 
 		err = node.server.Start()
 		if err != nil {

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()