p2p: add network simulation framework (#14982)

This commit introduces a network simulation framework which
can be used to run simulated networks of devp2p nodes. The
intention is to use this for testing protocols, performing
benchmarks and visualising emergent network behaviour.

cmd/p2psim/main.go

+// p2psim provides a command-line client for a simulation HTTP API.
+//
+// Here is an example of creating a 2 node network with the first node
+// connected to the second:
+//
+//     $ p2psim node create
+//     Created node01
+//
+//     $ p2psim node start node01
+//     Started node01
+//
+//     $ p2psim node create
+//     Created node02
+//
+//     $ p2psim node start node02
+//     Started node02
+//
+//     $ p2psim node connect node01 node02
+//     Connected node01 to node02
+//
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"text/tabwriter"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/simulations"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/rpc"
+	"gopkg.in/urfave/cli.v1"
+)
+
+var client *simulations.Client
+
+func main() {
+	app := cli.NewApp()
+	app.Usage = "devp2p simulation command-line client"
+	app.Flags = []cli.Flag{
+		cli.StringFlag{
+			Name:   "api",
+			Value:  "http://localhost:8888",
+			Usage:  "simulation API URL",
+			EnvVar: "P2PSIM_API_URL",
+		},
+	}
+	app.Before = func(ctx *cli.Context) error {
+		client = simulations.NewClient(ctx.GlobalString("api"))
+		return nil
+	}
+	app.Commands = []cli.Command{
+		{
+			Name:   "show",
+			Usage:  "show network information",
+			Action: showNetwork,
+		},
+		{
+			Name:   "events",
+			Usage:  "stream network events",
+			Action: streamNetwork,
+			Flags: []cli.Flag{
+				cli.BoolFlag{
+					Name:  "current",
+					Usage: "get existing nodes and conns first",
+				},
+				cli.StringFlag{
+					Name:  "filter",
+					Value: "",
+					Usage: "message filter",
+				},
+			},
+		},
+		{
+			Name:   "snapshot",
+			Usage:  "create a network snapshot to stdout",
+			Action: createSnapshot,
+		},
+		{
+			Name:   "load",
+			Usage:  "load a network snapshot from stdin",
+			Action: loadSnapshot,
+		},
+		{
+			Name:   "node",
+			Usage:  "manage simulation nodes",
+			Action: listNodes,
+			Subcommands: []cli.Command{
+				{
+					Name:   "list",
+					Usage:  "list nodes",
+					Action: listNodes,
+				},
+				{
+					Name:   "create",
+					Usage:  "create a node",
+					Action: createNode,
+					Flags: []cli.Flag{
+						cli.StringFlag{
+							Name:  "name",
+							Value: "",
+							Usage: "node name",
+						},
+						cli.StringFlag{
+							Name:  "services",
+							Value: "",
+							Usage: "node services (comma separated)",
+						},
+						cli.StringFlag{
+							Name:  "key",
+							Value: "",
+							Usage: "node private key (hex encoded)",
+						},
+					},
+				},
+				{
+					Name:      "show",
+					ArgsUsage: "<node>",
+					Usage:     "show node information",
+					Action:    showNode,
+				},
+				{
+					Name:      "start",
+					ArgsUsage: "<node>",
+					Usage:     "start a node",
+					Action:    startNode,
+				},
+				{
+					Name:      "stop",
+					ArgsUsage: "<node>",
+					Usage:     "stop a node",
+					Action:    stopNode,
+				},
+				{
+					Name:      "connect",
+					ArgsUsage: "<node> <peer>",
+					Usage:     "connect a node to a peer node",
+					Action:    connectNode,
+				},
+				{
+					Name:      "disconnect",
+					ArgsUsage: "<node> <peer>",
+					Usage:     "disconnect a node from a peer node",
+					Action:    disconnectNode,
+				},
+				{
+					Name:      "rpc",
+					ArgsUsage: "<node> <method> [<args>]",
+					Usage:     "call a node RPC method",
+					Action:    rpcNode,
+					Flags: []cli.Flag{
+						cli.BoolFlag{
+							Name:  "subscribe",
+							Usage: "method is a subscription",
+						},
+					},
+				},
+			},
+		},
+	}
+	app.Run(os.Args)
+}
+
+func showNetwork(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	network, err := client.GetNetwork()
+	if err != nil {
+		return err
+	}
+	w := tabwriter.NewWriter(ctx.App.Writer, 1, 2, 2, ' ', 0)
+	defer w.Flush()
+	fmt.Fprintf(w, "NODES\t%d\n", len(network.Nodes))
+	fmt.Fprintf(w, "CONNS\t%d\n", len(network.Conns))
+	return nil
+}
+
+func streamNetwork(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	events := make(chan *simulations.Event)
+	sub, err := client.SubscribeNetwork(events, simulations.SubscribeOpts{
+		Current: ctx.Bool("current"),
+		Filter:  ctx.String("filter"),
+	})
+	if err != nil {
+		return err
+	}
+	defer sub.Unsubscribe()
+	enc := json.NewEncoder(ctx.App.Writer)
+	for {
+		select {
+		case event := <-events:
+			if err := enc.Encode(event); err != nil {
+				return err
+			}
+		case err := <-sub.Err():
+			return err
+		}
+	}
+}
+
+func createSnapshot(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	snap, err := client.CreateSnapshot()
+	if err != nil {
+		return err
+	}
+	return json.NewEncoder(os.Stdout).Encode(snap)
+}
+
+func loadSnapshot(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	snap := &simulations.Snapshot{}
+	if err := json.NewDecoder(os.Stdin).Decode(snap); err != nil {
+		return err
+	}
+	return client.LoadSnapshot(snap)
+}
+
+func listNodes(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodes, err := client.GetNodes()
+	if err != nil {
+		return err
+	}
+	w := tabwriter.NewWriter(ctx.App.Writer, 1, 2, 2, ' ', 0)
+	defer w.Flush()
+	fmt.Fprintf(w, "NAME\tPROTOCOLS\tID\n")
+	for _, node := range nodes {
+		fmt.Fprintf(w, "%s\t%s\t%s\n", node.Name, strings.Join(protocolList(node), ","), node.ID)
+	}
+	return nil
+}
+
+func protocolList(node *p2p.NodeInfo) []string {
+	protos := make([]string, 0, len(node.Protocols))
+	for name := range node.Protocols {
+		protos = append(protos, name)
+	}
+	return protos
+}
+
+func createNode(ctx *cli.Context) error {
+	if len(ctx.Args()) != 0 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	config := &adapters.NodeConfig{
+		Name: ctx.String("name"),
+	}
+	if key := ctx.String("key"); key != "" {
+		privKey, err := crypto.HexToECDSA(key)
+		if err != nil {
+			return err
+		}
+		config.ID = discover.PubkeyID(&privKey.PublicKey)
+		config.PrivateKey = privKey
+	}
+	if services := ctx.String("services"); services != "" {
+		config.Services = strings.Split(services, ",")
+	}
+	node, err := client.CreateNode(config)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintln(ctx.App.Writer, "Created", node.Name)
+	return nil
+}
+
+func showNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 1 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	node, err := client.GetNode(nodeName)
+	if err != nil {
+		return err
+	}
+	w := tabwriter.NewWriter(ctx.App.Writer, 1, 2, 2, ' ', 0)
+	defer w.Flush()
+	fmt.Fprintf(w, "NAME\t%s\n", node.Name)
+	fmt.Fprintf(w, "PROTOCOLS\t%s\n", strings.Join(protocolList(node), ","))
+	fmt.Fprintf(w, "ID\t%s\n", node.ID)
+	fmt.Fprintf(w, "ENODE\t%s\n", node.Enode)
+	for name, proto := range node.Protocols {
+		fmt.Fprintln(w)
+		fmt.Fprintf(w, "--- PROTOCOL INFO: %s\n", name)
+		fmt.Fprintf(w, "%v\n", proto)
+		fmt.Fprintf(w, "---\n")
+	}
+	return nil
+}
+
+func startNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 1 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	if err := client.StartNode(nodeName); err != nil {
+		return err
+	}
+	fmt.Fprintln(ctx.App.Writer, "Started", nodeName)
+	return nil
+}
+
+func stopNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 1 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	if err := client.StopNode(nodeName); err != nil {
+		return err
+	}
+	fmt.Fprintln(ctx.App.Writer, "Stopped", nodeName)
+	return nil
+}
+
+func connectNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 2 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	peerName := args[1]
+	if err := client.ConnectNode(nodeName, peerName); err != nil {
+		return err
+	}
+	fmt.Fprintln(ctx.App.Writer, "Connected", nodeName, "to", peerName)
+	return nil
+}
+
+func disconnectNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 2 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	peerName := args[1]
+	if err := client.DisconnectNode(nodeName, peerName); err != nil {
+		return err
+	}
+	fmt.Fprintln(ctx.App.Writer, "Disconnected", nodeName, "from", peerName)
+	return nil
+}
+
+func rpcNode(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) < 2 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	nodeName := args[0]
+	method := args[1]
+	rpcClient, err := client.RPCClient(context.Background(), nodeName)
+	if err != nil {
+		return err
+	}
+	if ctx.Bool("subscribe") {
+		return rpcSubscribe(rpcClient, ctx.App.Writer, method, args[3:]...)
+	}
+	var result interface{}
+	params := make([]interface{}, len(args[3:]))
+	for i, v := range args[3:] {
+		params[i] = v
+	}
+	if err := rpcClient.Call(&result, method, params...); err != nil {
+		return err
+	}
+	return json.NewEncoder(ctx.App.Writer).Encode(result)
+}
+
+func rpcSubscribe(client *rpc.Client, out io.Writer, method string, args ...string) error {
+	parts := strings.SplitN(method, "_", 2)
+	namespace := parts[0]
+	method = parts[1]
+	ch := make(chan interface{})
+	subArgs := make([]interface{}, len(args)+1)
+	subArgs[0] = method
+	for i, v := range args {
+		subArgs[i+1] = v
+	}
+	sub, err := client.Subscribe(context.Background(), namespace, ch, subArgs...)
+	if err != nil {
+		return err
+	}
+	defer sub.Unsubscribe()
+	enc := json.NewEncoder(out)
+	for {
+		select {
+		case v := <-ch:
+			if err := enc.Encode(v); err != nil {
+				return err
+			}
+		case err := <-sub.Err():
+			return err
+		}
+	}
+}

