diff --git a/node/api.go b/node/api.go
index 1b2ef2ed90..1b04b70938 100644
--- a/node/api.go
+++ b/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
diff --git a/node/config.go b/node/config.go
index 61e0008ef9..7021eaadf0 100644
--- a/node/config.go
+++ b/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
diff --git a/node/node.go b/node/node.go
index a89110599f..854efd28cb 100644
--- a/node/node.go
+++ b/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
diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go
index f18f4c96f5..319e4353a2 100644
--- a/p2p/simulations/adapters/exec.go
+++ b/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,
- Dir: dir,
- Config: conf,
+ ID: config.ID,
+ Dir: dir,
+ 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
- rpcMux *rpcMux
- newCmd func() *exec.Cmd
- key *ecdsa.PrivateKey
+ adapter *ExecAdapter
+ client *rpc.Client
+ wsAddr string
+ 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
- pipe1, pipe2 := net.Pipe()
+ // use a pipe for stderr so we can both copy the node's stderr to
+ // 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 = pipe1
- cmd.Stderr = os.Stderr
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = 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
-// RPC multiplexer
-func (n *ExecNode) ServeRPC(conn net.Conn) error {
- if n.rpcMux == nil {
- return errors.New("RPC not started")
+// ServeRPC serves RPC requests over the given connection by dialling the
+// node's WebSocket address and joining the two connections
+func (n *ExecNode) ServeRPC(clientConn net.Conn) error {
+ conn, err := websocket.Dial(n.wsAddr, "", "http://localhost")
+ 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
- stack, err := startP2PNode(&conf.Stack, services)
+ // initialize the devp2p stack
+ 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
- // either the local filesystem or TCP stack (useful when running in
- // Docker)
- handler, err := stack.RPCHandler()
- if err != nil {
- log.Crit("error getting RPC server", "err", err)
+ // register the services, collecting them into a map so we can wrap
+ // them in a snapshot service
+ services := make(map[string]node.Service, len(serviceNames))
+ for _, name := range serviceNames {
+ serviceFunc, exists := serviceFuncs[name]
+ 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
-// use stdio for RPC messages
-type stdioConn struct {
- io.Reader
- io.Writer
+type wsRPCDialer struct {
+ addrs map[string]string
}
-func (r *stdioConn) Close() error {
- return nil
+func (w *wsRPCDialer) DialRPC(id discover.NodeID) (*rpc.Client, error) {
+ addr, ok := w.addrs[id.String()]
+ if !ok {
+ return nil, fmt.Errorf("unknown node: %s", id)
+ }
+ return rpc.DialWebsocket(context.Background(), addr, "http://localhost")
}
diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go
index 66b209dddb..f81611a470 100644
--- a/p2p/simulations/adapters/inproc.go
+++ b/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()
- defer self.lock.Unlock()
+ self.lock.RLock()
+ 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
-// RPC multiplexer
+// ServeRPC serves RPC requests over the given connection by creating an
+// in-process client to the node's RPC server
func (self *SimNode) ServeRPC(conn net.Conn) error {
- self.lock.Lock()
- mux := self.rpcMux
- self.lock.Unlock()
- if mux == nil {
- return errors.New("RPC not started")
+ self.lock.RLock()
+ node := self.node
+ self.lock.RUnlock()
+ if node == nil {
+ 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()
- if self.client == nil {
- return nil, errors.New("RPC not started")
+ services := self.running
+ self.lock.Unlock()
+ if len(services) == 0 {
+ return nil, errors.New("no running services")
}
- var snapshots map[string][]byte
- return snapshots, self.client.Call(&snapshots, "simulation_snapshot")
+ snapshots := make(map[string][]byte)
+ 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) {
- var snapshot []byte
+ return func(nodeCtx *node.ServiceContext) (node.Service, error) {
+ 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)
- self.running = append(self.running, service)
+ service, err := serviceFunc(ctx)
+ 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
}
-
- // 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.client = rpc.DialInProc(handler)
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 {
diff --git a/p2p/simulations/adapters/rpc_mux.go b/p2p/simulations/adapters/rpc_mux.go
deleted file mode 100644
index 6d2c5b58ae..0000000000
--- a/p2p/simulations/adapters/rpc_mux.go
+++ /dev/null
@@ -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 .
-
-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
-}
diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go
index 034d162297..f5ac3ec4ac 100644
--- a/p2p/simulations/adapters/types.go
+++ b/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
diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go
index 6174c094e8..95b7c8be7e 100644
--- a/p2p/simulations/examples/connectivity.go
+++ b/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 {
- return newPingPongService(id)
+ "ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) {
+ return newPingPongService(ctx.Config.ID), nil
},
}
diff --git a/p2p/simulations/http.go b/p2p/simulations/http.go
index 1b86fca438..0d1742cfaf 100644
--- a/p2p/simulations/http.go
+++ b/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//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)
diff --git a/p2p/simulations/http_test.go b/p2p/simulations/http_test.go
index fa55bd29f3..84ceb0f677 100644
--- a/p2p/simulations/http_test.go
+++ b/p2p/simulations/http_test.go
@@ -24,10 +24,10 @@ type testService struct {
state atomic.Value
}
-func newTestService(id discover.NodeID, snapshot []byte) node.Service {
- svc := &testService{id: id}
- svc.state.Store(snapshot)
- return svc
+func newTestService(ctx *adapters.ServiceContext) (node.Service, error) {
+ svc := &testService{id: ctx.Config.ID}
+ svc.state.Store(ctx.Snapshot)
+ 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, config)
+ node, err := client.CreateNode(network.ID, nil)
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 {
- t.Fatalf("expected node event %d to have id %q, got %q", i, expected.Node.ID().Label(), event.Node.ID().Label())
+ if event.Node.ID() != expected.Node.ID() {
+ 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 {
- t.Fatal("expected event.Conn.One to be set")
+ if event.Conn.One != expected.Conn.One {
+ 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 {
- t.Fatal("expected event.Conn.Other to be set")
- }
- 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.Other != expected.Conn.Other {
+ 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.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 {
- t.Fatalf("expected snapshot state %q, got %q", state, snap.Nodes[i].Snapshot)
+ gotState := snap.Nodes[i].Snapshots["test"]
+ if string(gotState) != state {
+ t.Fatalf("expected snapshot state %q, got %q", state, gotState)
}
}
diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go
index bbaa18a8ae..9e58a79d18 100644
--- a/p2p/testing/protocoltester.go
+++ b/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 {
- return &testNode{run}
+ "test": func(ctx *adapters.ServiceContext) (node.Service, error) {
+ return &testNode{run}, nil
},
- "mock": func(id discover.NodeID, _ []byte) node.Service {
- return newMockNode()
+ "mock": func(ctx *adapters.ServiceContext) (node.Service, error) {
+ return newMockNode(), nil
},
}
adapter := adapters.NewSimAdapter(services)
diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go
index 13d1f1fbef..fb038ed2a7 100644
--- a/swarm/network/simulations/discovery/discovery_test.go
+++ b/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 {
- addr := network.NewAddrFromNodeID(id)
+func newService(ctx *adapters.ServiceContext) (node.Service, error) {
+ 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
}
diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go
index dbf3ddb867..319eeec5c4 100644
--- a/swarm/network/simulations/overlay.go
+++ b/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 {
diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go
index f540d6aa3e..3b400ade20 100644
--- a/swarm/pss/pss_test.go
+++ b/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
+ return nil, err
}
dpa, err := storage.NewLocalDPA(cachedir)
if err != nil {
- log.Error("local dpa creation failed", "error", err)
- return nil
+ return nil, err
}
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)
- os.Exit(1)
+ return nil, err
}
- 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
},
}
}