gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

cmd, common, node, rpc: move HTTP RPC into node, drop singletone aspect

Browse Source
This commit is contained in:
Péter Szilágyi
2016-02-05 13:45:36 +02:00
parent ba7c125153
commit a13bc9d7a1
10 changed files with 280 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
node/api.go
View File

@ -27,7 +27,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/rpc"
"github.com/rcrowley/go-metrics"
"gopkg.in/fatih/set.v0"
)
@ -61,42 +60,29 @@ func (api *PrivateAdminAPI) AddPeer(url string) (bool, error) {
}
// StartRPC starts the HTTP RPC API server.
func (api *PrivateAdminAPI) StartRPC(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) StartRPC(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.httpHandler != nil {
return false, fmt.Errorf("HTTP RPC already running on %s", api.node.httpEndpoint)
}
if port == 0 {
port = 8545
if err := api.node.startHTTP(fmt.Sprintf("%s:%d", host, port), api.node.rpcAPIs, strings.Split(apis, ","), cors); err != nil {
return false, err
}
corsDomains := strings.Split(cors, " ")
err := rpc.StartHTTP(address, port, corsDomains, offeredAPIs)
return err == nil, err
return true, nil
}
// StopRPC terminates an already running HTTP RPC API endpoint.
func (api *PrivateAdminAPI) StopRPC() (bool, error) {
err := rpc.StopHTTP()
return err == nil, err
api.node.lock.Lock()
defer api.node.lock.Unlock()
if api.node.httpHandler == nil {
return false, fmt.Errorf("HTTP RPC not running")
}
api.node.stopHTTP()
return true, nil
}
// StartWS starts the websocket RPC API server.

37
node/config.go
View File

@ -19,6 +19,7 @@ package node
import (
"crypto/ecdsa"
"encoding/json"
"fmt"
"io/ioutil"
"net"
"os"
@ -97,6 +98,25 @@ type Config struct {
// handshake phase, counted separately for inbound and outbound connections.
// Zero defaults to preset values.
MaxPendingPeers int
// HttpHost is the host interface on which to start the HTTP RPC server. If this
// field is empty, no HTTP API endpoint will be started.
HttpHost string
// HttpPort is the TCP port number on which to start the HTTP RPC server. The
// default zero value is/ valid and will pick a port number randomly (useful
// for ephemeral nodes).
HttpPort int
// HttpCors 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 HTTP clients.
HttpCors string
// HttpModules is a list of API modules to expose via the HTTP RPC interface.
// If the module list is empty, all RPC API endpoints designated public will be
// exposed.
HttpModules []string
}
// IpcEndpoint resolves an IPC endpoint based on a configured value, taking into
@ -126,10 +146,25 @@ func (c *Config) IpcEndpoint() string {
// DefaultIpcEndpoint returns the IPC path used by default.
func DefaultIpcEndpoint() string {
config := &Config{DataDir: common.DefaultDataDir(), IpcPath: common.DefaultIpcSocket()}
config := &Config{DataDir: common.DefaultDataDir(), IpcPath: common.DefaultIpcSocket}
return config.IpcEndpoint()
}
// HttpEndpoint resolves an HTTP endpoint based on the configured host interface
// and port parameters.
func (c *Config) HttpEndpoint() string {
if c.HttpHost == "" {
return ""
}
return fmt.Sprintf("%s:%d", c.HttpHost, c.HttpPort)
}
// DefaultHttpEndpoint returns the HTTP endpoint used by default.
func DefaultHttpEndpoint() string {
config := &Config{HttpHost: common.DefaultHttpHost, HttpPort: common.DefaultHttpPort}
return config.HttpEndpoint()
}
// 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.

194
node/node.go
View File

@ -55,10 +55,17 @@ type Node struct {
serviceFuncs []ServiceConstructor // Service constructors (in dependency order)
services map[reflect.Type]Service // Currently running services
rpcAPIs []rpc.API // List of APIs currently provided by the node
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
httpEndpoint string // HTTP endpoint (interface + port) to listen at (empty = HTTP disabled)
httpWhitelist []string // HTTP RPC modules to allow through this endpoint
httpCors string // HTTP RPC Cross-Origin Resource Sharing header
httpListener net.Listener // HTTP RPC listener socket to server API requests
httpHandler *rpc.Server // HTTP RPC request handler to process the API requests
stop chan struct{} // Channel to wait for termination notifications
lock sync.RWMutex
}
@ -93,9 +100,12 @@ func New(conf *Config) (*Node, error) {
MaxPeers: conf.MaxPeers,
MaxPendingPeers: conf.MaxPendingPeers,
},
serviceFuncs: []ServiceConstructor{},
ipcEndpoint: conf.IpcEndpoint(),
eventmux: new(event.TypeMux),
serviceFuncs: []ServiceConstructor{},
ipcEndpoint: conf.IpcEndpoint(),
httpEndpoint: conf.HttpEndpoint(),
httpWhitelist: conf.HttpModules,
httpCors: conf.HttpCors,
eventmux: new(event.TypeMux),
}, nil
}
@ -188,58 +198,146 @@ func (n *Node) Start() error {
return nil
}
// startRPC initializes and starts the IPC RPC endpoints.
// startRPC is a helper method to start all the various RPC endpoint during node
// startup. It's not meant to be called at any time afterwards as it makes certain
// assumptions about the state of the node.
func (n *Node) startRPC(services map[reflect.Type]Service) error {
// Gather and register all the APIs exposed by the services
// Gather all the possible APIs to surface
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)
// Start the various API endpoints, terminating all in case of errors
if err := n.startIPC(apis); err != nil {
return err
}
// 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)
if err := n.startHTTP(n.httpEndpoint, apis, n.httpWhitelist, n.httpCors); err != nil {
n.stopIPC()
return err
}
// All API endpoints started successfully
n.rpcAPIs = apis
return nil
}
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))
}
}()
// startIPC initializes and starts the IPC RPC endpoint.
func (n *Node) startIPC(apis []rpc.API) error {
// Short circuit if the IPC endpoint isn't being exposed
if n.ipcEndpoint == "" {
return nil
}
// Register all the APIs exposed by the services
handler := rpc.NewServer()
for _, api := range apis {
if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
return err
}
glog.V(logger.Debug).Infof("IPC registered %T under '%s'", api.Service, api.Namespace)
}
// All APIs registered, start the IPC listener
var (
listener net.Listener
err error
)
if listener, err = rpc.CreateIPCListener(n.ipcEndpoint); err != nil {
return err
}
go func() {
glog.V(logger.Info).Infof("IPC endpoint opened: %s", n.ipcEndpoint)
for {
conn, err := listener.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 handler.ServeCodec(rpc.NewJSONCodec(conn))
}
}()
// All listeners booted successfully
n.ipcListener = ipcListener
n.ipcHandler = ipcHandler
n.ipcListener = listener
n.ipcHandler = handler
return nil
}
// stopIPC terminates the IPC RPC endpoint.
func (n *Node) stopIPC() {
if n.ipcListener != nil {
n.ipcListener.Close()
n.ipcListener = nil
glog.V(logger.Info).Infof("IPC endpoint closed: %s", n.ipcEndpoint)
}
if n.ipcHandler != nil {
n.ipcHandler.Stop()
n.ipcHandler = nil
}
}
// 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
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("HTTP 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.NewHTTPServer(cors, handler).Serve(listener)
glog.V(logger.Info).Infof("HTTP endpoint opened: http://%s", endpoint)
// All listeners booted successfully
n.httpEndpoint = endpoint
n.httpListener = listener
n.httpHandler = handler
n.httpCors = cors
return nil
}
// stopHTTP terminates the HTTP RPC endpoint.
func (n *Node) stopHTTP() {
if n.httpListener != nil {
n.httpListener.Close()
n.httpListener = nil
glog.V(logger.Info).Infof("HTTP endpoint closed: http://%s", n.httpEndpoint)
}
if n.httpHandler != nil {
n.httpHandler.Stop()
n.httpHandler = 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 {
@ -251,14 +349,10 @@ func (n *Node) Stop() error {
return ErrNodeStopped
}
// 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
}
n.stopIPC()
n.stopHTTP()
n.rpcAPIs = nil
failure := &StopError{
Services: make(map[reflect.Type]error),
}