mirror
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum
2
0
Fork 1
Code Issues Releases Wiki Activity

cmd, node, rpc: move websockets into node, break singleton

Browse Source
pull/2175/head
Péter Szilágyi 9 years ago
parent a13bc9d7a1
commit 7486904b92
11 changed files with 194 additions and 292 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 52
      cmd/geth/js_test.go
  2. 21
      cmd/geth/main.go
  3. 2
      cmd/geth/usage.go
  4. 3
      cmd/gethrpctest/main.go
  5. 118
      cmd/utils/client.go
  6. 40
      cmd/utils/flags.go
  7. 49
      node/api.go
  8. 34
      node/config.go
  9. 85
      node/node.go
  10. 6
      rpc/ipc_windows.go
  11. 76
      rpc/websocket.go

52
cmd/geth/js_test.go
Unescape View File

@ -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(&ethereum)
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(&ethereum)
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:

21
cmd/geth/main.go
Unescape View File

@ -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)

2
cmd/geth/usage.go
Unescape View File

@ -93,7 +93,7 @@ var AppHelpFlagGroups = []flagGroup{
utils.WSListenAddrFlag,
utils.WSPortFlag,
utils.WSApiFlag,
utils.WSAllowedDomainsFlag,
utils.WSCORSDomainFlag,
utils.IPCDisabledFlag,
utils.IPCApiFlag,
utils.IPCPathFlag,

3
cmd/gethrpctest/main.go
Unescape View File

@ -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 {

118
cmd/utils/client.go
Unescape View File

@ -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(&ethereum); 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) {

40
cmd/utils/flags.go
Unescape View File

@ -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")
}

49
node/api.go
Unescape View File

@ -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

34
node/config.go
Unescape View File

@ -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.

85
node/node.go
Unescape View File

@ -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
}

6
rpc/ipc_windows.go
Unescape View File

@ -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,

76
rpc/websocket.go
Unescape View File

@ -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())
}
rpcServer := NewServer()
for _, api := range apis {
if err := rpcServer.RegisterName(api.Namespace, api.Service); err != nil {
return err
}
// 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}))
},
},
}
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

Write Preview
Loading…
Cancel
Save