diff options
author | Péter Szilágyi <peterke@gmail.com> | 2016-02-05 17:33:24 +0800 |
---|---|---|
committer | Péter Szilágyi <peterke@gmail.com> | 2016-02-05 17:33:24 +0800 |
commit | ba7c125153ce1be30985784a18edf38645406d03 (patch) | |
tree | 92c1e2b8ec6ef0d4b39381148b3497e311e80b95 | |
parent | 212828963172b3921827df15abdc8602480e947d (diff) | |
parent | 188ab928c3f2a2eac5ee0f7ac42cbf2f35568bcd (diff) | |
download | go-tangerine-ba7c125153ce1be30985784a18edf38645406d03.tar.gz go-tangerine-ba7c125153ce1be30985784a18edf38645406d03.tar.zst go-tangerine-ba7c125153ce1be30985784a18edf38645406d03.zip |
Merge pull request #2168 from karalabe/move-rpc-into-node
cmd, common, node, rpc: move IPC into the node itself
-rw-r--r-- | cmd/geth/main.go | 7 | ||||
-rw-r--r-- | cmd/geth/monitorcmd.go | 4 | ||||
-rw-r--r-- | cmd/gethrpctest/main.go | 61 | ||||
-rw-r--r-- | cmd/utils/client.go | 4 | ||||
-rw-r--r-- | cmd/utils/flags.go | 71 | ||||
-rw-r--r-- | common/path.go | 9 | ||||
-rw-r--r-- | node/config.go | 40 | ||||
-rw-r--r-- | node/config_test.go | 32 | ||||
-rw-r--r-- | node/node.go | 100 | ||||
-rw-r--r-- | node/node_example_test.go | 1 | ||||
-rw-r--r-- | node/node_test.go | 101 | ||||
-rw-r--r-- | node/service_test.go | 2 | ||||
-rw-r--r-- | node/utils_test.go | 14 | ||||
-rw-r--r-- | rpc/json.go | 2 |
14 files changed, 292 insertions, 156 deletions
diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 09c7eee05..fa456a7ac 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -502,12 +502,7 @@ func startNode(ctx *cli.Context, stack *node.Node) { unlockAccount(ctx, accman, trimmed, i, passwords) } } - // Start auxiliary services if enabled. - if !ctx.GlobalBool(utils.IPCDisabledFlag.Name) { - if err := utils.StartIPC(stack, ctx); err != nil { - utils.Fatalf("Failed to start IPC: %v", err) - } - } + // Start auxiliary services if enabled if ctx.GlobalBool(utils.RPCEnabledFlag.Name) { if err := utils.StartRPC(stack, ctx); err != nil { utils.Fatalf("Failed to start RPC: %v", err) diff --git a/cmd/geth/monitorcmd.go b/cmd/geth/monitorcmd.go index 1d7bf3f6a..4d56f2289 100644 --- a/cmd/geth/monitorcmd.go +++ b/cmd/geth/monitorcmd.go @@ -28,7 +28,7 @@ import ( "github.com/codegangsta/cli" "github.com/ethereum/go-ethereum/cmd/utils" - "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/rpc" "github.com/gizak/termui" ) @@ -36,7 +36,7 @@ import ( var ( monitorCommandAttachFlag = cli.StringFlag{ Name: "attach", - Value: "ipc:" + common.DefaultIpcPath(), + Value: "ipc:" + node.DefaultIpcEndpoint(), Usage: "API endpoint to attach to", } monitorCommandRowsFlag = cli.IntFlag{ diff --git a/cmd/gethrpctest/main.go b/cmd/gethrpctest/main.go index 8522258a9..becd09f5a 100644 --- a/cmd/gethrpctest/main.go +++ b/cmd/gethrpctest/main.go @@ -18,25 +18,20 @@ package main import ( + "errors" "flag" "io/ioutil" "log" "os" "os/signal" - "path/filepath" - "runtime" - - "errors" "github.com/ethereum/go-ethereum/accounts" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/node" - "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/tests" "github.com/ethereum/go-ethereum/whisper" ) @@ -89,11 +84,6 @@ func main() { } log.Println("Initial test suite passed...") - if err := StartIPC(stack); err != nil { - log.Fatalf("Failed to start IPC interface: %v\n", err) - } - log.Println("IPC Interface started, accepting requests...") - // Start the RPC interface and wait until terminated if err := StartRPC(stack); err != nil { log.Fatalf("Failed to start RPC interface: %v", err) @@ -109,7 +99,7 @@ func main() { // keystore path and initial pre-state. func MakeSystemNode(keydir string, privkey string, test *tests.BlockTest) (*node.Node, error) { // Create a networkless protocol stack - stack, err := node.New(&node.Config{NoDiscovery: true}) + stack, err := node.New(&node.Config{IpcPath: node.DefaultIpcEndpoint(), NoDiscovery: true}) if err != nil { return nil, err } @@ -194,50 +184,3 @@ func StartRPC(stack *node.Node) error { glog.V(logger.Error).Infof("Unable to start RPC-HTTP interface, could not find admin API") return errors.New("Unable to start RPC-HTTP interface") } - -// StartIPC initializes an IPC interface to the given protocol stack. -func StartIPC(stack *node.Node) error { - var ethereum *eth.Ethereum - if err := stack.Service(ðereum); err != nil { - return err - } - - endpoint := `\\.\pipe\geth.ipc` - if runtime.GOOS != "windows" { - endpoint = filepath.Join(common.DefaultDataDir(), "geth.ipc") - } - - listener, err := rpc.CreateIPCListener(endpoint) - if err != nil { - return err - } - - server := rpc.NewServer() - - // register package API's this node provides - offered := stack.APIs() - for _, api := range offered { - server.RegisterName(api.Namespace, api.Service) - glog.V(logger.Debug).Infof("Register %T under namespace '%s' for IPC service\n", api.Service, api.Namespace) - } - - //var ethereum *eth.Ethereum - //if err := stack.Service(ðereum); err != nil { - // return err - //} - - go func() { - glog.V(logger.Info).Infof("Start IPC server on %s\n", endpoint) - for { - conn, err := listener.Accept() - if err != nil { - glog.V(logger.Error).Infof("Unable to accept connection - %v\n", err) - } - - codec := rpc.NewJSONCodec(conn) - go server.ServeCodec(codec) - } - }() - - return nil -} diff --git a/cmd/utils/client.go b/cmd/utils/client.go index bac456491..40ebcd729 100644 --- a/cmd/utils/client.go +++ b/cmd/utils/client.go @@ -150,10 +150,8 @@ func NewRemoteRPCClient(ctx *cli.Context) (rpc.Client, error) { endpoint := ctx.Args().First() return NewRemoteRPCClientFromString(endpoint) } - // use IPC by default - endpoint := IPCSocketPath(ctx) - return rpc.NewIPCClient(endpoint) + return rpc.NewIPCClient(node.DefaultIpcEndpoint()) } // NewRemoteRPCClientFromString returns a RPC client which connects to the given diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 1e3e58e51..5d56ba7d0 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -261,8 +261,8 @@ var ( } IPCPathFlag = DirectoryFlag{ Name: "ipcpath", - Usage: "Filename for IPC socket/pipe", - Value: DirectoryString{common.DefaultIpcPath()}, + Usage: "Filename for IPC socket/pipe within the datadir (explicit paths escape it)", + Value: DirectoryString{common.DefaultIpcSocket()}, } WSEnabledFlag = cli.BoolFlag{ Name: "ws", @@ -394,6 +394,15 @@ func MustMakeDataDir(ctx *cli.Context) string { return "" } +// MakeIpcPath creates an IPC path configuration from the set command line flags, +// returning an empty string if IPC was explicitly disabled, or the set path. +func MakeIpcPath(ctx *cli.Context) string { + if ctx.GlobalBool(IPCDisabledFlag.Name) { + return "" + } + return ctx.GlobalString(IPCPathFlag.Name) +} + // MakeNodeKey creates a node key from set command line flags, either loading it // from a file or as a specified hex value. If neither flags were provided, this // method returns nil and an emphemeral key is to be generated. @@ -582,6 +591,7 @@ func MakeSystemNode(name, version string, extra []byte, ctx *cli.Context) *node. // Configure the node's service container stackConf := &node.Config{ DataDir: MustMakeDataDir(ctx), + IpcPath: MakeIpcPath(ctx), PrivateKey: MakeNodeKey(ctx), Name: MakeNodeName(name, version, ctx), NoDiscovery: ctx.GlobalBool(NoDiscoverFlag.Name), @@ -734,63 +744,6 @@ func MakeChain(ctx *cli.Context) (chain *core.BlockChain, chainDb ethdb.Database return chain, chainDb } -func IPCSocketPath(ctx *cli.Context) (ipcpath string) { - if runtime.GOOS == "windows" { - ipcpath = common.DefaultIpcPath() - if ctx.GlobalIsSet(IPCPathFlag.Name) { - ipcpath = ctx.GlobalString(IPCPathFlag.Name) - } - } else { - ipcpath = common.DefaultIpcPath() - if ctx.GlobalIsSet(DataDirFlag.Name) { - ipcpath = filepath.Join(ctx.GlobalString(DataDirFlag.Name), "geth.ipc") - } - if ctx.GlobalIsSet(IPCPathFlag.Name) { - ipcpath = ctx.GlobalString(IPCPathFlag.Name) - } - } - - return -} - -func StartIPC(stack *node.Node, ctx *cli.Context) error { - var ethereum *eth.Ethereum - if err := stack.Service(ðereum); err != nil { - return err - } - - endpoint := IPCSocketPath(ctx) - listener, err := rpc.CreateIPCListener(endpoint) - if err != nil { - return err - } - - server := rpc.NewServer() - - // register package API's this node provides - offered := stack.APIs() - for _, api := range offered { - server.RegisterName(api.Namespace, api.Service) - glog.V(logger.Debug).Infof("Register %T under namespace '%s' for IPC service\n", api.Service, api.Namespace) - } - - go func() { - glog.V(logger.Info).Infof("Start IPC server on %s\n", endpoint) - for { - conn, err := listener.Accept() - if err != nil { - glog.V(logger.Error).Infof("Unable to accept connection - %v\n", err) - } - - codec := rpc.NewJSONCodec(conn) - go server.ServeCodec(codec) - } - }() - - return nil - -} - // StartRPC starts a HTTP JSON-RPC API server. func StartRPC(stack *node.Node, ctx *cli.Context) error { for _, api := range stack.APIs() { diff --git a/common/path.go b/common/path.go index 9ba2f93c0..38c213a12 100644 --- a/common/path.go +++ b/common/path.go @@ -89,9 +89,8 @@ func DefaultDataDir() string { return "" } -func DefaultIpcPath() string { - if runtime.GOOS == "windows" { - return `\\.\pipe\geth.ipc` - } - return filepath.Join(DefaultDataDir(), "geth.ipc") +// DefaultIpcSocket returns the relative name of the default IPC socket. The path +// resolution is done by a node with other contextual infos. +func DefaultIpcSocket() string { + return "geth.ipc" } diff --git a/node/config.go b/node/config.go index 93f0ba79d..d3eb1c78b 100644 --- a/node/config.go +++ b/node/config.go @@ -23,7 +23,10 @@ import ( "net" "os" "path/filepath" + "runtime" + "strings" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" @@ -49,6 +52,12 @@ type Config struct { // in memory. DataDir string + // IpcPath is the requested location to place the IPC endpoint. If the path is + // a simple file name, it is placed inside the data directory (or on the root + // pipe path on Windows), whereas if it's a resolvable path name (absolute or + // relative), then that specific path is enforced. An empty path disables IPC. + IpcPath string + // This field should be a valid secp256k1 private key that will be used for both // remote peer identification as well as network traffic encryption. If no key // is configured, the preset one is loaded from the data dir, generating it if @@ -90,6 +99,37 @@ type Config struct { MaxPendingPeers int } +// IpcEndpoint resolves an IPC endpoint based on a configured value, taking into +// account the set data folders as well as the designated platform we're currently +// running on. +func (c *Config) IpcEndpoint() string { + // Short circuit if IPC has not been enabled + if c.IpcPath == "" { + return "" + } + // On windows we can only use plain top-level pipes + if runtime.GOOS == "windows" { + if strings.HasPrefix(c.IpcPath, `\\.\pipe\`) { + return c.IpcPath + } + return `\\.\pipe\` + c.IpcPath + } + // Resolve names into the data directory full paths otherwise + if filepath.Base(c.IpcPath) == c.IpcPath { + if c.DataDir == "" { + return filepath.Join(os.TempDir(), c.IpcPath) + } + return filepath.Join(c.DataDir, c.IpcPath) + } + return c.IpcPath +} + +// DefaultIpcEndpoint returns the IPC path used by default. +func DefaultIpcEndpoint() string { + config := &Config{DataDir: common.DefaultDataDir(), IpcPath: common.DefaultIpcSocket()} + return config.IpcEndpoint() +} + // NodeKey retrieves the currently configured private key of the node, checking // first any manually set key, falling back to the one found in the configured // data folder. If no key can be found, a new one is generated. diff --git a/node/config_test.go b/node/config_test.go index f59f3c0fe..efb864ce4 100644 --- a/node/config_test.go +++ b/node/config_test.go @@ -21,6 +21,7 @@ import ( "io/ioutil" "os" "path/filepath" + "runtime" "testing" "github.com/ethereum/go-ethereum/crypto" @@ -60,6 +61,37 @@ func TestDatadirCreation(t *testing.T) { } } +// Tests that IPC paths are correctly resolved to valid endpoints of different +// platforms. +func TestIpcPathResolution(t *testing.T) { + var tests = []struct { + DataDir string + IpcPath string + Windows bool + Endpoint string + }{ + {"", "", false, ""}, + {"data", "", false, ""}, + {"", "geth.ipc", false, filepath.Join(os.TempDir(), "geth.ipc")}, + {"data", "geth.ipc", false, "data/geth.ipc"}, + {"data", "./geth.ipc", false, "./geth.ipc"}, + {"data", "/geth.ipc", false, "/geth.ipc"}, + {"", "", true, ``}, + {"data", "", true, ``}, + {"", "geth.ipc", true, `\\.\pipe\geth.ipc`}, + {"data", "geth.ipc", true, `\\.\pipe\geth.ipc`}, + {"data", `\\.\pipe\geth.ipc`, true, `\\.\pipe\geth.ipc`}, + } + for i, test := range tests { + // Only run when platform/test match + if (runtime.GOOS == "windows") == test.Windows { + if endpoint := (&Config{DataDir: test.DataDir, IpcPath: test.IpcPath}).IpcEndpoint(); endpoint != test.Endpoint { + t.Errorf("test %d: IPC endpoint mismatch: have %s, want %s", i, endpoint, test.Endpoint) + } + } + } +} + // Tests that node keys can be correctly created, persisted, loaded and/or made // ephemeral. func TestNodeKeyPersistency(t *testing.T) { diff --git a/node/node.go b/node/node.go index 4af111a41..e3fc03360 100644 --- a/node/node.go +++ b/node/node.go @@ -19,6 +19,7 @@ package node import ( "errors" + "net" "os" "path/filepath" "reflect" @@ -27,6 +28,8 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/internal/debug" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" ) @@ -52,6 +55,10 @@ type Node struct { serviceFuncs []ServiceConstructor // Service constructors (in dependency order) services map[reflect.Type]Service // Currently running services + ipcEndpoint string // IPC endpoint to listen at (empty = IPC disabled) + ipcListener net.Listener // IPC RPC listener socket to serve API requests + ipcHandler *rpc.Server // IPC RPC request handler to process the API requests + stop chan struct{} // Channel to wait for termination notifications lock sync.RWMutex } @@ -87,6 +94,7 @@ func New(conf *Config) (*Node, error) { MaxPendingPeers: conf.MaxPendingPeers, }, serviceFuncs: []ServiceConstructor{}, + ipcEndpoint: conf.IpcEndpoint(), eventmux: new(event.TypeMux), }, nil } @@ -164,6 +172,14 @@ func (n *Node) Start() error { // Mark the service started for potential cleanup started = append(started, kind) } + // Lastly start the configured RPC interfaces + if err := n.startRPC(services); err != nil { + for _, service := range services { + service.Stop() + } + running.Stop() + return err + } // Finish initializing the startup n.services = services n.server = running @@ -172,6 +188,58 @@ func (n *Node) Start() error { return nil } +// startRPC initializes and starts the IPC RPC endpoints. +func (n *Node) startRPC(services map[reflect.Type]Service) error { + // Gather and register all the APIs exposed by the services + apis := n.apis() + for _, service := range services { + apis = append(apis, service.APIs()...) + } + ipcHandler := rpc.NewServer() + for _, api := range apis { + if err := ipcHandler.RegisterName(api.Namespace, api.Service); err != nil { + return err + } + glog.V(logger.Debug).Infof("Register %T under namespace '%s'", api.Service, api.Namespace) + } + // All APIs registered, start the IPC and HTTP listeners + var ( + ipcListener net.Listener + err error + ) + if n.ipcEndpoint != "" { + if ipcListener, err = rpc.CreateIPCListener(n.ipcEndpoint); err != nil { + return err + } + go func() { + glog.V(logger.Info).Infof("IPC endpoint opened: %s", n.ipcEndpoint) + defer glog.V(logger.Info).Infof("IPC endpoint closed: %s", n.ipcEndpoint) + + for { + conn, err := ipcListener.Accept() + if err != nil { + // Terminate if the listener was closed + n.lock.RLock() + closed := n.ipcListener == nil + n.lock.RUnlock() + if closed { + return + } + // Not closed, just some error; report and continue + glog.V(logger.Error).Infof("IPC accept failed: %v", err) + continue + } + go ipcHandler.ServeCodec(rpc.NewJSONCodec(conn)) + } + }() + } + // All listeners booted successfully + n.ipcListener = ipcListener + n.ipcHandler = ipcHandler + + return nil +} + // Stop terminates a running node along with all it's services. In the node was // not started, an error is returned. func (n *Node) Stop() error { @@ -182,7 +250,15 @@ func (n *Node) Stop() error { if n.server == nil { return ErrNodeStopped } - // Otherwise terminate all the services and the P2P server too + // Otherwise terminate the API, all services and the P2P server too + if n.ipcListener != nil { + n.ipcListener.Close() + n.ipcListener = nil + } + if n.ipcHandler != nil { + n.ipcHandler.Stop() + n.ipcHandler = nil + } failure := &StopError{ Services: make(map[reflect.Type]error), } @@ -261,18 +337,20 @@ func (n *Node) DataDir() string { return n.datadir } +// IpcEndpoint retrieves the current IPC endpoint used by the protocol stack. +func (n *Node) IpcEndpoint() string { + return n.ipcEndpoint +} + // EventMux retrieves the event multiplexer used by all the network services in // the current protocol stack. func (n *Node) EventMux() *event.TypeMux { return n.eventmux } -// APIs returns the collection of RPC descriptor this node offers. This method -// is just a quick placeholder passthrough for the RPC update, which in the next -// step will be fully integrated into the node itself. -func (n *Node) APIs() []rpc.API { - // Define all the APIs owned by the node itself - apis := []rpc.API{ +// apis returns the collection of RPC descriptors this node offers. +func (n *Node) apis() []rpc.API { + return []rpc.API{ { Namespace: "admin", Version: "1.0", @@ -298,7 +376,13 @@ func (n *Node) APIs() []rpc.API { Public: true, }, } - // Inject all the APIs owned by various services +} + +// APIs returns the collection of RPC descriptor this node offers. This method +// is just a quick placeholder passthrough for the RPC update, which in the next +// step will be fully integrated into the node itself. +func (n *Node) APIs() []rpc.API { + apis := n.apis() for _, api := range n.services { apis = append(apis, api.APIs()...) } diff --git a/node/node_example_test.go b/node/node_example_test.go index 5ff5d06a6..01ff683c0 100644 --- a/node/node_example_test.go +++ b/node/node_example_test.go @@ -31,6 +31,7 @@ import ( // // The following methods are needed to implement a node.Service: // - Protocols() []p2p.Protocol - devp2p protocols the service can communicate on +// - APIs() []rpc.API - api methods the service wants to expose on rpc channels // - Start() error - method invoked when the node is ready to start the service // - Stop() error - method invoked when the node terminates the service type SampleService struct{} diff --git a/node/node_test.go b/node/node_test.go index ef096fc91..53dcbcf74 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -18,27 +18,34 @@ package node import ( "errors" + "fmt" "io/ioutil" + "math/rand" "os" "reflect" "testing" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rpc" ) var ( testNodeKey, _ = crypto.GenerateKey() +) - testNodeConfig = &Config{ +func testNodeConfig() *Config { + return &Config{ + IpcPath: fmt.Sprintf("test-%d.ipc", rand.Int63()), PrivateKey: testNodeKey, Name: "test node", } -) +} // Tests that an empty protocol stack can be started, restarted and stopped. func TestNodeLifeCycle(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -101,7 +108,7 @@ func TestNodeUsedDataDir(t *testing.T) { // Tests whether services can be registered and duplicates caught. func TestServiceRegistry(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -133,7 +140,7 @@ func TestServiceRegistry(t *testing.T) { // Tests that registered services get started and stopped correctly. func TestServiceLifeCycle(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -183,7 +190,7 @@ func TestServiceLifeCycle(t *testing.T) { // Tests that services are restarted cleanly as new instances. func TestServiceRestarts(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -231,7 +238,7 @@ func TestServiceRestarts(t *testing.T) { // Tests that if a service fails to initialize itself, none of the other services // will be allowed to even start. func TestServiceConstructionAbortion(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -278,7 +285,7 @@ func TestServiceConstructionAbortion(t *testing.T) { // Tests that if a service fails to start, all others started before it will be // shut down. func TestServiceStartupAbortion(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -331,7 +338,7 @@ func TestServiceStartupAbortion(t *testing.T) { // Tests that even if a registered service fails to shut down cleanly, it does // not influece the rest of the shutdown invocations. func TestServiceTerminationGuarantee(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -406,7 +413,7 @@ func TestServiceTerminationGuarantee(t *testing.T) { // TestServiceRetrieval tests that individual services can be retrieved. func TestServiceRetrieval(t *testing.T) { // Create a simple stack and register two service types - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -441,7 +448,7 @@ func TestServiceRetrieval(t *testing.T) { // Tests that all protocols defined by individual services get launched. func TestProtocolGather(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } @@ -494,3 +501,75 @@ func TestProtocolGather(t *testing.T) { } } } + +// Tests that all APIs defined by individual services get exposed. +func TestAPIGather(t *testing.T) { + stack, err := New(testNodeConfig()) + if err != nil { + t.Fatalf("failed to create protocol stack: %v", err) + } + // Register a batch of services with some configured APIs + calls := make(chan string, 1) + + services := map[string]struct { + APIs []rpc.API + Maker InstrumentingWrapper + }{ + "Zero APIs": {[]rpc.API{}, InstrumentedServiceMakerA}, + "Single API": {[]rpc.API{ + {"single", "1", &OneMethodApi{fun: func() { calls <- "single.v1" }}, true}, + }, InstrumentedServiceMakerB}, + "Many APIs": {[]rpc.API{ + {"multi", "1", &OneMethodApi{fun: func() { calls <- "multi.v1" }}, true}, + {"multi.v2", "2", &OneMethodApi{fun: func() { calls <- "multi.v2" }}, true}, + {"multi.v2.nested", "2", &OneMethodApi{fun: func() { calls <- "multi.v2.nested" }}, true}, + }, InstrumentedServiceMakerC}, + } + for id, config := range services { + config := config + constructor := func(*ServiceContext) (Service, error) { + return &InstrumentedService{apis: config.APIs}, nil + } + if err := stack.Register(config.Maker(constructor)); err != nil { + t.Fatalf("service %s: registration failed: %v", id, err) + } + } + // Start the services and ensure all API start successfully + if err := stack.Start(); err != nil { + t.Fatalf("failed to start protocol stack: %v", err) + } + defer stack.Stop() + + // Connect to the RPC server and verify the various registered endpoints + ipcClient, err := rpc.NewIPCClient(stack.IpcEndpoint()) + if err != nil { + t.Fatalf("failed to connect to the IPC API server: %v", err) + } + + tests := []struct { + Method string + Result string + }{ + {"single_theOneMethod", "single.v1"}, + {"multi_theOneMethod", "multi.v1"}, + {"multi.v2_theOneMethod", "multi.v2"}, + {"multi.v2.nested_theOneMethod", "multi.v2.nested"}, + } + for i, test := range tests { + if err := ipcClient.Send(rpc.JSONRequest{Id: new(int64), Version: "2.0", Method: test.Method}); err != nil { + t.Fatalf("test %d: failed to send API request: %v", i, err) + } + reply := new(rpc.JSONSuccessResponse) + if err := ipcClient.Recv(reply); err != nil { + t.Fatalf("test %d: failed to read API reply: %v", i, err) + } + select { + case result := <-calls: + if result != test.Result { + t.Errorf("test %d: result mismatch: have %s, want %s", i, result, test.Result) + } + case <-time.After(time.Second): + t.Fatalf("test %d: rpc execution timeout", i) + } + } +} diff --git a/node/service_test.go b/node/service_test.go index 1dfb61f73..cfe7fe5dc 100644 --- a/node/service_test.go +++ b/node/service_test.go @@ -63,7 +63,7 @@ func TestContextDatabases(t *testing.T) { // Tests that already constructed services can be retrieves by later ones. func TestContextServices(t *testing.T) { - stack, err := New(testNodeConfig) + stack, err := New(testNodeConfig()) if err != nil { t.Fatalf("failed to create protocol stack: %v", err) } diff --git a/node/utils_test.go b/node/utils_test.go index 7755605ae..7cdfc2b3a 100644 --- a/node/utils_test.go +++ b/node/utils_test.go @@ -52,6 +52,7 @@ func NewNoopServiceD(*ServiceContext) (Service, error) { return new(NoopServiceD // methods can be instrumented both return value as well as event hook wise. type InstrumentedService struct { protocols []p2p.Protocol + apis []rpc.API start error stop error @@ -70,7 +71,7 @@ func (s *InstrumentedService) Protocols() []p2p.Protocol { } func (s *InstrumentedService) APIs() []rpc.API { - return nil + return s.apis } func (s *InstrumentedService) Start(server *p2p.Server) error { @@ -121,3 +122,14 @@ func InstrumentedServiceMakerB(base ServiceConstructor) ServiceConstructor { func InstrumentedServiceMakerC(base ServiceConstructor) ServiceConstructor { return InstrumentingWrapperMaker(base, reflect.TypeOf(InstrumentedServiceC{})) } + +// OneMethodApi is a single-method API handler to be returned by test services. +type OneMethodApi struct { + fun func() +} + +func (api *OneMethodApi) TheOneMethod() { + if api.fun != nil { + api.fun() + } +} diff --git a/rpc/json.go b/rpc/json.go index 02bc71c26..1ed943c00 100644 --- a/rpc/json.go +++ b/rpc/json.go @@ -41,7 +41,7 @@ type JSONRequest struct { Method string `json:"method"` Version string `json:"jsonrpc"` Id *int64 `json:"id,omitempty"` - Payload json.RawMessage `json:"params"` + Payload json.RawMessage `json:"params,omitempty"` } // JSON-RPC response |