From 7486904b92449c5955bb682f4ff98752906912b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 5 Feb 2016 15:08:48 +0200 Subject: [PATCH] cmd, node, rpc: move websockets into node, break singleton --- cmd/geth/js_test.go | 52 +++++++++--------- cmd/geth/main.go | 21 ++++--- cmd/geth/usage.go | 2 +- cmd/gethrpctest/main.go | 3 + cmd/utils/client.go | 118 ---------------------------------------- cmd/utils/flags.go | 40 +++++--------- node/api.go | 49 ++++++----------- node/config.go | 34 ++++++++++++ node/node.go | 85 ++++++++++++++++++++++++----- rpc/ipc_windows.go | 6 ++ rpc/websocket.go | 76 ++++---------------------- 11 files changed, 194 insertions(+), 292 deletions(-) diff --git a/cmd/geth/js_test.go b/cmd/geth/js_test.go index 19583c5ef..58800ed44 100644 --- a/cmd/geth/js_test.go +++ b/cmd/geth/js_test.go @@ -20,6 +20,7 @@ import ( "fmt" "io/ioutil" "math/big" + "math/rand" "os" "path/filepath" "regexp" @@ -29,6 +30,7 @@ import ( "time" "github.com/ethereum/go-ethereum/accounts" + "github.com/ethereum/go-ethereum/cmd/utils" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/compiler" "github.com/ethereum/go-ethereum/common/httpclient" @@ -37,22 +39,21 @@ import ( "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/node" - "github.com/ethereum/go-ethereum/cmd/utils" ) const ( testSolcPath = "" - solcVersion = "0.9.23" + solcVersion = "0.9.23" - testKey = "e6fab74a43941f82d89cb7faa408e227cdad3153c4720e540e855c19b15e6674" + testKey = "e6fab74a43941f82d89cb7faa408e227cdad3153c4720e540e855c19b15e6674" testAddress = "0x8605cdbbdb6d264aa742e77020dcbc58fcdce182" testBalance = "10000000000000000000" -// of empty string + // of empty string testHash = "0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470" ) var ( - versionRE = regexp.MustCompile(strconv.Quote(`"compilerVersion":"` + solcVersion + `"`)) + versionRE = regexp.MustCompile(strconv.Quote(`"compilerVersion":"` + solcVersion + `"`)) testNodeKey = crypto.ToECDSA(common.Hex2Bytes("4b50fa71f5c3eeb8fdc452224b2395af2fcc3d125e06c32c82e048c0559db03f")) testGenesis = `{"` + testAddress[2:] + `": {"balance": "` + testBalance + `"}}` ) @@ -95,7 +96,7 @@ func testREPL(t *testing.T, config func(*eth.Config)) (string, *testjethre, *nod t.Fatal(err) } // Create a networkless protocol stack - stack, err := node.New(&node.Config{PrivateKey: testNodeKey, Name: "test", NoDiscovery: true}) + stack, err := node.New(&node.Config{PrivateKey: testNodeKey, Name: "test", NoDiscovery: true, IpcPath: fmt.Sprintf("geth-test-%d.ipc", rand.Int63())}) if err != nil { t.Fatalf("failed to create node: %v", err) } @@ -141,8 +142,10 @@ func testREPL(t *testing.T, config func(*eth.Config)) (string, *testjethre, *nod stack.Service(ðereum) assetPath := filepath.Join(os.Getenv("GOPATH"), "src", "github.com", "ethereum", "go-ethereum", "cmd", "mist", "assets", "ext") - //client := comms.NewInProcClient(codec.JSON) - client := utils.NewInProcRPCClient(stack) + client, err := utils.NewRemoteRPCClientFromString("ipc:" + stack.IpcEndpoint()) + if err != nil { + t.Fatalf("failed to attach to node: %v", err) + } tf := &testjethre{client: ethereum.HTTPClient()} repl := newJSRE(stack, assetPath, "", client, false) tf.jsre = repl @@ -152,9 +155,6 @@ func testREPL(t *testing.T, config func(*eth.Config)) (string, *testjethre, *nod func TestNodeInfo(t *testing.T) { t.Skip("broken after p2p update") tmp, repl, ethereum := testJEthRE(t) - if err := ethereum.Start(); err != nil { - t.Fatalf("error starting ethereum: %v", err) - } defer ethereum.Stop() defer os.RemoveAll(tmp) @@ -167,8 +167,8 @@ func TestAccounts(t *testing.T) { defer node.Stop() defer os.RemoveAll(tmp) - checkEvalJSON(t, repl, `eth.accounts`, `["` + testAddress + `"]`) - checkEvalJSON(t, repl, `eth.coinbase`, `"` + testAddress + `"`) + checkEvalJSON(t, repl, `eth.accounts`, `["`+testAddress+`"]`) + checkEvalJSON(t, repl, `eth.coinbase`, `"`+testAddress+`"`) val, err := repl.re.Run(`jeth.newAccount("password")`) if err != nil { t.Errorf("expected no error, got %v", err) @@ -178,7 +178,7 @@ func TestAccounts(t *testing.T) { t.Errorf("address not hex: %q", addr) } - checkEvalJSON(t, repl, `eth.accounts`, `["` + testAddress + `","` + addr + `"]`) + checkEvalJSON(t, repl, `eth.accounts`, `["`+testAddress+`","`+addr+`"]`) } @@ -206,13 +206,13 @@ func TestBlockChain(t *testing.T) { node.Service(ðereum) ethereum.BlockChain().Reset() - checkEvalJSON(t, repl, `admin.exportChain(` + tmpfileq + `)`, `true`) + checkEvalJSON(t, repl, `admin.exportChain(`+tmpfileq+`)`, `true`) if _, err := os.Stat(tmpfile); err != nil { t.Fatal(err) } // check import, verify that dumpBlock gives the same result. - checkEvalJSON(t, repl, `admin.importChain(` + tmpfileq + `)`, `true`) + checkEvalJSON(t, repl, `admin.importChain(`+tmpfileq+`)`, `true`) checkEvalJSON(t, repl, `debug.dumpBlock(eth.blockNumber)`, beforeExport) } @@ -240,7 +240,7 @@ func TestCheckTestAccountBalance(t *testing.T) { defer os.RemoveAll(tmp) repl.re.Run(`primary = "` + testAddress + `"`) - checkEvalJSON(t, repl, `eth.getBalance(primary)`, `"` + testBalance + `"`) + checkEvalJSON(t, repl, `eth.getBalance(primary)`, `"`+testBalance+`"`) } func TestSignature(t *testing.T) { @@ -301,11 +301,11 @@ func TestContract(t *testing.T) { */ source := `contract test {\n` + - " /// @notice Will multiply `a` by 7." + `\n` + - ` function multiply(uint a) returns(uint d) {\n` + - ` return a * 7;\n` + - ` }\n` + - `}\n` + " /// @notice Will multiply `a` by 7." + `\n` + + ` function multiply(uint a) returns(uint d) {\n` + + ` return a * 7;\n` + + ` }\n` + + `}\n` if checkEvalJSON(t, repl, `admin.stopNatSpec()`, `true`) != nil { return @@ -315,10 +315,10 @@ func TestContract(t *testing.T) { if err != nil { t.Fatalf("%v", err) } - if checkEvalJSON(t, repl, `primary = eth.accounts[0]`, `"` + testAddress + `"`) != nil { + if checkEvalJSON(t, repl, `primary = eth.accounts[0]`, `"`+testAddress+`"`) != nil { return } - if checkEvalJSON(t, repl, `source = "` + source + `"`, `"` + source + `"`) != nil { + if checkEvalJSON(t, repl, `source = "`+source+`"`, `"`+source+`"`) != nil { return } @@ -396,7 +396,7 @@ multiply7 = Multiply7.at(contractaddress); var contentHash = `"0x86d2b7cf1e72e9a7a3f8d96601f0151742a2f780f1526414304fbe413dc7f9bd"` if sol != nil && solcVersion != sol.Version() { - modContractInfo := versionRE.ReplaceAll(contractInfo, []byte(`"compilerVersion":"` + sol.Version() + `"`)) + modContractInfo := versionRE.ReplaceAll(contractInfo, []byte(`"compilerVersion":"`+sol.Version()+`"`)) fmt.Printf("modified contractinfo:\n%s\n", modContractInfo) contentHash = `"` + common.ToHex(crypto.Sha3([]byte(modContractInfo))) + `"` } @@ -481,7 +481,7 @@ func processTxs(repl *testjethre, t *testing.T, expTxc int) bool { repl.wait <- height select { case <-timer.C: - // if times out make sure the xeth loop does not block + // if times out make sure the xeth loop does not block go func() { select { case repl.wait <- nil: diff --git a/cmd/geth/main.go b/cmd/geth/main.go index a321181a1..61f0632db 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -312,7 +312,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso utils.WSListenAddrFlag, utils.WSPortFlag, utils.WSApiFlag, - utils.WSAllowedDomainsFlag, + utils.WSCORSDomainFlag, utils.IPCDisabledFlag, utils.IPCApiFlag, utils.IPCPathFlag, @@ -399,7 +399,7 @@ func attach(ctx *cli.Context) { // attach to a running geth instance client, err := utils.NewRemoteRPCClient(ctx) if err != nil { - utils.Fatalf("Unable to attach to geth - %v", err) + utils.Fatalf("Unable to attach to geth: %v", err) } repl := newLightweightJSRE( @@ -425,8 +425,10 @@ func console(ctx *cli.Context) { startNode(ctx, node) // Attach to the newly started node, and either execute script or become interactive - client := utils.NewInProcRPCClient(node) - + client, err := utils.NewRemoteRPCClientFromString("ipc:" + node.IpcEndpoint()) + if err != nil { + utils.Fatalf("Failed to attach to the inproc geth: %v", err) + } repl := newJSRE(node, ctx.GlobalString(utils.JSpathFlag.Name), ctx.GlobalString(utils.RPCCORSDomainFlag.Name), @@ -449,8 +451,10 @@ func execScripts(ctx *cli.Context) { startNode(ctx, node) // Attach to the newly started node and execute the given scripts - client := utils.NewInProcRPCClient(node) - + client, err := utils.NewRemoteRPCClientFromString("ipc:" + node.IpcEndpoint()) + if err != nil { + utils.Fatalf("Failed to attach to the inproc geth: %v", err) + } repl := newJSRE(node, ctx.GlobalString(utils.JSpathFlag.Name), ctx.GlobalString(utils.RPCCORSDomainFlag.Name), @@ -503,11 +507,6 @@ func startNode(ctx *cli.Context, stack *node.Node) { } } // Start auxiliary services if enabled - if ctx.GlobalBool(utils.WSEnabledFlag.Name) { - if err := utils.StartWS(stack, ctx); err != nil { - utils.Fatalf("Failed to start WS: %v", err) - } - } if ctx.GlobalBool(utils.MiningEnabledFlag.Name) { if err := ethereum.StartMining(ctx.GlobalInt(utils.MinerThreadsFlag.Name), ctx.GlobalString(utils.MiningGPUFlag.Name)); err != nil { utils.Fatalf("Failed to start mining: %v", err) diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index 051c51878..e20c67bf8 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -93,7 +93,7 @@ var AppHelpFlagGroups = []flagGroup{ utils.WSListenAddrFlag, utils.WSPortFlag, utils.WSApiFlag, - utils.WSAllowedDomainsFlag, + utils.WSCORSDomainFlag, utils.IPCDisabledFlag, utils.IPCApiFlag, utils.IPCPathFlag, diff --git a/cmd/gethrpctest/main.go b/cmd/gethrpctest/main.go index b0907f8c5..8b54fa2c1 100644 --- a/cmd/gethrpctest/main.go +++ b/cmd/gethrpctest/main.go @@ -97,6 +97,9 @@ func MakeSystemNode(keydir string, privkey string, test *tests.BlockTest) (*node HttpHost: common.DefaultHttpHost, HttpPort: common.DefaultHttpPort, HttpModules: []string{"admin", "db", "eth", "debug", "miner", "net", "shh", "txpool", "personal", "web3"}, + WsHost: common.DefaultWsHost, + WsPort: common.DefaultWsPort, + WsModules: []string{"admin", "db", "eth", "debug", "miner", "net", "shh", "txpool", "personal", "web3"}, NoDiscovery: true, }) if err != nil { diff --git a/cmd/utils/client.go b/cmd/utils/client.go index 40ebcd729..8595cd90b 100644 --- a/cmd/utils/client.go +++ b/cmd/utils/client.go @@ -17,132 +17,14 @@ package utils import ( - "encoding/json" "fmt" - "strings" "github.com/codegangsta/cli" - "github.com/ethereum/go-ethereum/eth" - "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" ) -// NewInProcRPCClient will start a new RPC server for the given node and returns a client to interact with it. -func NewInProcRPCClient(stack *node.Node) *inProcClient { - server := rpc.NewServer() - - offered := stack.APIs() - for _, api := range offered { - server.RegisterName(api.Namespace, api.Service) - } - - web3 := node.NewPublicWeb3API(stack) - server.RegisterName("web3", web3) - - var ethereum *eth.Ethereum - if err := stack.Service(ðereum); err == nil { - net := eth.NewPublicNetAPI(stack.Server(), ethereum.NetVersion()) - server.RegisterName("net", net) - } else { - glog.V(logger.Warn).Infof("%v\n", err) - } - - buf := &buf{ - requests: make(chan []byte), - responses: make(chan []byte), - } - client := &inProcClient{ - server: server, - buf: buf, - } - - go func() { - server.ServeCodec(rpc.NewJSONCodec(client.buf)) - }() - - return client -} - -// buf represents the connection between the RPC server and console -type buf struct { - readBuf []byte // store remaining request bytes after a partial read - requests chan []byte // list with raw serialized requests - responses chan []byte // list with raw serialized responses -} - -// will read the next request in json format -func (b *buf) Read(p []byte) (int, error) { - // last read didn't read entire request, return remaining bytes - if len(b.readBuf) > 0 { - n := copy(p, b.readBuf) - if n < len(b.readBuf) { - b.readBuf = b.readBuf[:n] - } else { - b.readBuf = b.readBuf[:0] - } - return n, nil - } - - // read next request - req := <-b.requests - n := copy(p, req) - if n < len(req) { - // buf too small, store remaining chunk for next read - b.readBuf = req[n:] - } - - return n, nil -} - -// Write send the given buffer to the backend -func (b *buf) Write(p []byte) (n int, err error) { - b.responses <- p - return len(p), nil -} - -// Close cleans up obtained resources. -func (b *buf) Close() error { - close(b.requests) - close(b.responses) - - return nil -} - -// inProcClient starts a RPC server and uses buf to communicate with it. -type inProcClient struct { - server *rpc.Server - buf *buf -} - -// Close will stop the RPC server -func (c *inProcClient) Close() { - c.server.Stop() -} - -// Send a msg to the endpoint -func (c *inProcClient) Send(msg interface{}) error { - d, err := json.Marshal(msg) - if err != nil { - return err - } - c.buf.requests <- d - return nil -} - -// Recv reads a message and tries to parse it into the given msg -func (c *inProcClient) Recv(msg interface{}) error { - data := <-c.buf.responses - return json.Unmarshal(data, &msg) -} - -// Returns the collection of modules the RPC server offers. -func (c *inProcClient) SupportedModules() (map[string]string, error) { - return rpc.SupportedModules(c) -} - // NewRemoteRPCClient returns a RPC client which connects to a running geth instance. // Depending on the given context this can either be a IPC or a HTTP client. func NewRemoteRPCClient(ctx *cli.Context) (rpc.Client, error) { diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 40ea29d78..28c692689 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -18,7 +18,6 @@ package utils import ( "crypto/ecdsa" - "errors" "fmt" "io/ioutil" "math" @@ -283,8 +282,8 @@ var ( Usage: "API's offered over the WS-RPC interface", Value: rpc.DefaultHttpRpcApis, } - WSAllowedDomainsFlag = cli.StringFlag{ - Name: "wscors", + WSCORSDomainFlag = cli.StringFlag{ + Name: "wscorsdomain", Usage: "Domains from which to accept websockets requests", Value: "", } @@ -491,6 +490,15 @@ func MakeHttpRpcHost(ctx *cli.Context) string { return ctx.GlobalString(RPCListenAddrFlag.Name) } +// MakeWsRpcHost creates the WebSocket RPC listener interface string from the set +// command line flags, returning empty if the HTTP endpoint is disabled. +func MakeWsRpcHost(ctx *cli.Context) string { + if !ctx.GlobalBool(WSEnabledFlag.Name) { + return "" + } + return ctx.GlobalString(WSListenAddrFlag.Name) +} + // MakeGenesisBlock loads up a genesis block from an input file specified in the // command line, or returns the empty string if none set. func MakeGenesisBlock(ctx *cli.Context) string { @@ -613,6 +621,10 @@ func MakeSystemNode(name, version string, extra []byte, ctx *cli.Context) *node. HttpPort: ctx.GlobalInt(RPCPortFlag.Name), HttpCors: ctx.GlobalString(RPCCORSDomainFlag.Name), HttpModules: strings.Split(ctx.GlobalString(RPCApiFlag.Name), ","), + WsHost: MakeWsRpcHost(ctx), + WsPort: ctx.GlobalInt(WSPortFlag.Name), + WsCors: ctx.GlobalString(WSCORSDomainFlag.Name), + WsModules: strings.Split(ctx.GlobalString(WSApiFlag.Name), ","), } // Configure the Ethereum service accman := MakeAccountManager(ctx) @@ -753,27 +765,5 @@ func MakeChain(ctx *cli.Context) (chain *core.BlockChain, chainDb ethdb.Database if err != nil { Fatalf("Could not start chainmanager: %v", err) } - return chain, chainDb } - -// StartWS starts a websocket JSON-RPC API server. -func StartWS(stack *node.Node, ctx *cli.Context) error { - for _, api := range stack.APIs() { - if adminApi, ok := api.Service.(*node.PrivateAdminAPI); ok { - address := ctx.GlobalString(WSListenAddrFlag.Name) - port := ctx.GlobalInt(WSAllowedDomainsFlag.Name) - allowedDomains := ctx.GlobalString(WSAllowedDomainsFlag.Name) - apiStr := "" - if ctx.GlobalIsSet(WSApiFlag.Name) { - apiStr = ctx.GlobalString(WSApiFlag.Name) - } - - _, err := adminApi.StartWS(address, port, allowedDomains, apiStr) - return err - } - } - - glog.V(logger.Error).Infof("Unable to start RPC-WS interface, could not find admin API") - return errors.New("Unable to start RPC-WS interface") -} diff --git a/node/api.go b/node/api.go index 1b185c6f1..879b33816 100644 --- a/node/api.go +++ b/node/api.go @@ -25,9 +25,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" - "gopkg.in/fatih/set.v0" ) // PrivateAdminAPI is the collection of administrative API methods exposed only @@ -86,44 +84,29 @@ func (api *PrivateAdminAPI) StopRPC() (bool, error) { } // StartWS starts the websocket RPC API server. -func (api *PrivateAdminAPI) StartWS(address string, port int, cors string, apis string) (bool, error) { - var offeredAPIs []rpc.API - if len(apis) > 0 { - namespaces := set.New() - for _, a := range strings.Split(apis, ",") { - namespaces.Add(strings.TrimSpace(a)) - } - for _, api := range api.node.APIs() { - if namespaces.Has(api.Namespace) { - offeredAPIs = append(offeredAPIs, api) - } - } - } else { - // use by default all public API's - for _, api := range api.node.APIs() { - if api.Public { - offeredAPIs = append(offeredAPIs, api) - } - } - } +func (api *PrivateAdminAPI) StartWS(host string, port int, cors string, apis string) (bool, error) { + api.node.lock.Lock() + defer api.node.lock.Unlock() - if address == "" { - address = "127.0.0.1" + if api.node.wsHandler != nil { + return false, fmt.Errorf("WebSocker RPC already running on %s", api.node.wsEndpoint) } - if port == 0 { - port = 8546 + if err := api.node.startWS(fmt.Sprintf("%s:%d", host, port), api.node.rpcAPIs, strings.Split(apis, ","), cors); err != nil { + return false, err } - - corsDomains := strings.Split(cors, " ") - - err := rpc.StartWS(address, port, corsDomains, offeredAPIs) - return err == nil, err + return true, nil } // StopRPC terminates an already running websocket RPC API endpoint. func (api *PrivateAdminAPI) StopWS() (bool, error) { - err := rpc.StopWS() - return err == nil, err + api.node.lock.Lock() + defer api.node.lock.Unlock() + + if api.node.wsHandler == nil { + return false, fmt.Errorf("WebSocket RPC not running") + } + api.node.stopWS() + return true, nil } // PublicAdminAPI is the collection of administrative API methods exposed over diff --git a/node/config.go b/node/config.go index 94c6e2e56..f8252b63a 100644 --- a/node/config.go +++ b/node/config.go @@ -117,6 +117,25 @@ type Config struct { // If the module list is empty, all RPC API endpoints designated public will be // exposed. HttpModules []string + + // WsHost is the host interface on which to start the websocket RPC server. If + // this field is empty, no websocket API endpoint will be started. + WsHost string + + // WsPort is the TCP port number on which to start the websocket RPC server. The + // default zero value is/ valid and will pick a port number randomly (useful for + // ephemeral nodes). + WsPort int + + // WsCors is the Cross-Origin Resource Sharing header to send to requesting clients. + // Please be aware that CORS is a browser enforced security, it's fully useless + // for custom websocket clients. + WsCors string + + // WsModules is a list of API modules to expose via the websocket RPC interface. + // If the module list is empty, all RPC API endpoints designated public will be + // exposed. + WsModules []string } // IpcEndpoint resolves an IPC endpoint based on a configured value, taking into @@ -165,6 +184,21 @@ func DefaultHttpEndpoint() string { return config.HttpEndpoint() } +// WsEndpoint resolves an websocket endpoint based on the configured host interface +// and port parameters. +func (c *Config) WsEndpoint() string { + if c.WsHost == "" { + return "" + } + return fmt.Sprintf("%s:%d", c.WsHost, c.WsPort) +} + +// DefaultWsEndpoint returns the websocket endpoint used by default. +func DefaultWsEndpoint() string { + config := &Config{WsHost: common.DefaultWsHost, WsPort: common.DefaultWsPort} + return config.WsEndpoint() +} + // 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/node.go b/node/node.go index 44c88d378..804748b6b 100644 --- a/node/node.go +++ b/node/node.go @@ -66,6 +66,12 @@ type Node struct { httpListener net.Listener // HTTP RPC listener socket to server API requests httpHandler *rpc.Server // HTTP RPC request handler to process the API requests + wsEndpoint string // Websocket endpoint (interface + port) to listen at (empty = websocket disabled) + wsWhitelist []string // Websocket RPC modules to allow through this endpoint + wsCors string // Websocket RPC Cross-Origin Resource Sharing header + wsListener net.Listener // Websocket RPC listener socket to server API requests + wsHandler *rpc.Server // Websocket RPC request handler to process the API requests + stop chan struct{} // Channel to wait for termination notifications lock sync.RWMutex } @@ -105,6 +111,9 @@ func New(conf *Config) (*Node, error) { httpEndpoint: conf.HttpEndpoint(), httpWhitelist: conf.HttpModules, httpCors: conf.HttpCors, + wsEndpoint: conf.WsEndpoint(), + wsWhitelist: conf.WsModules, + wsCors: conf.WsCors, eventmux: new(event.TypeMux), }, nil } @@ -215,6 +224,11 @@ func (n *Node) startRPC(services map[reflect.Type]Service) error { n.stopIPC() return err } + if err := n.startWS(n.wsEndpoint, apis, n.wsWhitelist, n.wsCors); err != nil { + n.stopHTTP() + n.stopIPC() + return err + } // All API endpoints started successfully n.rpcAPIs = apis return nil @@ -285,7 +299,7 @@ func (n *Node) stopIPC() { // startHTTP initializes and starts the HTTP RPC endpoint. func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors string) error { - // Short circuit if the IPC endpoint isn't being exposed + // Short circuit if the HTTP endpoint isn't being exposed if endpoint == "" { return nil } @@ -338,6 +352,61 @@ func (n *Node) stopHTTP() { } } +// startWS initializes and starts the websocket RPC endpoint. +func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, cors string) error { + // Short circuit if the WS endpoint isn't being exposed + if endpoint == "" { + return nil + } + // Generate the whitelist based on the allowed modules + whitelist := make(map[string]bool) + for _, module := range modules { + whitelist[module] = true + } + // 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 err := handler.RegisterName(api.Namespace, api.Service); err != nil { + return err + } + glog.V(logger.Debug).Infof("WebSocket registered %T under '%s'", api.Service, api.Namespace) + } + } + // All APIs registered, start the HTTP listener + var ( + listener net.Listener + err error + ) + if listener, err = net.Listen("tcp", endpoint); err != nil { + return err + } + go rpc.NewWSServer(cors, handler).Serve(listener) + glog.V(logger.Info).Infof("WebSocket endpoint opened: ws://%s", endpoint) + + // All listeners booted successfully + n.wsEndpoint = endpoint + n.wsListener = listener + n.wsHandler = handler + n.wsCors = cors + + return nil +} + +// stopWS terminates the websocket RPC endpoint. +func (n *Node) stopWS() { + if n.wsListener != nil { + n.wsListener.Close() + n.wsListener = nil + + glog.V(logger.Info).Infof("WebSocket endpoint closed: ws://%s", n.wsEndpoint) + } + if n.wsHandler != nil { + n.wsHandler.Stop() + n.wsHandler = 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 { @@ -349,8 +418,9 @@ func (n *Node) Stop() error { return ErrNodeStopped } // Otherwise terminate the API, all services and the P2P server too - n.stopIPC() + n.stopWS() n.stopHTTP() + n.stopIPC() n.rpcAPIs = nil failure := &StopError{ @@ -471,14 +541,3 @@ func (n *Node) apis() []rpc.API { }, } } - -// 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()...) - } - return apis -} diff --git a/rpc/ipc_windows.go b/rpc/ipc_windows.go index 1d4672ad2..09b01974e 100644 --- a/rpc/ipc_windows.go +++ b/rpc/ipc_windows.go @@ -239,6 +239,9 @@ func Dial(address string) (*PipeConn, error) { for { conn, err := dial(address, nmpwait_wait_forever) if err == nil { + // Ugly hack working around some async connectivity issues + time.Sleep(100 * time.Millisecond) + return conn, nil } if isPipeNotReady(err) { @@ -360,6 +363,9 @@ func Listen(address string) (*PipeListener, error) { if err != nil { return nil, err } + // Ugly hack working around some async connectivity issues + time.Sleep(100 * time.Millisecond) + return &PipeListener{ addr: PipeAddr(address), handle: handle, diff --git a/rpc/websocket.go b/rpc/websocket.go index b5bcbf4f6..548847602 100644 --- a/rpc/websocket.go +++ b/rpc/websocket.go @@ -17,13 +17,11 @@ package rpc import ( - "errors" "fmt" - "net" "net/http" - "sync" - "os" + "strings" + "sync" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" @@ -31,12 +29,6 @@ import ( "gopkg.in/fatih/set.v0" ) -var ( - wsServerMu sync.Mutex - wsRPCServer *Server - wsListener net.Listener -) - // wsReaderWriterCloser reads and write payloads from and to a websocket connection. type wsReaderWriterCloser struct { c *websocket.Conn @@ -57,14 +49,6 @@ func (rw *wsReaderWriterCloser) Close() error { return rw.c.Close() } -// wsHandler accepts a websocket connection and handles incoming RPC requests. -// Will return when the websocket connection is closed, either by the client or -// server. -func wsHandler(conn *websocket.Conn) { - rwc := &wsReaderWriterCloser{conn} - wsRPCServer.ServeCodec(NewJSONCodec(rwc)) -} - // wsHandshakeValidator returns a handler that verifies the origin during the // websocket upgrade process. When a '*' is specified as an allowed origins all // connections are accepted. @@ -103,54 +87,16 @@ func wsHandshakeValidator(allowedOrigins []string) func(*websocket.Config, *http return f } -// StartWS will start a websocket RPC server on the given address and port. -func StartWS(address string, port int, corsdomains []string, apis []API) error { - wsServerMu.Lock() - defer wsServerMu.Unlock() - - if wsRPCServer != nil { - return fmt.Errorf("WS RPC interface already started on %s", wsListener.Addr()) +// NewWSServer creates a new websocket RPC server around an API provider. +func NewWSServer(cors string, handler *Server) *http.Server { + return &http.Server{ + Handler: websocket.Server{ + Handshake: wsHandshakeValidator(strings.Split(cors, ",")), + Handler: func(conn *websocket.Conn) { + handler.ServeCodec(NewJSONCodec(&wsReaderWriterCloser{conn})) + }, + }, } - - rpcServer := NewServer() - for _, api := range apis { - if err := rpcServer.RegisterName(api.Namespace, api.Service); err != nil { - return err - } - } - - listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", address, port)) - if err != nil { - return err - } - - wsServer := websocket.Server{Handshake: wsHandshakeValidator(corsdomains), Handler: wsHandler} - wsHTTPServer := http.Server{Handler: wsServer} - - go wsHTTPServer.Serve(listener) - - wsListener = listener - wsRPCServer = rpcServer - - return nil -} - -// StopWS stops the running websocket RPC server. -func StopWS() error { - wsServerMu.Lock() - defer wsServerMu.Unlock() - - if wsRPCServer == nil { - return errors.New("HTTP RPC interface not started") - } - - wsListener.Close() - wsRPCServer.Stop() - - wsRPCServer = nil - wsListener = nil - - return nil } // wsClient represents a RPC client that communicates over websockets with a