node/api.go

@@ -17,6 +17,7 @@
 package node
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -25,6 +26,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"
 	"github.com/rcrowley/go-metrics"
 )
 
@@ -73,6 +75,44 @@ func (api *PrivateAdminAPI) RemovePeer(url string) (bool, error) {
 	return true, nil
 }
 
+// PeerEvents creates an RPC subscription which receives peer events from the
+// node's p2p.Server
+func (api *PrivateAdminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) {
+	// Make sure the server is running, fail otherwise
+	server := api.node.Server()
+	if server == nil {
+		return nil, ErrNodeStopped
+	}
+
+	// Create the subscription
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return nil, rpc.ErrNotificationsUnsupported
+	}
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		events := make(chan *p2p.PeerEvent)
+		sub := server.SubscribeEvents(events)
+		defer sub.Unsubscribe()
+
+		for {
+			select {
+			case event := <-events:
+				notifier.Notify(rpcSub.ID, event)
+			case <-sub.Err():
+				return
+			case <-rpcSub.Err():
+				return
+			case <-notifier.Closed():
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
 // StartRPC starts the HTTP RPC API server.
 func (api *PrivateAdminAPI) StartRPC(host *string, port *int, cors *string, apis *string) (bool, error) {
 	api.node.lock.Lock()
@@ -163,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

@@ -261,7 +261,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()
@@ -412,7 +412,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
@@ -425,7 +425,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
 			}
@@ -441,7 +441,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
@@ -556,6 +556,17 @@ func (n *Node) Attach() (*rpc.Client, error) {
 	return rpc.DialInProc(n.inprocHandler), nil
 }
 
+// RPCHandler returns the in-process RPC request handler.
+func (n *Node) RPCHandler() (*rpc.Server, error) {
+	n.lock.RLock()
+	defer n.lock.RUnlock()
+
+	if n.inprocHandler == nil {
+		return nil, ErrNodeStopped
+	}
+	return n.inprocHandler, nil
+}
+
 // 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.

p2p/dial.go

@@ -47,6 +47,24 @@ const (
 	maxResolveDelay     = time.Hour
 )
 
+// NodeDialer is used to connect to nodes in the network, typically by using
+// an underlying net.Dialer but also using net.Pipe in tests
+type NodeDialer interface {
+	Dial(*discover.Node) (net.Conn, error)
+}
+
+// TCPDialer implements the NodeDialer interface by using a net.Dialer to
+// create TCP connections to nodes in the network
+type TCPDialer struct {
+	*net.Dialer
+}
+
+// Dial creates a TCP connection to the node
+func (t TCPDialer) Dial(dest *discover.Node) (net.Conn, error) {
+	addr := &net.TCPAddr{IP: dest.IP, Port: int(dest.TCP)}
+	return t.Dialer.Dial("tcp", addr.String())
+}
+
 // 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.
@@ -318,14 +336,13 @@ func (t *dialTask) resolve(srv *Server) bool {
 
 // dial performs the actual connection attempt.
 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())
+	fd, err := srv.Dialer.Dial(dest)
 	if err != nil {
 		log.Trace("Dial error", "task", t, "err", err)
 		return false
 	}
 	mfd := newMeteredConn(fd, false)
-	srv.setupConn(mfd, t.flags, dest)
+	srv.SetupConn(mfd, t.flags, dest)
 	return true
 }
 

p2p/dial_test.go

@@ -597,7 +597,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)}}
+	config := Config{Dialer: TCPDialer{&net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}}}
 	srv := &Server{ntab: table, Config: config}
 	tasks[0].Do(srv)
 	if !reflect.DeepEqual(table.resolveCalls, []discover.NodeID{dest.ID}) {

p2p/discover/node.go

@@ -225,6 +225,11 @@ func (n *Node) UnmarshalText(text []byte) error {
 // The node identifier is a marshaled elliptic curve public key.
 type NodeID [NodeIDBits / 8]byte
 
+// Bytes returns a byte slice representation of the NodeID
+func (n NodeID) Bytes() []byte {
+	return n[:]
+}
+
 // NodeID prints as a long hexadecimal number.
 func (n NodeID) String() string {
 	return fmt.Sprintf("%x", n[:])
@@ -240,6 +245,41 @@ func (n NodeID) TerminalString() string {
 	return hex.EncodeToString(n[:8])
 }
 
+// MarshalText implements the encoding.TextMarshaler interface.
+func (n NodeID) MarshalText() ([]byte, error) {
+	return []byte(hex.EncodeToString(n[:])), nil
+}
+
+// UnmarshalText implements the encoding.TextUnmarshaler interface.
+func (n *NodeID) UnmarshalText(text []byte) error {
+	id, err := HexID(string(text))
+	if err != nil {
+		return err
+	}
+	*n = id
+	return nil
+}
+
+// BytesID converts a byte slice to a NodeID
+func BytesID(b []byte) (NodeID, error) {
+	var id NodeID
+	if len(b) != len(id) {
+		return id, fmt.Errorf("wrong length, want %d bytes", len(id))
+	}
+	copy(id[:], b)
+	return id, nil
+}
+
+// MustBytesID converts a byte slice to a NodeID.
+// It panics if the byte slice is not a valid NodeID.
+func MustBytesID(b []byte) NodeID {
+	id, err := BytesID(b)
+	if err != nil {
+		panic(err)
+	}
+	return id
+}
+
 // HexID converts a hex string to a NodeID.
 // The string may be prefixed with 0x.
 func HexID(in string) (NodeID, error) {

p2p/discover/node_test.go

@@ -17,6 +17,7 @@
 package discover
 
 import (
+	"bytes"
 	"fmt"
 	"math/big"
 	"math/rand"
@@ -192,6 +193,35 @@ func TestHexID(t *testing.T) {
 	}
 }
 
+func TestNodeID_textEncoding(t *testing.T) {
+	ref := NodeID{
+		0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x10,
+		0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x20,
+		0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x29, 0x30,
+		0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x40,
+		0x41, 0x42, 0x43, 0x44, 0x45, 0x46, 0x47, 0x48, 0x49, 0x50,
+		0x51, 0x52, 0x53, 0x54, 0x55, 0x56, 0x57, 0x58, 0x59, 0x60,
+		0x61, 0x62, 0x63, 0x64,
+	}
+	hex := "01020304050607080910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364"
+
+	text, err := ref.MarshalText()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(text, []byte(hex)) {
+		t.Fatalf("text encoding did not match\nexpected: %s\ngot:      %s", hex, text)
+	}
+
+	id := new(NodeID)
+	if err := id.UnmarshalText(text); err != nil {
+		t.Fatal(err)
+	}
+	if *id != ref {
+		t.Fatalf("text decoding did not match\nexpected: %s\ngot:      %s", ref, id)
+	}
+}
+
 func TestNodeID_recover(t *testing.T) {
 	prv := newkey()
 	hash := make([]byte, 32)

p2p/message.go

@@ -27,6 +27,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
@@ -271,3 +273,67 @@ func ExpectMsg(r MsgReader, code uint64, content interface{}) error {
 	}
 	return nil
 }
+
+// msgEventer wraps a MsgReadWriter and sends events whenever a message is sent
+// or received
+type msgEventer struct {
+	MsgReadWriter
+
+	feed     *event.Feed
+	peerID   discover.NodeID
+	Protocol string
+}
+
+// newMsgEventer returns a msgEventer which sends message events to the given
+// feed
+func newMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID discover.NodeID, proto string) *msgEventer {
+	return &msgEventer{
+		MsgReadWriter: rw,
+		feed:          feed,
+		peerID:        peerID,
+		Protocol:      proto,
+	}
+}
+
+// ReadMsg reads a message from the underlying MsgReadWriter and emits a
+// "message received" event
+func (self *msgEventer) ReadMsg() (Msg, error) {
+	msg, err := self.MsgReadWriter.ReadMsg()
+	if err != nil {
+		return msg, err
+	}
+	self.feed.Send(&PeerEvent{
+		Type:     PeerEventTypeMsgRecv,
+		Peer:     self.peerID,
+		Protocol: self.Protocol,
+		MsgCode:  &msg.Code,
+		MsgSize:  &msg.Size,
+	})
+	return msg, nil
+}
+
+// WriteMsg writes a message to the underlying MsgReadWriter and emits a
+// "message sent" event
+func (self *msgEventer) WriteMsg(msg Msg) error {
+	err := self.MsgReadWriter.WriteMsg(msg)
+	if err != nil {
+		return err
+	}
+	self.feed.Send(&PeerEvent{
+		Type:     PeerEventTypeMsgSend,
+		Peer:     self.peerID,
+		Protocol: self.Protocol,
+		MsgCode:  &msg.Code,
+		MsgSize:  &msg.Size,
+	})
+	return nil
+}
+
+// Close closes the underlying MsgReadWriter if it implements the io.Closer
+// interface
+func (self *msgEventer) Close() error {
+	if v, ok := self.MsgReadWriter.(io.Closer); ok {
+		return v.Close()
+	}
+	return nil
+}

p2p/peer.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -60,6 +61,38 @@ type protoHandshake struct {
 	Rest []rlp.RawValue `rlp:"tail"`
 }
 
+// PeerEventType is the type of peer events emitted by a p2p.Server
+type PeerEventType string
+
+const (
+	// PeerEventTypeAdd is the type of event emitted when a peer is added
+	// to a p2p.Server
+	PeerEventTypeAdd PeerEventType = "add"
+
+	// PeerEventTypeDrop is the type of event emitted when a peer is
+	// dropped from a p2p.Server
+	PeerEventTypeDrop PeerEventType = "drop"
+
+	// PeerEventTypeMsgSend is the type of event emitted when a
+	// message is successfully sent to a peer
+	PeerEventTypeMsgSend PeerEventType = "msgsend"
+
+	// PeerEventTypeMsgRecv is the type of event emitted when a
+	// message is received from a peer
+	PeerEventTypeMsgRecv PeerEventType = "msgrecv"
+)
+
+// PeerEvent is an event emitted when peers are either added or dropped from
+// a p2p.Server or when a message is sent or received on a peer connection
+type PeerEvent struct {
+	Type     PeerEventType   `json:"type"`
+	Peer     discover.NodeID `json:"peer"`
+	Error    string          `json:"error,omitempty"`
+	Protocol string          `json:"protocol,omitempty"`
+	MsgCode  *uint64         `json:"msg_code,omitempty"`
+	MsgSize  *uint32         `json:"msg_size,omitempty"`
+}
+
 // Peer represents a connected remote node.
 type Peer struct {
 	rw      *conn
@@ -71,6 +104,9 @@ type Peer struct {
 	protoErr chan error
 	closed   chan struct{}
 	disc     chan DiscReason
+
+	// events receives message send / receive events if set
+	events *event.Feed
 }
 
 // NewPeer returns a peer for testing purposes.
@@ -297,9 +333,13 @@ func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error)
 		proto.closed = p.closed
 		proto.wstart = writeStart
 		proto.werr = writeErr
+		var rw MsgReadWriter = proto
+		if p.events != nil {
+			rw = newMsgEventer(rw, p.events, p.ID(), proto.Name)
+		}
 		p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
 		go func() {
-			err := proto.Run(p, proto)
+			err := proto.Run(p, rw)
 			if err == nil {
 				p.log.Trace(fmt.Sprintf("Protocol %s/%d returned", proto.Name, proto.Version))
 				err = errProtocolReturned

p2p/server.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/discv5"
@@ -130,10 +131,14 @@ type Config struct {
 
 	// If Dialer is set to a non-nil value, the given Dialer
 	// is used to dial outbound peer connections.
-	Dialer *net.Dialer `toml:"-"`
+	Dialer NodeDialer `toml:"-"`
 
 	// If NoDial is true, the server will not dial any peers.
 	NoDial bool `toml:",omitempty"`
+
+	// If EnableMsgEvents is set then the server will emit PeerEvents
+	// whenever a message is sent to or received from a peer
+	EnableMsgEvents bool
 }
 
 // Server manages all peer connections.
@@ -166,6 +171,7 @@ type Server struct {
 	addpeer       chan *conn
 	delpeer       chan peerDrop
 	loopWG        sync.WaitGroup // loop, listenLoop
+	peerFeed      event.Feed
 }
 
 type peerOpFunc func(map[discover.NodeID]*Peer)
@@ -191,7 +197,7 @@ type conn struct {
 	fd net.Conn
 	transport
 	flags connFlag
-	cont  chan error      // The run loop uses cont to signal errors to setupConn.
+	cont  chan error      // The run loop uses cont to signal errors to SetupConn.
 	id    discover.NodeID // valid after the encryption handshake
 	caps  []Cap           // valid after the protocol handshake
 	name  string          // valid after the protocol handshake
@@ -291,6 +297,11 @@ func (srv *Server) RemovePeer(node *discover.Node) {
 	}
 }
 
+// SubscribePeers subscribes the given channel to peer events
+func (srv *Server) SubscribeEvents(ch chan *PeerEvent) event.Subscription {
+	return srv.peerFeed.Subscribe(ch)
+}
+
 // Self returns the local node's endpoint information.
 func (srv *Server) Self() *discover.Node {
 	srv.lock.Lock()
@@ -358,7 +369,7 @@ func (srv *Server) Start() (err error) {
 		srv.newTransport = newRLPX
 	}
 	if srv.Dialer == nil {
-		srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout}
+		srv.Dialer = TCPDialer{&net.Dialer{Timeout: defaultDialTimeout}}
 	}
 	srv.quit = make(chan struct{})
 	srv.addpeer = make(chan *conn)
@@ -536,7 +547,11 @@ running:
 				c.flags |= trustedConn
 			}
 			// TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
-			c.cont <- srv.encHandshakeChecks(peers, c)
+			select {
+			case c.cont <- srv.encHandshakeChecks(peers, c):
+			case <-srv.quit:
+				break running
+			}
 		case c := <-srv.addpeer:
 			// At this point the connection is past the protocol handshake.
 			// Its capabilities are known and the remote identity is verified.
@@ -544,6 +559,11 @@ running:
 			if err == nil {
 				// The handshakes are done and it passed all checks.
 				p := newPeer(c, srv.Protocols)
+				// If message events are enabled, pass the peerFeed
+				// to the peer
+				if srv.EnableMsgEvents {
+					p.events = &srv.peerFeed
+				}
 				name := truncateName(c.name)
 				log.Debug("Adding p2p peer", "id", c.id, "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
 				peers[c.id] = p
@@ -552,7 +572,11 @@ running:
 			// The dialer logic relies on the assumption that
 			// dial tasks complete after the peer has been added or
 			// discarded. Unblock the task last.
-			c.cont <- err
+			select {
+			case c.cont <- err:
+			case <-srv.quit:
+				break running
+			}
 		case pd := <-srv.delpeer:
 			// A peer disconnected.
 			d := common.PrettyDuration(mclock.Now() - pd.created)
@@ -665,16 +689,16 @@ func (srv *Server) listenLoop() {
 		// Spawn the handler. It will give the slot back when the connection
 		// has been established.
 		go func() {
-			srv.setupConn(fd, inboundConn, nil)
+			srv.SetupConn(fd, inboundConn, nil)
 			slots <- struct{}{}
 		}()
 	}
 }
 
-// setupConn runs the handshakes and attempts to add the connection
+// 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
@@ -755,7 +779,23 @@ func (srv *Server) runPeer(p *Peer) {
 	if srv.newPeerHook != nil {
 		srv.newPeerHook(p)
 	}
+
+	// broadcast peer add
+	srv.peerFeed.Send(&PeerEvent{
+		Type: PeerEventTypeAdd,
+		Peer: p.ID(),
+	})
+
+	// run the protocol
 	remoteRequested, err := p.run()
+
+	// broadcast peer drop
+	srv.peerFeed.Send(&PeerEvent{
+		Type:  PeerEventTypeDrop,
+		Peer:  p.ID(),
+		Error: err.Error(),
+	})
+
 	// 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}

p2p/server_test.go

@@ -435,7 +435,7 @@ func TestServerSetupConn(t *testing.T) {
 			}
 		}
 		p1, _ := net.Pipe()
-		srv.setupConn(p1, test.flags, test.dialDest)
+		srv.SetupConn(p1, test.flags, test.dialDest)
 		if !reflect.DeepEqual(test.tt.closeErr, test.wantCloseErr) {
 			t.Errorf("test %d: close error mismatch: got %q, want %q", i, test.tt.closeErr, test.wantCloseErr)
 		}

p2p/simulations/README.md

+# devp2p Simulations
+
+The `p2p/simulations` package implements a simulation framework which supports
+creating a collection of devp2p nodes, connecting them together to form a
+simulation network, performing simulation actions in that network and then
+extracting useful information.
+
+## Nodes
+
+Each node in a simulation network runs multiple services by wrapping a collection
+of objects which implement the `node.Service` interface meaning they:
+
+* can be started and stopped
+* run p2p protocols
+* expose RPC APIs
+
+This means that any object which implements the `node.Service` interface can be
+used to run a node in the simulation.
+
+## Services
+
+Before running a simulation, a set of service initializers must be registered
+which can then be used to run nodes in the network.
+
+A service initializer is a function with the following signature:
+
+```go
+func(ctx *adapters.ServiceContext) (node.Service, error)
+```
+
+These initializers should be registered by calling the `adapters.RegisterServices`
+function in an `init()` hook:
+
+```go
+func init() {
+	adapters.RegisterServices(adapters.Services{
+		"service1": initService1,
+		"service2": initService2,
+	})
+}
+```
+
+## Node Adapters
+
+The simulation framework includes multiple "node adapters" which are
+responsible for creating an environment in which a node runs.
+
+### SimAdapter
+
+The `SimAdapter` runs nodes in-memory, connecting them using an in-memory,
+synchronous `net.Pipe` and connecting to their RPC server using an in-memory
+`rpc.Client`.
+
+### ExecAdapter
+
+The `ExecAdapter` runs nodes as child processes of the running simulation.
+
+It does this by executing the binary which is running the simulation but
+setting `argv[0]` (i.e. the program name) to `p2p-node` which is then
+detected by an init hook in the child process which runs the `node.Service`
+using the devp2p node stack rather than executing `main()`.
+
+The nodes listen for devp2p connections and WebSocket RPC clients on random
+localhost ports.
+
+### DockerAdapter
+
+The `DockerAdapter` is similar to the `ExecAdapter` but executes `docker run`
+to run the node in a Docker container using a Docker image containing the
+simulation binary at `/bin/p2p-node`.
+
+The Docker image is built using `docker build` when the adapter is initialised,
+meaning no prior setup is necessary other than having a working Docker client.
+
+Each node listens on the external IP of the container and the default p2p and
+RPC ports (`30303` and `8546` respectively).
+
+## Network
+
+A simulation network is created with an ID and default service (which is used
+if a node is created without an explicit service), exposes methods for
+creating, starting, stopping, connecting and disconnecting nodes, and emits
+events when certain actions occur.
+
+### Events
+
+A simulation network emits the following events:
+
+* node event       - when nodes are created / started / stopped
+* connection event - when nodes are connected / disconnected
+* message event    - when a protocol message is sent between two nodes
+
+The events have a "control" flag which when set indicates that the event is the
+outcome of a controlled simulation action (e.g. creating a node or explicitly
+connecting two nodes together).
+
+This is in contrast to a non-control event, otherwise called a "live" event,
+which is the outcome of something happening in the network as a result of a
+control event (e.g. a node actually started up or a connection was actually
+established between two nodes).
+
+Live events are detected by the simulation network by subscribing to node peer
+events via RPC when the nodes start up.
+
+## Testing Framework
+
+The `Simulation` type can be used in tests to perform actions in a simulation
+network and then wait for expectations to be met.
+
+With a running simulation network, the `Simulation.Run` method can be called
+with a `Step` which has the following fields:
+
+* `Action` - a function which performs some action in the network
+
+* `Expect` - an expectation function which returns whether or not a
+    given node meets the expectation
+
+* `Trigger` - a channel which receives node IDs which then trigger a check
+    of the expectation function to be performed against that node
+
+As a concrete example, consider a simulated network of Ethereum nodes. An
+`Action` could be the sending of a transaction, `Expect` it being included in
+a block, and `Trigger` a check for every block that is mined.
+
+On return, the `Simulation.Run` method returns a `StepResult` which can be used
+to determine if all nodes met the expectation, how long it took them to meet
+the expectation and what network events were emitted during the step run.
+
+## HTTP API
+
+The simulation framework includes a HTTP API which can be used to control the
+simulation.
+
+The API is initialised with a particular node adapter and has the following
+endpoints:
+
+```
+GET    /                            Get network information
+POST   /start                       Start all nodes in the network
+POST   /stop                        Stop all nodes in the network
+GET    /events                      Stream network events
+GET    /snapshot                    Take a network snapshot
+POST   /snapshot                    Load a network snapshot
+POST   /nodes                       Create a node
+GET    /nodes                       Get all nodes in the network
+GET    /nodes/:nodeid               Get node information
+POST   /nodes/:nodeid/start         Start a node
+POST   /nodes/:nodeid/stop          Stop a node
+POST   /nodes/:nodeid/conn/:peerid  Connect two nodes
+DELETE /nodes/:nodeid/conn/:peerid  Disconnect two nodes
+GET    /nodes/:nodeid/rpc           Make RPC requests to a node via WebSocket
+```
+
+For convenience, `nodeid` in the URL can be the name of a node rather than its
+ID.
+
+## Command line client
+
+`p2psim` is a command line client for the HTTP API, located in
+`cmd/p2psim`.
+
+It provides the following commands:
+
+```
+p2psim show
+p2psim events [--current] [--filter=FILTER]
+p2psim snapshot
+p2psim load
+p2psim node create [--name=NAME] [--services=SERVICES] [--key=KEY]
+p2psim node list
+p2psim node show <node>
+p2psim node start <node>
+p2psim node stop <node>
+p2psim node connect <node> <peer>
+p2psim node disconnect <node> <peer>
+p2psim node rpc <node> <method> [<args>] [--subscribe]
+```
+
+## Example
+
+See [p2p/simulations/examples/README.md](examples/README.md).

p2p/simulations/adapters/docker.go

+// Copyright 2017 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 (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+
+	"github.com/docker/docker/pkg/reexec"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+)
+
+// DockerAdapter is a NodeAdapter which runs simulation nodes inside Docker
+// containers.
+//
+// A Docker image is built which contains the current binary at /bin/p2p-node
+// which when executed runs the underlying service (see the description
+// of the execP2PNode function for more details)
+type DockerAdapter struct {
+	ExecAdapter
+}
+
+// NewDockerAdapter builds the p2p-node Docker image containing the current
+// binary and returns a DockerAdapter
+func NewDockerAdapter() (*DockerAdapter, error) {
+	// Since Docker containers run on Linux and this adapter runs the
+	// current binary in the container, it must be compiled for Linux.
+	//
+	// It is reasonable to require this because the caller can just
+	// compile the current binary in a Docker container.
+	if runtime.GOOS != "linux" {
+		return nil, errors.New("DockerAdapter can only be used on Linux as it uses the current binary (which must be a Linux binary)")
+	}
+
+	if err := buildDockerImage(); err != nil {
+		return nil, err
+	}
+
+	return &DockerAdapter{
+		ExecAdapter{
+			nodes: make(map[discover.NodeID]*ExecNode),
+		},
+	}, nil
+}
+
+// Name returns the name of the adapter for logging purposes
+func (d *DockerAdapter) Name() string {
+	return "docker-adapter"
+}
+
+// NewNode returns a new DockerNode using the given config
+func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
+	if len(config.Services) == 0 {
+		return nil, errors.New("node must have at least one service")
+	}
+	for _, service := range config.Services {
+		if _, exists := serviceFuncs[service]; !exists {
+			return nil, fmt.Errorf("unknown node service %q", service)
+		}
+	}
+
+	// generate the config
+	conf := &execNodeConfig{
+		Stack: node.DefaultConfig,
+		Node:  config,
+	}
+	conf.Stack.DataDir = "/data"
+	conf.Stack.WSHost = "0.0.0.0"
+	conf.Stack.WSOrigins = []string{"*"}
+	conf.Stack.WSExposeAll = true
+	conf.Stack.P2P.EnableMsgEvents = false
+	conf.Stack.P2P.NoDiscovery = true
+	conf.Stack.P2P.NAT = nil
+	conf.Stack.NoUSB = true
+
+	node := &DockerNode{
+		ExecNode: ExecNode{
+			ID:      config.ID,
+			Config:  conf,
+			adapter: &d.ExecAdapter,
+		},
+	}
+	node.newCmd = node.dockerCommand
+	d.ExecAdapter.nodes[node.ID] = &node.ExecNode
+	return node, nil
+}
+
+// DockerNode wraps an ExecNode but exec's the current binary in a docker
+// container rather than locally
+type DockerNode struct {
+	ExecNode
+}
+
+// dockerCommand returns a command which exec's the binary in a Docker
+// container.
+//
+// It uses a shell so that we can pass the _P2P_NODE_CONFIG environment
+// variable to the container using the --env flag.
+func (n *DockerNode) dockerCommand() *exec.Cmd {
+	return exec.Command(
+		"sh", "-c",
+		fmt.Sprintf(
+			`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`,
+			dockerImage, strings.Join(n.Config.Node.Services, ","), n.ID.String(),
+		),
+	)
+}
+
+// dockerImage is the name of the Docker image which gets built to run the
+// simulation node
+const dockerImage = "p2p-node"
+
+// buildDockerImage builds the Docker image which is used to run the simulation
+// node in a Docker container.
+//
+// It adds the current binary as "p2p-node" so that it runs execP2PNode
+// when executed.
+func buildDockerImage() error {
+	// create a directory to use as the build context
+	dir, err := ioutil.TempDir("", "p2p-docker")
+	if err != nil {
+		return err
+	}
+	defer os.RemoveAll(dir)
+
+	// copy the current binary into the build context
+	bin, err := os.Open(reexec.Self())
+	if err != nil {
+		return err
+	}
+	defer bin.Close()
+	dst, err := os.OpenFile(filepath.Join(dir, "self.bin"), os.O_WRONLY|os.O_CREATE, 0755)
+	if err != nil {
+		return err
+	}
+	defer dst.Close()
+	if _, err := io.Copy(dst, bin); err != nil {
+		return err
+	}
+
+	// create the Dockerfile
+	dockerfile := []byte(`
+FROM ubuntu:16.04
+RUN mkdir /data
+ADD self.bin /bin/p2p-node
+	`)
+	if err := ioutil.WriteFile(filepath.Join(dir, "Dockerfile"), dockerfile, 0644); err != nil {
+		return err
+	}
+
+	// run 'docker build'
+	cmd := exec.Command("docker", "build", "-t", dockerImage, dir)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("error building docker image: %s", err)
+	}
+
+	return nil
+}

p2p/simulations/adapters/inproc.go

+// Copyright 2017 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 (
+	"errors"
+	"fmt"
+	"math"
+	"net"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/event"
+	"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"
+)
+
+// SimAdapter is a NodeAdapter which creates in-memory simulation nodes and
+// connects them using in-memory net.Pipe connections
+type SimAdapter struct {
+	mtx      sync.RWMutex
+	nodes    map[discover.NodeID]*SimNode
+	services map[string]ServiceFunc
+}
+
+// NewSimAdapter creates a SimAdapter which is capable of running in-memory
+// simulation nodes running any of the given services (the services to run on a
+// particular node are passed to the NewNode function in the NodeConfig)
+func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter {
+	return &SimAdapter{
+		nodes:    make(map[discover.NodeID]*SimNode),
+		services: services,
+	}
+}
+
+// Name returns the name of the adapter for logging purposes
+func (s *SimAdapter) Name() string {
+	return "sim-adapter"
+}
+
+// NewNode returns a new SimNode using the given config
+func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	// check a node with the ID doesn't already exist
+	id := config.ID
+	if _, exists := s.nodes[id]; exists {
+		return nil, fmt.Errorf("node already exists: %s", id)
+	}
+
+	// check the services are valid
+	if len(config.Services) == 0 {
+		return nil, errors.New("node must have at least one service")
+	}
+	for _, service := range config.Services {
+		if _, exists := s.services[service]; !exists {
+			return nil, fmt.Errorf("unknown node service %q", service)
+		}
+	}
+
+	n, err := node.New(&node.Config{
+		P2P: p2p.Config{
+			PrivateKey:      config.PrivateKey,
+			MaxPeers:        math.MaxInt32,
+			NoDiscovery:     true,
+			Dialer:          s,
+			EnableMsgEvents: true,
+		},
+		NoUSB: true,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	simNode := &SimNode{
+		ID:      id,
+		config:  config,
+		node:    n,
+		adapter: s,
+		running: make(map[string]node.Service),
+	}
+	s.nodes[id] = simNode
+	return simNode, nil
+}
+
+// Dial implements the p2p.NodeDialer interface by connecting to the node using
+// an in-memory net.Pipe connection
+func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
+	node, ok := s.GetNode(dest.ID)
+	if !ok {
+		return nil, fmt.Errorf("unknown node: %s", dest.ID)
+	}
+	srv := node.Server()
+	if srv == nil {
+		return nil, fmt.Errorf("node not running: %s", dest.ID)
+	}
+	pipe1, pipe2 := net.Pipe()
+	go srv.SetupConn(pipe1, 0, nil)
+	return pipe2, nil
+}
+
+// DialRPC implements the RPCDialer interface by creating an in-memory RPC
+// client of the given node
+func (s *SimAdapter) DialRPC(id discover.NodeID) (*rpc.Client, error) {
+	node, ok := s.GetNode(id)
+	if !ok {
+		return nil, fmt.Errorf("unknown node: %s", id)
+	}
+	handler, err := node.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()
+	defer s.mtx.RUnlock()
+	node, ok := s.nodes[id]
+	return node, ok
+}
+
+// SimNode is an in-memory simulation node which connects to other nodes using
+// an in-memory net.Pipe connection (see SimAdapter.Dial), running devp2p
+// protocols directly over that pipe
+type SimNode struct {
+	lock         sync.RWMutex
+	ID           discover.NodeID
+	config       *NodeConfig
+	adapter      *SimAdapter
+	node         *node.Node
+	running      map[string]node.Service
+	client       *rpc.Client
+	registerOnce sync.Once
+}
+
+// Addr returns the node's discovery address
+func (self *SimNode) Addr() []byte {
+	return []byte(self.Node().String())
+}
+
+// Node returns a discover.Node representing the SimNode
+func (self *SimNode) Node() *discover.Node {
+	return discover.NewNode(self.ID, net.IP{127, 0, 0, 1}, 30303, 30303)
+}
+
+// Client returns an rpc.Client which can be used to communicate with the
+// underlying services (it is set once the node has started)
+func (self *SimNode) Client() (*rpc.Client, error) {
+	self.lock.RLock()
+	defer self.lock.RUnlock()
+	if self.client == nil {
+		return nil, errors.New("node not started")
+	}
+	return self.client, nil
+}
+
+// ServeRPC serves RPC requests over the given connection by creating an
+// in-memory client to the node's RPC server
+func (self *SimNode) ServeRPC(conn net.Conn) error {
+	handler, err := self.node.RPCHandler()
+	if err != nil {
+		return err
+	}
+	handler.ServeCodec(rpc.NewJSONCodec(conn), rpc.OptionMethodInvocation|rpc.OptionSubscriptions)
+	return nil
+}
+
+// Snapshots creates snapshots of the services by calling the
+// simulation_snapshot RPC method
+func (self *SimNode) Snapshots() (map[string][]byte, error) {
+	self.lock.RLock()
+	services := make(map[string]node.Service, len(self.running))
+	for name, service := range self.running {
+		services[name] = service
+	}
+	self.lock.RUnlock()
+	if len(services) == 0 {
+		return nil, errors.New("no running services")
+	}
+	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 registers the services and starts the underlying devp2p node
+func (self *SimNode) Start(snapshots map[string][]byte) error {
+	newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) {
+		return func(nodeCtx *node.ServiceContext) (node.Service, error) {
+			ctx := &ServiceContext{
+				RPCDialer:   self.adapter,
+				NodeContext: nodeCtx,
+				Config:      self.config,
+			}
+			if snapshots != nil {
+				ctx.Snapshot = snapshots[name]
+			}
+			serviceFunc := self.adapter.services[name]
+			service, err := serviceFunc(ctx)
+			if err != nil {
+				return nil, err
+			}
+			self.running[name] = service
+			return service, nil
+		}
+	}
+
+	// ensure we only register the services once in the case of the node
+	// being stopped and then started again
+	var regErr error
+	self.registerOnce.Do(func() {
+		for _, name := range self.config.Services {
+			if err := self.node.Register(newService(name)); err != nil {
+				regErr = err
+				return
+			}
+		}
+	})
+	if regErr != nil {
+		return regErr
+	}
+
+	if err := self.node.Start(); err != nil {
+		return err
+	}
+
+	// create an in-process RPC client
+	handler, err := self.node.RPCHandler()
+	if err != nil {
+		return err
+	}
+
+	self.lock.Lock()
+	self.client = rpc.DialInProc(handler)
+	self.lock.Unlock()
+
+	return nil
+}
+
+// Stop closes the RPC client and stops the underlying devp2p node
+func (self *SimNode) Stop() error {
+	self.lock.Lock()
+	if self.client != nil {
+		self.client.Close()
+		self.client = nil
+	}
+	self.lock.Unlock()
+	return self.node.Stop()
+}
+
+// Services returns a copy of the underlying services
+func (self *SimNode) Services() []node.Service {
+	self.lock.RLock()
+	defer self.lock.RUnlock()
+	services := make([]node.Service, 0, len(self.running))
+	for _, service := range self.running {
+		services = append(services, service)
+	}
+	return services
+}
+
+// Server returns the underlying p2p.Server
+func (self *SimNode) Server() *p2p.Server {
+	return self.node.Server()
+}
+
+// SubscribeEvents subscribes the given channel to peer events from the
+// underlying p2p.Server
+func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
+	srv := self.Server()
+	if srv == nil {
+		panic("node not running")
+	}
+	return srv.SubscribeEvents(ch)
+}
+
+// NodeInfo returns information about the node
+func (self *SimNode) NodeInfo() *p2p.NodeInfo {
+	server := self.Server()
+	if server == nil {
+		return &p2p.NodeInfo{
+			ID:    self.ID.String(),
+			Enode: self.Node().String(),
+		}
+	}
+	return server.NodeInfo()
+}

