Merge pull request #97 from ethersphere/network-testing-framework-internode-rpc

p2p/simulations: Support inter-node RPC connections

node/api.go

@@ -203,7 +203,7 @@ func (api *PrivateAdminAPI) StartWS(host *string, port *int, allowedOrigins *str
 		}
 	}
 
-	if err := api.node.startWS(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, origins); err != nil {
+	if err := api.node.startWS(fmt.Sprintf("%s:%d", *host, *port), api.node.rpcAPIs, modules, origins, api.node.config.WSExposeAll); err != nil {
 		return false, err
 	}
 	return true, nil

node/config.go

@@ -128,6 +128,13 @@ type Config struct {
 	// If the module list is empty, all RPC API endpoints designated public will be
 	// exposed.
 	WSModules []string `toml:",omitempty"`
+
+	// WSExposeAll exposes all API modules via the WebSocket RPC interface rather
+	// than just the public ones.
+	//
+	// *WARNING* Only set this if the node is running in a trusted network, exposing
+	// private APIs to untrusted users is a major security risk.
+	WSExposeAll bool `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into

node/node.go

@@ -275,7 +275,7 @@ func (n *Node) startRPC(services map[reflect.Type]Service) error {
 		n.stopInProc()
 		return err
 	}
-	if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins); err != nil {
+	if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil {
 		n.stopHTTP()
 		n.stopIPC()
 		n.stopInProc()
@@ -426,7 +426,7 @@ func (n *Node) stopHTTP() {
 }
 
 // startWS initializes and starts the websocket RPC endpoint.
-func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins []string) error {
+func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins []string, exposeAll bool) error {
 	// Short circuit if the WS endpoint isn't being exposed
 	if endpoint == "" {
 		return nil
@@ -439,7 +439,7 @@ func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrig
 	// Register all the APIs exposed by the services
 	handler := rpc.NewServer()
 	for _, api := range apis {
-		if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
+		if exposeAll || whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
 			if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
 				return err
 			}
@@ -455,7 +455,7 @@ func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrig
 		return err
 	}
 	go rpc.NewWSServer(wsOrigins, handler).Serve(listener)
-	log.Info(fmt.Sprintf("WebSocket endpoint opened: ws://%s", endpoint))
+	log.Info(fmt.Sprintf("WebSocket endpoint opened: ws://%s", listener.Addr()))
 
 	// All listeners booted successfully
 	n.wsEndpoint = endpoint

p2p/simulations/adapters/exec.go

@@ -1,6 +1,7 @@
 package adapters
 
 import (
+	"bufio"
 	"context"
 	"crypto/ecdsa"
 	"encoding/json"
@@ -12,7 +13,9 @@ import (
 	"os/exec"
 	"os/signal"
 	"path/filepath"
+	"regexp"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -22,6 +25,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/rpc"
+	"golang.org/x/net/websocket"
 )
 
 // ExecAdapter is a NodeAdapter which runs nodes by executing the current
@@ -32,12 +36,17 @@ import (
 // execP2PNode function for more information.
 type ExecAdapter struct {
 	BaseDir string
+
+	nodes map[discover.NodeID]*ExecNode
 }
 
 // NewExecAdapter returns an ExecAdapter which stores node data in
 // subdirectories of the given base directory
 func NewExecAdapter(baseDir string) *ExecAdapter {
-	return &ExecAdapter{BaseDir: baseDir}
+	return &ExecAdapter{
+		BaseDir: baseDir,
+		nodes:   make(map[discover.NodeID]*ExecNode),
+	}
 }
 
 // Name returns the name of the adapter for logging purpoeses
@@ -69,6 +78,10 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 		Node:  config,
 	}
 	conf.Stack.DataDir = filepath.Join(dir, "data")
+	conf.Stack.WSHost = "127.0.0.1"
+	conf.Stack.WSPort = 0
+	conf.Stack.WSOrigins = []string{"*"}
+	conf.Stack.WSExposeAll = true
 	conf.Stack.P2P.EnableMsgEvents = false
 	conf.Stack.P2P.NoDiscovery = true
 	conf.Stack.P2P.NAT = nil
@@ -78,11 +91,13 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 	conf.Stack.P2P.ListenAddr = "127.0.0.1:0"
 
 	node := &ExecNode{
-		ID:     config.ID,
+		ID:      config.ID,
-		Dir:    dir,
+		Dir:     dir,
-		Config: conf,
+		Config:  conf,
+		adapter: e,
 	}
 	node.newCmd = node.execCommand
+	e.nodes[node.ID] = node
 	return node, nil
 }
 
@@ -100,10 +115,11 @@ type ExecNode struct {
 	Cmd    *exec.Cmd
 	Info   *p2p.NodeInfo
 
-	client *rpc.Client
+	adapter *ExecAdapter
-	rpcMux *rpcMux
+	client  *rpc.Client
-	newCmd func() *exec.Cmd
+	wsAddr  string
-	key    *ecdsa.PrivateKey
+	newCmd  func() *exec.Cmd
+	key     *ecdsa.PrivateKey
 }
 
 // Addr returns the node's enode URL
@@ -120,6 +136,10 @@ func (n *ExecNode) Client() (*rpc.Client, error) {
 	return n.client, nil
 }
 
+// wsAddrPattern is a regex used to read the WebSocket address from the node's
+// log
+var wsAddrPattern = regexp.MustCompile(`ws://[\d.:]+`)
+
 // 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
@@ -137,34 +157,70 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) {
 	// encode a copy of the config containing the snapshot
 	confCopy := *n.Config
 	confCopy.Snapshots = snapshots
+	confCopy.PeerAddrs = make(map[string]string)
+	for id, node := range n.adapter.nodes {
+		confCopy.PeerAddrs[id.String()] = node.wsAddr
+	}
 	confData, err := json.Marshal(confCopy)
 	if err != nil {
 		return fmt.Errorf("error generating node config: %s", err)
 	}
 
-	// create a net.Pipe for RPC communication over stdin / stdout
+	// use a pipe for stderr so we can both copy the node's stderr to
-	pipe1, pipe2 := net.Pipe()
+	// os.Stderr and read the WebSocket address from the logs
+	stderrR, stderrW := io.Pipe()
+	stderr := io.MultiWriter(os.Stderr, stderrW)
 
 	// start the node
 	cmd := n.newCmd()
-	cmd.Stdin = pipe1
+	cmd.Stdout = os.Stdout
-	cmd.Stdout = pipe1
+	cmd.Stderr = stderr
-	cmd.Stderr = os.Stderr
 	cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", confData))
 	if err := cmd.Start(); err != nil {
 		return fmt.Errorf("error starting node: %s", err)
 	}
 	n.Cmd = cmd
 
+	// read the WebSocket address from the stderr logs
+	var wsAddr string
+	errC := make(chan error)
+	go func() {
+		s := bufio.NewScanner(stderrR)
+		for s.Scan() {
+			if strings.Contains(s.Text(), "WebSocket endpoint opened:") {
+				wsAddr = wsAddrPattern.FindString(s.Text())
+				break
+			}
+		}
+		select {
+		case errC <- s.Err():
+		default:
+		}
+	}()
+	select {
+	case err := <-errC:
+		if err != nil {
+			return fmt.Errorf("error reading WebSocket address from stderr: %s", err)
+		} else if wsAddr == "" {
+			return errors.New("failed to read WebSocket address from stderr")
+		}
+	case <-time.After(10 * time.Second):
+		return errors.New("timed out waiting for WebSocket address on stderr")
+	}
+
 	// create the RPC client and load the node info
-	n.rpcMux = newRPCMux(pipe2)
-	n.client = n.rpcMux.Client()
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
+	client, err := rpc.DialWebsocket(ctx, wsAddr, "")
+	if err != nil {
+		return fmt.Errorf("error dialing rpc websocket: %s", err)
+	}
 	var info p2p.NodeInfo
-	if err := n.client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil {
+	if err := client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil {
 		return fmt.Errorf("error getting node info: %s", err)
 	}
+	n.client = client
+	n.wsAddr = wsAddr
 	n.Info = &info
 
 	return nil
@@ -193,6 +249,7 @@ func (n *ExecNode) Stop() error {
 	if n.client != nil {
 		n.client.Close()
 		n.client = nil
+		n.wsAddr = ""
 		n.Info = nil
 	}
 
@@ -222,13 +279,30 @@ func (n *ExecNode) NodeInfo() *p2p.NodeInfo {
 	return info
 }
 
-// ServeRPC serves RPC requests over the given connection using the node's
+// ServeRPC serves RPC requests over the given connection by dialling the
-// RPC multiplexer
+// node's WebSocket address and joining the two connections
-func (n *ExecNode) ServeRPC(conn net.Conn) error {
+func (n *ExecNode) ServeRPC(clientConn net.Conn) error {
-	if n.rpcMux == nil {
+	conn, err := websocket.Dial(n.wsAddr, "", "http://localhost")
-		return errors.New("RPC not started")
+	if err != nil {
+		return err
 	}
-	n.rpcMux.Serve(conn)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	join := func(src, dst net.Conn) {
+		defer wg.Done()
+		io.Copy(dst, src)
+		// close the write end of the destination connection
+		if cw, ok := dst.(interface {
+			CloseWrite() error
+		}); ok {
+			cw.CloseWrite()
+		} else {
+			dst.Close()
+		}
+	}
+	go join(conn, clientConn)
+	go join(clientConn, conn)
+	wg.Wait()
 	return nil
 }
 
@@ -253,7 +327,8 @@ func init() {
 type execNodeConfig struct {
 	Stack     node.Config       `json:"stack"`
 	Node      *NodeConfig       `json:"node"`
-	Snapshots map[string][]byte `json:"snapshot,omitempty"`
+	Snapshots map[string][]byte `json:"snapshots,omitempty"`
+	PeerAddrs map[string]string `json:"peer_addrs,omitempty"`
 }
 
 // execP2PNode starts a devp2p node when the current binary is executed with
@@ -264,9 +339,8 @@ func execP2PNode() {
 	glogger.Verbosity(log.LvlInfo)
 	log.Root().SetHandler(glogger)
 
-	// read the services and ID from argv
+	// read the services from argv
 	serviceNames := strings.Split(os.Args[1], ",")
-	id := discover.MustHexID(os.Args[2])
 
 	// decode the config
 	confEnv := os.Getenv("_P2P_NODE_CONFIG")
@@ -279,20 +353,6 @@ func execP2PNode() {
 	}
 	conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey
 
-	// initialize the services
-	services := make(map[string]node.Service, len(serviceNames))
-	for _, name := range serviceNames {
-		serviceFunc, exists := serviceFuncs[name]
-		if !exists {
-			log.Crit(fmt.Sprintf("unknown node service %q", name))
-		}
-		var snapshot []byte
-		if conf.Snapshots != nil {
-			snapshot = conf.Snapshots[name]
-		}
-		services[name] = serviceFunc(id, snapshot)
-	}
-
 	// use explicit IP address in ListenAddr so that Enode URL is usable
 	if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") {
 		addrs, err := net.InterfaceAddrs()
@@ -307,21 +367,54 @@ func execP2PNode() {
 		}
 	}
 
-	// start the devp2p stack
+	// initialize the devp2p stack
-	stack, err := startP2PNode(&conf.Stack, services)
+	stack, err := node.New(&conf.Stack)
 	if err != nil {
-		log.Crit("error starting p2p node", "err", err)
+		log.Crit("error creating node stack", "err", err)
 	}
 
-	// use stdin / stdout for RPC to avoid the parent needing to access
+	// register the services, collecting them into a map so we can wrap
-	// either the local filesystem or TCP stack (useful when running in
+	// them in a snapshot service
-	// Docker)
+	services := make(map[string]node.Service, len(serviceNames))
-	handler, err := stack.RPCHandler()
+	for _, name := range serviceNames {
-	if err != nil {
+		serviceFunc, exists := serviceFuncs[name]
-		log.Crit("error getting RPC server", "err", err)
+		if !exists {
+			log.Crit("unknown node service", "name", name)
+		}
+		constructor := func(nodeCtx *node.ServiceContext) (node.Service, error) {
+			ctx := &ServiceContext{
+				RPCDialer:   &wsRPCDialer{addrs: conf.PeerAddrs},
+				NodeContext: nodeCtx,
+				Config:      conf.Node,
+			}
+			if conf.Snapshots != nil {
+				ctx.Snapshot = conf.Snapshots[name]
+			}
+			service, err := serviceFunc(ctx)
+			if err != nil {
+				return nil, err
+			}
+			services[name] = service
+			return service, nil
+		}
+		if err := stack.Register(constructor); err != nil {
+			log.Crit("error starting service", "name", name, "err", err)
+		}
 	}
-	go handler.ServeCodec(rpc.NewJSONCodec(&stdioConn{os.Stdin, os.Stdout}), rpc.OptionMethodInvocation|rpc.OptionSubscriptions)
 
+	// register the snapshot service
+	if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
+		return &snapshotService{services}, nil
+	}); err != nil {
+		log.Crit("error starting snapshot service", "err", err)
+	}
+
+	// start the stack
+	if err := stack.Start(); err != nil {
+		log.Crit("error stating node stack", "err", err)
+	}
+
+	// stop the stack if we get a SIGTERM signal
 	go func() {
 		sigc := make(chan os.Signal, 1)
 		signal.Notify(sigc, syscall.SIGTERM)
@@ -331,33 +424,10 @@ func execP2PNode() {
 		stack.Stop()
 	}()
 
+	// wait for the stack to exit
 	stack.Wait()
 }
 
-func startP2PNode(conf *node.Config, services map[string]node.Service) (*node.Node, error) {
-	stack, err := node.New(conf)
-	if err != nil {
-		return nil, err
-	}
-	constructor := func(service node.Service) func(ctx *node.ServiceContext) (node.Service, error) {
-		return func(ctx *node.ServiceContext) (node.Service, error) {
-			return service, nil
-		}
-	}
-	for _, service := range services {
-		if err := stack.Register(constructor(service)); err != nil {
-			return nil, err
-		}
-	}
-	if err := stack.Register(constructor(&snapshotService{services})); err != nil {
-		return nil, err
-	}
-	if err := stack.Start(); err != nil {
-		return nil, err
-	}
-	return stack, nil
-}
-
 // snapshotService is a node.Service which wraps a list of services and
 // exposes an API to generate a snapshot of those services
 type snapshotService struct {
@@ -405,13 +475,14 @@ func (api SnapshotAPI) Snapshot() (map[string][]byte, error) {
 	return snapshots, nil
 }
 
-// stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can
+type wsRPCDialer struct {
-// use stdio for RPC messages
+	addrs map[string]string
-type stdioConn struct {
-	io.Reader
-	io.Writer
 }
 
-func (r *stdioConn) Close() error {
+func (w *wsRPCDialer) DialRPC(id discover.NodeID) (*rpc.Client, error) {
-	return nil
+	addr, ok := w.addrs[id.String()]
+	if !ok {
+		return nil, fmt.Errorf("unknown node: %s", id)
+	}
+	return rpc.DialWebsocket(context.Background(), addr, "http://localhost")
 }

p2p/simulations/adapters/inproc.go

@@ -78,6 +78,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 		ID:      id,
 		config:  config,
 		adapter: s,
+		running: make(map[string]node.Service),
 	}
 	s.nodes[id] = node
 	return node, nil
@@ -97,6 +98,24 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
 	return pipe2, nil
 }
 
+func (s *SimAdapter) DialRPC(id discover.NodeID) (*rpc.Client, error) {
+	simNode, ok := s.GetNode(id)
+	if !ok {
+		return nil, fmt.Errorf("unknown node: %s", id)
+	}
+	simNode.lock.RLock()
+	node := simNode.node
+	simNode.lock.RUnlock()
+	if node == nil {
+		return nil, errors.New("node not started")
+	}
+	handler, err := node.RPCHandler()
+	if err != nil {
+		return nil, err
+	}
+	return rpc.DialInProc(handler), nil
+}
+
 // GetNode returns the node with the given ID if it exists
 func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) {
 	s.mtx.RLock()
@@ -117,9 +136,8 @@ type SimNode struct {
 	config  *NodeConfig
 	adapter *SimAdapter
 	node    *node.Node
-	running []node.Service
+	running map[string]node.Service
 	client  *rpc.Client
-	rpcMux  *rpcMux
 }
 
 // Addr returns the node's discovery address
@@ -135,24 +153,28 @@ func (self *SimNode) Node() *discover.Node {
 // Client returns an rpc.Client which can be used to communicate with the
 // underlying service (it is set once the node has started)
 func (self *SimNode) Client() (*rpc.Client, error) {
-	self.lock.Lock()
+	self.lock.RLock()
-	defer self.lock.Unlock()
+	defer self.lock.RUnlock()
 	if self.client == nil {
-		return nil, errors.New("RPC not started")
+		return nil, errors.New("node not started")
 	}
 	return self.client, nil
 }
 
-// ServeRPC serves RPC requests over the given connection using the node's
+// ServeRPC serves RPC requests over the given connection by creating an
-// RPC multiplexer
+// in-process client to the node's RPC server
 func (self *SimNode) ServeRPC(conn net.Conn) error {
-	self.lock.Lock()
+	self.lock.RLock()
-	mux := self.rpcMux
+	node := self.node
-	self.lock.Unlock()
+	self.lock.RUnlock()
-	if mux == nil {
+	if node == nil {
-		return errors.New("RPC not started")
+		return errors.New("node not started")
 	}
-	mux.Serve(conn)
+	handler, err := node.RPCHandler()
+	if err != nil {
+		return err
+	}
+	handler.ServeCodec(rpc.NewJSONCodec(conn), rpc.OptionMethodInvocation|rpc.OptionSubscriptions)
 	return nil
 }
 
@@ -160,12 +182,24 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
 // simulation_snapshot RPC method
 func (self *SimNode) Snapshots() (map[string][]byte, error) {
 	self.lock.Lock()
-	defer self.lock.Unlock()
+	services := self.running
-	if self.client == nil {
+	self.lock.Unlock()
-		return nil, errors.New("RPC not started")
+	if len(services) == 0 {
+		return nil, errors.New("no running services")
 	}
-	var snapshots map[string][]byte
+	snapshots := make(map[string][]byte)
-	return snapshots, self.client.Call(&snapshots, "simulation_snapshot")
+	for name, service := range services {
+		if s, ok := service.(interface {
+			Snapshot() ([]byte, error)
+		}); ok {
+			snap, err := s.Snapshot()
+			if err != nil {
+				return nil, err
+			}
+			snapshots[name] = snap
+		}
+	}
+	return snapshots, nil
 }
 
 // Start starts the RPC handler and the underlying service
@@ -177,14 +211,21 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
 	}
 
 	newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) {
-		return func(ctx *node.ServiceContext) (node.Service, error) {
+		return func(nodeCtx *node.ServiceContext) (node.Service, error) {
-			var snapshot []byte
+			ctx := &ServiceContext{
+				RPCDialer:   self.adapter,
+				NodeContext: nodeCtx,
+				Config:      self.config,
+			}
 			if snapshots != nil {
-				snapshot = snapshots[name]
+				ctx.Snapshot = snapshots[name]
 			}
 			serviceFunc := self.adapter.services[name]
-			service := serviceFunc(self.ID, snapshot)
+			service, err := serviceFunc(ctx)
-			self.running = append(self.running, service)
+			if err != nil {
+				return nil, err
+			}
+			self.running[name] = service
 			return service, nil
 		}
 	}
@@ -213,18 +254,12 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
 		return err
 	}
 
+	// create an in-process RPC client
 	handler, err := node.RPCHandler()
 	if err != nil {
 		return err
 	}
-
+	self.client = rpc.DialInProc(handler)
-	// create an in-process RPC multiplexer
-	pipe1, pipe2 := net.Pipe()
-	go handler.ServeCodec(rpc.NewJSONCodec(pipe1), rpc.OptionMethodInvocation|rpc.OptionSubscriptions)
-	self.rpcMux = newRPCMux(pipe2)
-
-	// create an in-process RPC client
-	self.client = self.rpcMux.Client()
 
 	self.node = node
 
@@ -248,7 +283,11 @@ func (self *SimNode) Stop() error {
 func (self *SimNode) Services() []node.Service {
 	self.lock.Lock()
 	defer self.lock.Unlock()
-	return self.running
+	services := make([]node.Service, 0, len(self.running))
+	for _, service := range self.running {
+		services = append(services, service)
+	}
+	return services
 }
 
 func (self *SimNode) Server() *p2p.Server {

p2p/simulations/adapters/rpc_mux.go

@@ -1,218 +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 (
-	"encoding/json"
-	"net"
-	"strconv"
-	"strings"
-	"sync"
-
-	"github.com/ethereum/go-ethereum/rpc"
-)
-
-// rpcMux is an RPC multiplexer which allows many clients to make RPC requests
-// over a single connection by changing each request's ID to a unique value.
-//
-// This is used by node adapters so that simulations can create many RPC
-// clients all sending requests over the underlying node's stdin / stdout.
-type rpcMux struct {
-	conn net.Conn
-
-	mtx       sync.Mutex
-	idCounter uint64
-	msgMap    map[uint64]*rpcMsg
-	subMap    map[string]*rpcReply
-	send      chan *rpcMsg
-}
-
-type rpcMsg struct {
-	Method  string          `json:"method,omitempty"`
-	Version string          `json:"jsonrpc,omitempty"`
-	ID      json.RawMessage `json:"id,omitempty"`
-	Payload json.RawMessage `json:"params,omitempty"`
-	Result  json.RawMessage `json:"result,omitempty"`
-	Error   json.RawMessage `json:"error,omitempty"`
-
-	id    uint64
-	reply *rpcReply
-}
-
-// rpcSub is the payload or result of a subscription RPC message
-type rpcSub struct {
-	Subscription string          `json:"subscription"`
-	Result       json.RawMessage `json:"result,omitempty"`
-}
-
-// rpcReply receives replies to RPC messages for a particular client
-type rpcReply struct {
-	ch        chan *rpcMsg
-	closeOnce sync.Once
-}
-
-func (r *rpcReply) close() {
-	r.closeOnce.Do(func() { close(r.ch) })
-}
-
-func newRPCMux(conn net.Conn) *rpcMux {
-	mux := &rpcMux{
-		msgMap: make(map[uint64]*rpcMsg),
-		subMap: make(map[string]*rpcReply),
-		send:   make(chan *rpcMsg),
-	}
-	go mux.sendLoop(conn)
-	go mux.recvLoop(conn)
-	return mux
-}
-
-// Client creates a new RPC client which sends messages through the multiplexer
-func (mux *rpcMux) Client() *rpc.Client {
-	pipe1, pipe2 := net.Pipe()
-	go mux.Serve(pipe1)
-	return rpc.NewClientWithConn(pipe2)
-}
-
-// Serve reads RPC messages from the given connection, forwards them to the
-// multiplexed connnection and writes replies back to the given connection
-func (mux *rpcMux) Serve(conn net.Conn) {
-	// reply will receive replies to any messages we send
-	reply := &rpcReply{ch: make(chan *rpcMsg)}
-	defer func() {
-		// drain the channel to prevent blocking the recvLoop
-		for range reply.ch {
-		}
-	}()
-
-	// start a goroutine to read RPC messages from the connection and
-	// forward them to the sendLoop
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		dec := json.NewDecoder(conn)
-		for {
-			msg := &rpcMsg{}
-			if err := dec.Decode(msg); err != nil {
-				return
-			}
-			msg.reply = reply
-			mux.send <- msg
-		}
-	}()
-
-	// write message replies to the connection
-	enc := json.NewEncoder(conn)
-	for {
-		select {
-		case msg, ok := <-reply.ch:
-			if !ok {
-				return
-			}
-			if err := enc.Encode(msg); err != nil {
-				return
-			}
-		case <-done:
-			return
-		}
-	}
-}
-
-// sendLoop receives messages from the send channel, changes their ID and
-// writes them to the given connection
-func (mux *rpcMux) sendLoop(conn net.Conn) {
-	enc := json.NewEncoder(conn)
-	for msg := range mux.send {
-		if err := enc.Encode(mux.newMsg(msg)); err != nil {
-			return
-		}
-	}
-}
-
-// recvLoop reads messages from the given connection, changes their ID back
-// to the oringal value and sends them to the message's reply channel
-func (mux *rpcMux) recvLoop(conn net.Conn) {
-	// close all reply channels if we get an error
-	defer func() {
-		mux.mtx.Lock()
-		defer mux.mtx.Unlock()
-		for _, msg := range mux.msgMap {
-			msg.reply.close()
-		}
-	}()
-
-	dec := json.NewDecoder(conn)
-	for {
-		msg := &rpcMsg{}
-		if err := dec.Decode(msg); err != nil {
-			return
-		}
-		if reply := mux.lookup(msg); reply != nil {
-			reply.ch <- msg
-		}
-	}
-}
-
-// newMsg copies the given message and changes it's ID to a unique value
-func (mux *rpcMux) newMsg(msg *rpcMsg) *rpcMsg {
-	mux.mtx.Lock()
-	defer mux.mtx.Unlock()
-	id := mux.idCounter
-	mux.idCounter++
-	mux.msgMap[id] = msg
-	newMsg := *msg
-	newMsg.ID = json.RawMessage(strconv.FormatUint(id, 10))
-	return &newMsg
-}
-
-// lookup looks up the original message for which the given message is a reply
-func (mux *rpcMux) lookup(msg *rpcMsg) *rpcReply {
-	mux.mtx.Lock()
-	defer mux.mtx.Unlock()
-
-	// if the message has no ID, it is a subscription notification so
-	// lookup the original subscribe message
-	if msg.ID == nil {
-		sub := &rpcSub{}
-		if err := json.Unmarshal(msg.Payload, sub); err != nil {
-			return nil
-		}
-		return mux.subMap[sub.Subscription]
-	}
-
-	// lookup the original message and restore the ID
-	id, err := strconv.ParseUint(string(msg.ID), 10, 64)
-	if err != nil {
-		return nil
-	}
-	origMsg, ok := mux.msgMap[id]
-	if !ok {
-		return nil
-	}
-	delete(mux.msgMap, id)
-	msg.ID = origMsg.ID
-
-	// if the original message was a subscription, store the subscription
-	// ID so we can detect notifications
-	if strings.HasSuffix(string(origMsg.Method), "_subscribe") {
-		var result string
-		if err := json.Unmarshal(msg.Result, &result); err == nil {
-			mux.subMap[result] = origMsg.reply
-		}
-	}
-
-	return origMsg.reply
-}

p2p/simulations/adapters/types.go

@@ -158,11 +158,23 @@ func RandomNodeConfig() *NodeConfig {
 	}
 }
 
+type ServiceContext struct {
+	RPCDialer
+
+	NodeContext *node.ServiceContext
+	Config      *NodeConfig
+	Snapshot    []byte
+}
+
+type RPCDialer interface {
+	DialRPC(id discover.NodeID) (*rpc.Client, error)
+}
+
 // Services is a collection of services which can be run in a simulation
 type Services map[string]ServiceFunc
 
 // ServiceFunc returns a node.Service which can be used to boot devp2p nodes
-type ServiceFunc func(id discover.NodeID, snapshot []byte) node.Service
+type ServiceFunc func(ctx *ServiceContext) (node.Service, error)
 
 // serviceFuncs is a map of registered services which are used to boot devp2p
 // nodes

p2p/simulations/examples/connectivity.go

@@ -27,8 +27,8 @@ func main() {
 	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
 
 	services := map[string]adapters.ServiceFunc{
-		"ping-pong": func(id discover.NodeID, snapshot []byte) node.Service {
+		"ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return newPingPongService(id)
+			return newPingPongService(ctx.Config.ID), nil
 		},
 	}
 

p2p/simulations/http.go

@@ -255,6 +255,10 @@ type ServerConfig struct {
 	// network
 	NewAdapter func() adapters.NodeAdapter
 
+	// ExternalNetworks are externally defined networks to expose via the
+	// HTTP server
+	ExternalNetworks map[string]*Network
+
 	// Mocker is the function which will be called when a client sends a
 	// POST request to /networks/<netid>/mock and is expected to
 	// generate some mock events in the network
@@ -286,6 +290,9 @@ func NewServer(config *ServerConfig) *Server {
 		router:       httprouter.New(),
 		networks:     make(map[string]*Network),
 	}
+	for name, network := range config.ExternalNetworks {
+		s.networks[name] = network
+	}
 
 	s.OPTIONS("/networks", s.Options)
 	s.POST("/networks", s.CreateNetwork)

p2p/simulations/http_test.go

@@ -24,10 +24,10 @@ type testService struct {
 	state atomic.Value
 }
 
-func newTestService(id discover.NodeID, snapshot []byte) node.Service {
+func newTestService(ctx *adapters.ServiceContext) (node.Service, error) {
-	svc := &testService{id: id}
+	svc := &testService{id: ctx.Config.ID}
-	svc.state.Store(snapshot)
+	svc.state.Store(ctx.Snapshot)
-	return svc
+	return svc, nil
 }
 
 func (t *testService) Protocols() []p2p.Protocol {
@@ -185,8 +185,7 @@ func TestHTTPNetwork(t *testing.T) {
 	// create 2 nodes
 	nodeIDs := make([]string, 2)
 	for i := 0; i < 2; i++ {
-		config := &adapters.NodeConfig{}
+		node, err := client.CreateNode(network.ID, nil)
-		node, err := client.CreateNode(network.ID, config)
 		if err != nil {
 			t.Fatalf("error creating node: %s", err)
 		}
@@ -286,8 +285,8 @@ func (t *expectEvents) expect(events ...*Event) {
 				if event.Node == nil {
 					t.Fatal("expected event.Node to be set")
 				}
-				if event.Node.ID().NodeID != expected.Node.ID().NodeID {
+				if event.Node.ID() != expected.Node.ID() {
-					t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().Label(), event.Node.ID().Label())
+					t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().TerminalString(), event.Node.ID().TerminalString())
 				}
 				if event.Node.Up != expected.Node.Up {
 					t.Fatalf("expected node event %d to have up=%t, got up=%t", i, expected.Node.Up, event.Node.Up)
@@ -297,17 +296,11 @@ func (t *expectEvents) expect(events ...*Event) {
 				if event.Conn == nil {
 					t.Fatal("expected event.Conn to be set")
 				}
-				if event.Conn.One == nil {
+				if event.Conn.One != expected.Conn.One {
-					t.Fatal("expected event.Conn.One to be set")
+					t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.TerminalString(), event.Conn.One.TerminalString())
 				}
-				if event.Conn.Other == nil {
+				if event.Conn.Other != expected.Conn.Other {
-					t.Fatal("expected event.Conn.Other to be set")
+					t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.TerminalString(), event.Conn.Other.TerminalString())
-				}
-				if event.Conn.One.NodeID != expected.Conn.One.NodeID {
-					t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.Label(), event.Conn.One.Label())
-				}
-				if event.Conn.Other.NodeID != expected.Conn.Other.NodeID {
-					t.Fatalf("expected conn event %d to have other=%q, got other=%q", i, expected.Conn.Other.Label(), event.Conn.Other.Label())
 				}
 				if event.Conn.Up != expected.Conn.Up {
 					t.Fatalf("expected conn event %d to have up=%t, got up=%t", i, expected.Conn.Up, event.Conn.Up)
@@ -335,7 +328,7 @@ func TestHTTPNodeRPC(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating network: %s", err)
 	}
-	node, err := client.CreateNode(network.ID, &adapters.NodeConfig{})
+	node, err := client.CreateNode(network.ID, nil)
 	if err != nil {
 		t.Fatalf("error creating node: %s", err)
 	}
@@ -401,7 +394,7 @@ func TestHTTPSnapshot(t *testing.T) {
 	nodeCount := 2
 	nodes := make([]*p2p.NodeInfo, nodeCount)
 	for i := 0; i < nodeCount; i++ {
-		node, err := client.CreateNode(network.ID, &adapters.NodeConfig{})
+		node, err := client.CreateNode(network.ID, nil)
 		if err != nil {
 			t.Fatalf("error creating node: %s", err)
 		}
@@ -435,8 +428,9 @@ func TestHTTPSnapshot(t *testing.T) {
 		t.Fatalf("error creating snapshot: %s", err)
 	}
 	for i, state := range states {
-		if string(snap.Nodes[i].Snapshot) != state {
+		gotState := snap.Nodes[i].Snapshots["test"]
-			t.Fatalf("expected snapshot state %q, got %q", state, snap.Nodes[i].Snapshot)
+		if string(gotState) != state {
+			t.Fatalf("expected snapshot state %q, got %q", state, gotState)
 		}
 	}
 

p2p/testing/protocoltester.go

@@ -21,11 +21,11 @@ type ProtocolTester struct {
 
 func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Peer, p2p.MsgReadWriter) error) *ProtocolTester {
 	services := adapters.Services{
-		"test": func(id discover.NodeID, _ []byte) node.Service {
+		"test": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return &testNode{run}
+			return &testNode{run}, nil
 		},
-		"mock": func(id discover.NodeID, _ []byte) node.Service {
+		"mock": func(ctx *adapters.ServiceContext) (node.Service, error) {
-			return newMockNode()
+			return newMockNode(), nil
 		},
 	}
 	adapter := adapters.NewSimAdapter(services)

swarm/network/simulations/discovery/discovery_test.go

@@ -179,8 +179,8 @@ func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id di
 	return nil
 }
 
-func newService(id discover.NodeID, snapshot []byte) node.Service {
+func newService(ctx *adapters.ServiceContext) (node.Service, error) {
-	addr := network.NewAddrFromNodeID(id)
+	addr := network.NewAddrFromNodeID(ctx.Config.ID)
 
 	kp := network.NewKadParams()
 	kp.MinProxBinSize = testMinProxBinSize
@@ -200,5 +200,5 @@ func newService(id discover.NodeID, snapshot []byte) node.Service {
 		HiveParams:   hp,
 	}
 
-	return network.NewBzz(config, kad, nil)
+	return network.NewBzz(config, kad, nil), nil
 }

swarm/network/simulations/overlay.go

@@ -33,7 +33,8 @@ func NewSimulation() *Simulation {
 	}
 }
 
-func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Service {
+func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, error) {
+	id := ctx.Config.ID
 	s.mtx.Lock()
 	store, ok := s.stores[id]
 	if !ok {
@@ -64,7 +65,7 @@ func (s *Simulation) NewService(id discover.NodeID, snapshot []byte) node.Servic
 		HiveParams:   hp,
 	}
 
-	return network.NewBzz(config, kad, store)
+	return network.NewBzz(config, kad, store), nil
 }
 
 func createMockers() map[string]*simulations.MockerConfig {

swarm/pss/pss_test.go

@@ -453,16 +453,15 @@ func newServices() adapters.Services {
 		return kademlias[id]
 	}
 	return adapters.Services{
-		"pss": func(id discover.NodeID, snapshot []byte) node.Service {
+		"pss": func(ctx *adapters.ServiceContext) (node.Service, error) {
+			id := ctx.Config.ID
 			cachedir, err := ioutil.TempDir("", "pss-cache")
 			if err != nil {
-				log.Error("create pss cache tmpdir failed", "error", err)
+				return nil, err
-				return nil
 			}
 			dpa, err := storage.NewLocalDPA(cachedir)
 			if err != nil {
-				log.Error("local dpa creation failed", "error", err)
+				return nil, err
-				return nil
 			}
 
 			pssp := NewPssParams()
@@ -473,20 +472,20 @@ func newServices() adapters.Services {
 			}
 			err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler))
 			if err != nil {
-				log.Error("Couldnt register pss protocol", "err", err)
+				return nil, err
-				os.Exit(1)
 			}
 
-			return ps
+			return ps, nil
 		},
-		"bzz": func(id discover.NodeID, snapshot []byte) node.Service {
+		"bzz": func(ctx *adapters.ServiceContext) (node.Service, error) {
+			id := ctx.Config.ID
 			addr := network.NewAddrFromNodeID(id)
 			config := &network.BzzConfig{
 				OverlayAddr:  addr.Over(),
 				UnderlayAddr: addr.Under(),
 				HiveParams:   network.NewHiveParams(),
 			}
-			return network.NewBzz(config, kademlia(id), stateStore)
+			return network.NewBzz(config, kademlia(id), stateStore), nil
 		},
 	}
 }