p2p/simulations/adapters/types.go

+// Copyright 2017 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 (
+	"crypto/ecdsa"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net"
+	"os"
+
+	"github.com/docker/docker/pkg/reexec"
+	"github.com/ethereum/go-ethereum/crypto"
+	"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"
+)
+
+// Node represents a node in a simulation network which is created by a
+// NodeAdapter, for example:
+//
+// * SimNode    - An in-memory node
+// * ExecNode   - A child process node
+// * DockerNode - A Docker container node
+//
+type Node interface {
+	// Addr returns the node's address (e.g. an Enode URL)
+	Addr() []byte
+
+	// Client returns the RPC client which is created once the node is
+	// up and running
+	Client() (*rpc.Client, error)
+
+	// ServeRPC serves RPC requests over the given connection
+	ServeRPC(net.Conn) error
+
+	// Start starts the node with the given snapshots
+	Start(snapshots map[string][]byte) error
+
+	// Stop stops the node
+	Stop() error
+
+	// NodeInfo returns information about the node
+	NodeInfo() *p2p.NodeInfo
+
+	// Snapshots creates snapshots of the running services
+	Snapshots() (map[string][]byte, error)
+}
+
+// NodeAdapter is used to create Nodes in a simulation network
+type NodeAdapter interface {
+	// Name returns the name of the adapter for logging purposes
+	Name() string
+
+	// NewNode creates a new node with the given configuration
+	NewNode(config *NodeConfig) (Node, error)
+}
+
+// NodeConfig is the configuration used to start a node in a simulation
+// network
+type NodeConfig struct {
+	// ID is the node's ID which is used to identify the node in the
+	// simulation network
+	ID discover.NodeID
+
+	// PrivateKey is the node's private key which is used by the devp2p
+	// stack to encrypt communications
+	PrivateKey *ecdsa.PrivateKey
+
+	// Name is a human friendly name for the node like "node01"
+	Name string
+
+	// Services are the names of the services which should be run when
+	// starting the node (for SimNodes it should be the names of services
+	// contained in SimAdapter.services, for other nodes it should be
+	// services registered by calling the RegisterService function)
+	Services []string
+}
+
+// nodeConfigJSON is used to encode and decode NodeConfig as JSON by encoding
+// all fields as strings
+type nodeConfigJSON struct {
+	ID         string   `json:"id"`
+	PrivateKey string   `json:"private_key"`
+	Name       string   `json:"name"`
+	Services   []string `json:"services"`
+}
+
+// MarshalJSON implements the json.Marshaler interface by encoding the config
+// fields as strings
+func (n *NodeConfig) MarshalJSON() ([]byte, error) {
+	confJSON := nodeConfigJSON{
+		ID:       n.ID.String(),
+		Name:     n.Name,
+		Services: n.Services,
+	}
+	if n.PrivateKey != nil {
+		confJSON.PrivateKey = hex.EncodeToString(crypto.FromECDSA(n.PrivateKey))
+	}
+	return json.Marshal(confJSON)
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface by decoding the json
+// string values into the config fields
+func (n *NodeConfig) UnmarshalJSON(data []byte) error {
+	var confJSON nodeConfigJSON
+	if err := json.Unmarshal(data, &confJSON); err != nil {
+		return err
+	}
+
+	if confJSON.ID != "" {
+		nodeID, err := discover.HexID(confJSON.ID)
+		if err != nil {
+			return err
+		}
+		n.ID = nodeID
+	}
+
+	if confJSON.PrivateKey != "" {
+		key, err := hex.DecodeString(confJSON.PrivateKey)
+		if err != nil {
+			return err
+		}
+		privKey, err := crypto.ToECDSA(key)
+		if err != nil {
+			return err
+		}
+		n.PrivateKey = privKey
+	}
+
+	n.Name = confJSON.Name
+	n.Services = confJSON.Services
+
+	return nil
+}
+
+// RandomNodeConfig returns node configuration with a randomly generated ID and
+// PrivateKey
+func RandomNodeConfig() *NodeConfig {
+	key, err := crypto.GenerateKey()
+	if err != nil {
+		panic("unable to generate key")
+	}
+	var id discover.NodeID
+	pubkey := crypto.FromECDSAPub(&key.PublicKey)
+	copy(id[:], pubkey[1:])
+	return &NodeConfig{
+		ID:         id,
+		PrivateKey: key,
+	}
+}
+
+// ServiceContext is a collection of options and methods which can be utilised
+// when starting services
+type ServiceContext struct {
+	RPCDialer
+
+	NodeContext *node.ServiceContext
+	Config      *NodeConfig
+	Snapshot    []byte
+}
+
+// RPCDialer is used when initialising services which need to connect to
+// other nodes in the network (for example a simulated Swarm node which needs
+// to connect to a Geth node to resolve ENS names)
+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 a devp2p node
+type ServiceFunc func(ctx *ServiceContext) (node.Service, error)
+
+// serviceFuncs is a map of registered services which are used to boot devp2p
+// nodes
+var serviceFuncs = make(Services)
+
+// RegisterServices registers the given Services which can then be used to
+// start devp2p nodes using either the Exec or Docker adapters.
+//
+// It should be called in an init function so that it has the opportunity to
+// execute the services before main() is called.
+func RegisterServices(services Services) {
+	for name, f := range services {
+		if _, exists := serviceFuncs[name]; exists {
+			panic(fmt.Sprintf("node service already exists: %q", name))
+		}
+		serviceFuncs[name] = f
+	}
+
+	// now we have registered the services, run reexec.Init() which will
+	// potentially start one of the services if the current binary has
+	// been exec'd with argv[0] set to "p2p-node"
+	if reexec.Init() {
+		os.Exit(0)
+	}
+}

p2p/simulations/events.go

+// Copyright 2017 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 simulations
+
+import (
+	"fmt"
+	"time"
+)
+
+// EventType is the type of event emitted by a simulation network
+type EventType string
+
+const (
+	// EventTypeNode is the type of event emitted when a node is either
+	// created, started or stopped
+	EventTypeNode EventType = "node"
+
+	// EventTypeConn is the type of event emitted when a connection is
+	// is either established or dropped between two nodes
+	EventTypeConn EventType = "conn"
+
+	// EventTypeMsg is the type of event emitted when a p2p message it
+	// sent between two nodes
+	EventTypeMsg EventType = "msg"
+)
+
+// Event is an event emitted by a simulation network
+type Event struct {
+	// Type is the type of the event
+	Type EventType `json:"type"`
+
+	// Time is the time the event happened
+	Time time.Time `json:"time"`
+
+	// Control indicates whether the event is the result of a controlled
+	// action in the network
+	Control bool `json:"control"`
+
+	// Node is set if the type is EventTypeNode
+	Node *Node `json:"node,omitempty"`
+
+	// Conn is set if the type is EventTypeConn
+	Conn *Conn `json:"conn,omitempty"`
+
+	// Msg is set if the type is EventTypeMsg
+	Msg *Msg `json:"msg,omitempty"`
+}
+
+// NewEvent creates a new event for the given object which should be either a
+// Node, Conn or Msg.
+//
+// The object is copied so that the event represents the state of the object
+// when NewEvent is called.
+func NewEvent(v interface{}) *Event {
+	event := &Event{Time: time.Now()}
+	switch v := v.(type) {
+	case *Node:
+		event.Type = EventTypeNode
+		node := *v
+		event.Node = &node
+	case *Conn:
+		event.Type = EventTypeConn
+		conn := *v
+		event.Conn = &conn
+	case *Msg:
+		event.Type = EventTypeMsg
+		msg := *v
+		event.Msg = &msg
+	default:
+		panic(fmt.Sprintf("invalid event type: %T", v))
+	}
+	return event
+}
+
+// ControlEvent creates a new control event
+func ControlEvent(v interface{}) *Event {
+	event := NewEvent(v)
+	event.Control = true
+	return event
+}
+
+// String returns the string representation of the event
+func (e *Event) String() string {
+	switch e.Type {
+	case EventTypeNode:
+		return fmt.Sprintf("<node-event> id: %s up: %t", e.Node.ID().TerminalString(), e.Node.Up)
+	case EventTypeConn:
+		return fmt.Sprintf("<conn-event> nodes: %s->%s up: %t", e.Conn.One.TerminalString(), e.Conn.Other.TerminalString(), e.Conn.Up)
+	case EventTypeMsg:
+		return fmt.Sprintf("<msg-event> nodes: %s->%s proto: %s, code: %d, received: %t", e.Msg.One.TerminalString(), e.Msg.Other.TerminalString(), e.Msg.Protocol, e.Msg.Code, e.Msg.Received)
+	default:
+		return ""
+	}
+}

p2p/simulations/examples/README.md

+# devp2p simulation examples
+
+## ping-pong
+
+`ping-pong.go` implements a simulation network which contains nodes running a
+simple "ping-pong" protocol where nodes send a ping message to all their
+connected peers every 10s and receive pong messages in return.
+
+To run the simulation, run `go run ping-pong.go` in one terminal to start the
+simulation API and `./ping-pong.sh` in another to start and connect the nodes:
+
+```
+$ go run ping-pong.go
+INFO [08-15|13:53:49] using sim adapter
+INFO [08-15|13:53:49] starting simulation server on 0.0.0.0:8888...
+```
+
+```
+$ ./ping-pong.sh
+---> 13:58:12 creating 10 nodes
+Created node01
+Started node01
+...
+Created node10
+Started node10
+---> 13:58:13 connecting node01 to all other nodes
+Connected node01 to node02
+...
+Connected node01 to node10
+---> 13:58:14 done
+```
+
+Use the `--adapter` flag to choose the adapter type:
+
+```
+$ go run ping-pong.go --adapter exec
+INFO [08-15|14:01:14] using exec adapter                       tmpdir=/var/folders/k6/wpsgfg4n23ddbc6f5cnw5qg00000gn/T/p2p-example992833779
+INFO [08-15|14:01:14] starting simulation server on 0.0.0.0:8888...
+```

p2p/simulations/examples/ping-pong.go

+// Copyright 2017 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 main
+
+import (
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"sync/atomic"
+	"time"
+
+	"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/p2p/simulations"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+var adapterType = flag.String("adapter", "sim", `node adapter to use (one of "sim", "exec" or "docker")`)
+
+// main() starts a simulation network which contains nodes running a simple
+// ping-pong protocol
+func main() {
+	flag.Parse()
+
+	// set the log level to Trace
+	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+
+	// register a single ping-pong service
+	services := map[string]adapters.ServiceFunc{
+		"ping-pong": func(ctx *adapters.ServiceContext) (node.Service, error) {
+			return newPingPongService(ctx.Config.ID), nil
+		},
+	}
+	adapters.RegisterServices(services)
+
+	// create the NodeAdapter
+	var adapter adapters.NodeAdapter
+
+	switch *adapterType {
+
+	case "sim":
+		log.Info("using sim adapter")
+		adapter = adapters.NewSimAdapter(services)
+
+	case "exec":
+		tmpdir, err := ioutil.TempDir("", "p2p-example")
+		if err != nil {
+			log.Crit("error creating temp dir", "err", err)
+		}
+		defer os.RemoveAll(tmpdir)
+		log.Info("using exec adapter", "tmpdir", tmpdir)
+		adapter = adapters.NewExecAdapter(tmpdir)
+
+	case "docker":
+		log.Info("using docker adapter")
+		var err error
+		adapter, err = adapters.NewDockerAdapter()
+		if err != nil {
+			log.Crit("error creating docker adapter", "err", err)
+		}
+
+	default:
+		log.Crit(fmt.Sprintf("unknown node adapter %q", *adapterType))
+	}
+
+	// start the HTTP API
+	log.Info("starting simulation server on 0.0.0.0:8888...")
+	network := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
+		DefaultService: "ping-pong",
+	})
+	if err := http.ListenAndServe(":8888", simulations.NewServer(network)); err != nil {
+		log.Crit("error starting simulation server", "err", err)
+	}
+}
+
+// pingPongService runs a ping-pong protocol between nodes where each node
+// sends a ping to all its connected peers every 10s and receives a pong in
+// return
+type pingPongService struct {
+	id       discover.NodeID
+	log      log.Logger
+	received int64
+}
+
+func newPingPongService(id discover.NodeID) *pingPongService {
+	return &pingPongService{
+		id:  id,
+		log: log.New("node.id", id),
+	}
+}
+
+func (p *pingPongService) Protocols() []p2p.Protocol {
+	return []p2p.Protocol{{
+		Name:     "ping-pong",
+		Version:  1,
+		Length:   2,
+		Run:      p.Run,
+		NodeInfo: p.Info,
+	}}
+}
+
+func (p *pingPongService) APIs() []rpc.API {
+	return nil
+}
+
+func (p *pingPongService) Start(server *p2p.Server) error {
+	p.log.Info("ping-pong service starting")
+	return nil
+}
+
+func (p *pingPongService) Stop() error {
+	p.log.Info("ping-pong service stopping")
+	return nil
+}
+
+func (p *pingPongService) Info() interface{} {
+	return struct {
+		Received int64 `json:"received"`
+	}{
+		atomic.LoadInt64(&p.received),
+	}
+}
+
+const (
+	pingMsgCode = iota
+	pongMsgCode
+)
+
+// Run implements the ping-pong protocol which sends ping messages to the peer
+// at 10s intervals, and responds to pings with pong messages.
+func (p *pingPongService) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
+	log := p.log.New("peer.id", peer.ID())
+
+	errC := make(chan error)
+	go func() {
+		for range time.Tick(10 * time.Second) {
+			log.Info("sending ping")
+			if err := p2p.Send(rw, pingMsgCode, "PING"); err != nil {
+				errC <- err
+				return
+			}
+		}
+	}()
+	go func() {
+		for {
+			msg, err := rw.ReadMsg()
+			if err != nil {
+				errC <- err
+				return
+			}
+			payload, err := ioutil.ReadAll(msg.Payload)
+			if err != nil {
+				errC <- err
+				return
+			}
+			log.Info("received message", "msg.code", msg.Code, "msg.payload", string(payload))
+			atomic.AddInt64(&p.received, 1)
+			if msg.Code == pingMsgCode {
+				log.Info("sending pong")
+				go p2p.Send(rw, pongMsgCode, "PONG")
+			}
+		}
+	}()
+	return <-errC
+}

p2p/simulations/examples/ping-pong.sh

+#!/bin/bash
+#
+# Boot a ping-pong network simulation using the HTTP API started by ping-pong.go
+
+set -e
+
+main() {
+  if ! which p2psim &>/dev/null; then
+    fail "missing p2psim binary (you need to build cmd/p2psim and put it in \$PATH)"
+  fi
+
+  info "creating 10 nodes"
+  for i in $(seq 1 10); do
+    p2psim node create --name "$(node_name $i)"
+    p2psim node start "$(node_name $i)"
+  done
+
+  info "connecting node01 to all other nodes"
+  for i in $(seq 2 10); do
+    p2psim node connect "node01" "$(node_name $i)"
+  done
+
+  info "done"
+}
+
+node_name() {
+  local num=$1
+  echo "node$(printf '%02d' $num)"
+}
+
+info() {
+  echo -e "\033[1;32m---> $(date +%H:%M:%S) ${@}\033[0m"
+}
+
+fail() {
+  echo -e "\033[1;31mERROR: ${@}\033[0m" >&2
+  exit 1
+}
+
+main "$@"

rpc/client_test.go

@@ -251,6 +251,38 @@ func TestClientSubscribe(t *testing.T) {
 	}
 }
 
+func TestClientSubscribeCustomNamespace(t *testing.T) {
+	namespace := "custom"
+	server := newTestServer(namespace, new(NotificationTestService))
+	defer server.Stop()
+	client := DialInProc(server)
+	defer client.Close()
+
+	nc := make(chan int)
+	count := 10
+	sub, err := client.Subscribe(context.Background(), namespace, nc, "someSubscription", count, 0)
+	if err != nil {
+		t.Fatal("can't subscribe:", err)
+	}
+	for i := 0; i < count; i++ {
+		if val := <-nc; val != i {
+			t.Fatalf("value mismatch: got %d, want %d", val, i)
+		}
+	}
+
+	sub.Unsubscribe()
+	select {
+	case v := <-nc:
+		t.Fatal("received value after unsubscribe:", v)
+	case err := <-sub.Err():
+		if err != nil {
+			t.Fatalf("Err returned a non-nil error after explicit unsubscribe: %q", err)
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatalf("subscription not closed within 1s after unsubscribe")
+	}
+}
+
 // In this test, the connection drops while EthSubscribe is
 // waiting for a response.
 func TestClientSubscribeClose(t *testing.T) {

vendor/github.com/julienschmidt/httprouter/LICENSE

+Copyright (c) 2013 Julien Schmidt. All rights reserved.
+
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above copyright
+      notice, this list of conditions and the following disclaimer in the
+      documentation and/or other materials provided with the distribution.
+    * The names of the contributors may not be used to endorse or promote
+      products derived from this software without specific prior written
+      permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL JULIEN SCHMIDT BE LIABLE FOR ANY
+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
\ No newline at end of file

vendor/vendor.json

@@ -182,6 +182,12 @@
 			"revision": "1fa385a6f45828c83361136b45b1a21a12139493",
 			"revisionTime": "2016-06-03T03:41:37Z"
 		},
+		{
+			"checksumSHA1": "gKyBj05YkfuLFruAyPZ4KV9nFp8=",
+			"path": "github.com/julienschmidt/httprouter",
+			"revision": "975b5c4c7c21c0e3d2764200bf2aa8e34657ae6e",
+			"revisionTime": "2017-04-30T22:20:11Z"
+		},
 		{
 			"checksumSHA1": "UpjhOUZ1+0zNt+iIvdtECSHXmTs=",
 			"path": "github.com/karalabe/hid",