This PR refactors RPC subsystem for status-go: cleans up two different RPC client implementations (RPCManager and rpc.Client) creates new package under geth/rpc moves routing logic into this package (rpc.(*router)) add tests for routing cleans up NodeManager
This commit is contained in:
Ivan Daniluk
Ivan Tomilov
parent
fd8c110293
commit
9a5c1774c3
|
|
@ -21,7 +21,6 @@ type StatusBackend struct {
|
|||
accountManager common.AccountManager
|
||||
txQueueManager common.TxQueueManager
|
||||
jailManager common.JailManager
|
||||
rpcManager common.RPCManager
|
||||
}
|
||||
|
||||
// NewStatusBackend create a new NewStatusBackend instance
|
||||
|
|
@ -36,7 +35,6 @@ func NewStatusBackend() *StatusBackend {
|
|||
nodeManager: nodeManager,
|
||||
accountManager: accountManager,
|
||||
jailManager: jail.New(nodeManager, accountManager, txQueueManager),
|
||||
rpcManager: node.NewRPCManager(nodeManager),
|
||||
txQueueManager: txQueueManager,
|
||||
}
|
||||
}
|
||||
|
|
@ -180,7 +178,8 @@ func (m *StatusBackend) ResetChainData() (<-chan struct{}, error) {
|
|||
|
||||
// CallRPC executes RPC request on node's in-proc RPC server
|
||||
func (m *StatusBackend) CallRPC(inputJSON string) string {
|
||||
return m.rpcManager.Call(inputJSON)
|
||||
client := m.nodeManager.RPCClient()
|
||||
return client.CallRaw(inputJSON)
|
||||
}
|
||||
|
||||
// SendTransaction creates a new transaction and waits until it's complete.
|
||||
|
|
|
|||
|
|
@ -209,7 +209,7 @@ func (s *BackendTestSuite) TestCallRPC() {
|
|||
{
|
||||
`{"jsonrpc":"2.0","method":"shh_version","params":[],"id":67}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"5.0"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"5.0"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("shh_version: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
|
|
@ -218,7 +218,7 @@ func (s *BackendTestSuite) TestCallRPC() {
|
|||
{
|
||||
`{"jsonrpc":"2.0","method":"web3_sha3","params":["0x68656c6c6f20776f726c64"],"id":64}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("web3_sha3: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
|
|
@ -227,7 +227,7 @@ func (s *BackendTestSuite) TestCallRPC() {
|
|||
{
|
||||
`{"jsonrpc":"2.0","method":"net_version","params":[],"id":67}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"4"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"4"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("net_version: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
|
|
|
|||
|
|
@ -15,10 +15,10 @@ import (
|
|||
"github.com/ethereum/go-ethereum/common/hexutil"
|
||||
"github.com/ethereum/go-ethereum/les"
|
||||
"github.com/ethereum/go-ethereum/node"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
|
||||
"github.com/robertkrimen/otto"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
"github.com/status-im/status-go/geth/rpc"
|
||||
"github.com/status-im/status-go/static"
|
||||
)
|
||||
|
||||
|
|
@ -87,16 +87,7 @@ type NodeManager interface {
|
|||
AccountKeyStore() (*keystore.KeyStore, error)
|
||||
|
||||
// RPCClient exposes reference to RPC client connected to the running node
|
||||
RPCClient() (*rpc.Client, error)
|
||||
|
||||
// RPCLocalClient exposes reference to RPC client connected to the running local node rpcserver
|
||||
RPCLocalClient() (*rpc.Client, error)
|
||||
|
||||
// RPCUpstreamClient exposes reference to RPC client connected to the upstream node server
|
||||
RPCUpstreamClient() (*rpc.Client, error)
|
||||
|
||||
// RPCServer exposes reference to running node's in-proc RPC server/handler
|
||||
RPCServer() (*rpc.Server, error)
|
||||
RPCClient() *rpc.Client
|
||||
}
|
||||
|
||||
// AccountManager defines expected methods for managing Status accounts
|
||||
|
|
@ -143,12 +134,6 @@ type AccountManager interface {
|
|||
AddressToDecryptedAccount(address, password string) (accounts.Account, *keystore.Key, error)
|
||||
}
|
||||
|
||||
// RPCManager defines expected methods for managing RPC client/server
|
||||
type RPCManager interface {
|
||||
// Call executes RPC request on node's in-proc RPC server
|
||||
Call(inputJSON string) string
|
||||
}
|
||||
|
||||
// RawCompleteTransactionResult is a JSON returned from transaction complete function (used internally)
|
||||
type RawCompleteTransactionResult struct {
|
||||
Hash common.Hash
|
||||
|
|
|
|||
|
|
@ -11,11 +11,11 @@ import (
|
|||
common "github.com/ethereum/go-ethereum/common"
|
||||
les "github.com/ethereum/go-ethereum/les"
|
||||
node "github.com/ethereum/go-ethereum/node"
|
||||
rpc "github.com/ethereum/go-ethereum/rpc"
|
||||
whisperv5 "github.com/ethereum/go-ethereum/whisper/whisperv5"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
otto "github.com/robertkrimen/otto"
|
||||
params "github.com/status-im/status-go/geth/params"
|
||||
rpc "github.com/status-im/status-go/geth/rpc"
|
||||
reflect "reflect"
|
||||
)
|
||||
|
||||
|
|
@ -209,11 +209,10 @@ func (mr *MockNodeManagerMockRecorder) AccountKeyStore() *gomock.Call {
|
|||
}
|
||||
|
||||
// RPCClient mocks base method
|
||||
func (m *MockNodeManager) RPCClient() (*rpc.Client, error) {
|
||||
func (m *MockNodeManager) RPCClient() *rpc.Client {
|
||||
ret := m.ctrl.Call(m, "RPCClient")
|
||||
ret0, _ := ret[0].(*rpc.Client)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
return ret0
|
||||
}
|
||||
|
||||
// RPCClient indicates an expected call of RPCClient
|
||||
|
|
@ -221,45 +220,6 @@ func (mr *MockNodeManagerMockRecorder) RPCClient() *gomock.Call {
|
|||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RPCClient", reflect.TypeOf((*MockNodeManager)(nil).RPCClient))
|
||||
}
|
||||
|
||||
// RPCLocalClient mocks base method
|
||||
func (m *MockNodeManager) RPCLocalClient() (*rpc.Client, error) {
|
||||
ret := m.ctrl.Call(m, "RPCLocalClient")
|
||||
ret0, _ := ret[0].(*rpc.Client)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// RPCLocalClient indicates an expected call of RPCLocalClient
|
||||
func (mr *MockNodeManagerMockRecorder) RPCLocalClient() *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RPCLocalClient", reflect.TypeOf((*MockNodeManager)(nil).RPCLocalClient))
|
||||
}
|
||||
|
||||
// RPCUpstreamClient mocks base method
|
||||
func (m *MockNodeManager) RPCUpstreamClient() (*rpc.Client, error) {
|
||||
ret := m.ctrl.Call(m, "RPCUpstreamClient")
|
||||
ret0, _ := ret[0].(*rpc.Client)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// RPCUpstreamClient indicates an expected call of RPCUpstreamClient
|
||||
func (mr *MockNodeManagerMockRecorder) RPCUpstreamClient() *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RPCUpstreamClient", reflect.TypeOf((*MockNodeManager)(nil).RPCUpstreamClient))
|
||||
}
|
||||
|
||||
// RPCServer mocks base method
|
||||
func (m *MockNodeManager) RPCServer() (*rpc.Server, error) {
|
||||
ret := m.ctrl.Call(m, "RPCServer")
|
||||
ret0, _ := ret[0].(*rpc.Server)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// RPCServer indicates an expected call of RPCServer
|
||||
func (mr *MockNodeManagerMockRecorder) RPCServer() *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RPCServer", reflect.TypeOf((*MockNodeManager)(nil).RPCServer))
|
||||
}
|
||||
|
||||
// MockAccountManager is a mock of AccountManager interface
|
||||
type MockAccountManager struct {
|
||||
ctrl *gomock.Controller
|
||||
|
|
@ -414,41 +374,6 @@ func (mr *MockAccountManagerMockRecorder) AddressToDecryptedAccount(address, pas
|
|||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddressToDecryptedAccount", reflect.TypeOf((*MockAccountManager)(nil).AddressToDecryptedAccount), address, password)
|
||||
}
|
||||
|
||||
// MockRPCManager is a mock of RPCManager interface
|
||||
type MockRPCManager struct {
|
||||
ctrl *gomock.Controller
|
||||
recorder *MockRPCManagerMockRecorder
|
||||
}
|
||||
|
||||
// MockRPCManagerMockRecorder is the mock recorder for MockRPCManager
|
||||
type MockRPCManagerMockRecorder struct {
|
||||
mock *MockRPCManager
|
||||
}
|
||||
|
||||
// NewMockRPCManager creates a new mock instance
|
||||
func NewMockRPCManager(ctrl *gomock.Controller) *MockRPCManager {
|
||||
mock := &MockRPCManager{ctrl: ctrl}
|
||||
mock.recorder = &MockRPCManagerMockRecorder{mock}
|
||||
return mock
|
||||
}
|
||||
|
||||
// EXPECT returns an object that allows the caller to indicate expected use
|
||||
func (m *MockRPCManager) EXPECT() *MockRPCManagerMockRecorder {
|
||||
return m.recorder
|
||||
}
|
||||
|
||||
// Call mocks base method
|
||||
func (m *MockRPCManager) Call(inputJSON string) string {
|
||||
ret := m.ctrl.Call(m, "Call", inputJSON)
|
||||
ret0, _ := ret[0].(string)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// Call indicates an expected call of Call
|
||||
func (mr *MockRPCManagerMockRecorder) Call(inputJSON interface{}) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Call", reflect.TypeOf((*MockRPCManager)(nil).Call), inputJSON)
|
||||
}
|
||||
|
||||
// MockTxQueue is a mock of TxQueue interface
|
||||
type MockTxQueue struct {
|
||||
ctrl *gomock.Controller
|
||||
|
|
@ -759,7 +684,7 @@ func (mr *MockJailCellMockRecorder) Get(arg0 interface{}) *gomock.Call {
|
|||
}
|
||||
|
||||
// Run mocks base method
|
||||
func (m *MockJailCell) Run(arg0 string) (otto.Value, error) {
|
||||
func (m *MockJailCell) Run(arg0 interface{}) (otto.Value, error) {
|
||||
ret := m.ctrl.Call(m, "Run", arg0)
|
||||
ret0, _ := ret[0].(otto.Value)
|
||||
ret1, _ := ret[1].(error)
|
||||
|
|
|
|||
|
|
@ -5,48 +5,11 @@ import (
|
|||
"encoding/json"
|
||||
|
||||
gethcommon "github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
gethrpc "github.com/ethereum/go-ethereum/rpc"
|
||||
"github.com/robertkrimen/otto"
|
||||
"github.com/status-im/status-go/geth/common"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
)
|
||||
|
||||
// map of command routes
|
||||
var (
|
||||
//TODO(influx6): Replace this with a registry of commands to functions that
|
||||
// call appropriate op for command with ExecutionPolicy.
|
||||
rpcLocalCommandRoute = map[string]bool{
|
||||
//Whisper commands
|
||||
"shh_post": true,
|
||||
"shh_version": true,
|
||||
"shh_newIdentity": true,
|
||||
"shh_hasIdentity": true,
|
||||
"shh_newGroup": true,
|
||||
"shh_addToGroup": true,
|
||||
"shh_newFilter": true,
|
||||
"shh_uninstallFilter": true,
|
||||
"shh_getFilterChanges": true,
|
||||
"shh_getMessages": true,
|
||||
|
||||
// DB commands
|
||||
"db_putString": true,
|
||||
"db_getString": true,
|
||||
"db_putHex": true,
|
||||
"db_getHex": true,
|
||||
|
||||
// Other commands
|
||||
"net_version": true,
|
||||
"net_peerCount": true,
|
||||
"net_listening": true,
|
||||
|
||||
// blockchain commands
|
||||
"eth_sign": true,
|
||||
"eth_accounts": true,
|
||||
"eth_getCompilers": true,
|
||||
"eth_compileLLL": true,
|
||||
"eth_compileSolidity": true,
|
||||
"eth_compileSerpent": true,
|
||||
}
|
||||
"github.com/status-im/status-go/geth/rpc"
|
||||
)
|
||||
|
||||
// ExecutionPolicy provides a central container for the executions of RPCCall requests for both
|
||||
|
|
@ -70,48 +33,11 @@ func NewExecutionPolicy(
|
|||
|
||||
// Execute handles the execution of a RPC request and routes appropriately to either a local or remote ethereum node.
|
||||
func (ep *ExecutionPolicy) Execute(req common.RPCCall, call otto.FunctionCall) (*otto.Object, error) {
|
||||
config, err := ep.nodeManager.NodeConfig()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if config.UpstreamConfig.Enabled {
|
||||
if rpcLocalCommandRoute[req.Method] {
|
||||
return ep.ExecuteLocally(req, call)
|
||||
}
|
||||
|
||||
return ep.ExecuteOnRemote(req, call)
|
||||
}
|
||||
|
||||
return ep.ExecuteLocally(req, call)
|
||||
}
|
||||
|
||||
// ExecuteLocally defines a function which handles the processing of all RPC requests from the jail object
|
||||
// to be processed with the internal ethereum node server(light.LightEthereum).
|
||||
func (ep *ExecutionPolicy) ExecuteLocally(req common.RPCCall, call otto.FunctionCall) (*otto.Object, error) {
|
||||
if params.SendTransactionMethodName == req.Method {
|
||||
return ep.executeSendTransaction(req, call)
|
||||
}
|
||||
|
||||
client, err := ep.nodeManager.RPCLocalClient()
|
||||
if err != nil {
|
||||
return nil, common.StopRPCCallError{Err: err}
|
||||
}
|
||||
|
||||
return ep.executeWithClient(client, req, call)
|
||||
}
|
||||
|
||||
// ExecuteOnRemote defines a function which handles the processing of all RPC requests from the jail object
|
||||
// to be processed by a remote ethereum node server with responses returned as needed.
|
||||
func (ep *ExecutionPolicy) ExecuteOnRemote(req common.RPCCall, call otto.FunctionCall) (*otto.Object, error) {
|
||||
if params.SendTransactionMethodName == req.Method {
|
||||
return ep.executeSendTransaction(req, call)
|
||||
}
|
||||
|
||||
client, err := ep.nodeManager.RPCUpstreamClient()
|
||||
if err != nil {
|
||||
return nil, common.StopRPCCallError{Err: err}
|
||||
}
|
||||
client := ep.nodeManager.RPCClient()
|
||||
|
||||
return ep.executeWithClient(client, req, call)
|
||||
}
|
||||
|
|
@ -174,37 +100,28 @@ func (ep *ExecutionPolicy) executeWithClient(client *rpc.Client, req common.RPCC
|
|||
}
|
||||
|
||||
err = client.Call(&result, req.Method, req.Params...)
|
||||
|
||||
switch err := err.(type) {
|
||||
case nil:
|
||||
if result == nil {
|
||||
|
||||
// Special case null because it is decoded as an empty
|
||||
// raw message for some reason.
|
||||
resp.Set("result", otto.NullValue())
|
||||
|
||||
if err != nil {
|
||||
if err2, ok := err.(gethrpc.Error); ok {
|
||||
resp.Set("error", map[string]interface{}{
|
||||
"code": err2.ErrorCode(),
|
||||
"message": err2.Error(),
|
||||
})
|
||||
} else {
|
||||
|
||||
resultVal, callErr := JSON.Call("parse", string(result))
|
||||
|
||||
if callErr != nil {
|
||||
resp = newErrorResponse(call.Otto, -32603, callErr.Error(), &req.ID).Object()
|
||||
} else {
|
||||
resp.Set("result", resultVal)
|
||||
}
|
||||
|
||||
resp = newErrorResponse(call.Otto, -32603, err.Error(), &req.ID).Object()
|
||||
}
|
||||
}
|
||||
|
||||
case rpc.Error:
|
||||
|
||||
resp.Set("error", map[string]interface{}{
|
||||
"code": err.ErrorCode(),
|
||||
"message": err.Error(),
|
||||
})
|
||||
|
||||
default:
|
||||
|
||||
resp = newErrorResponse(call.Otto, -32603, err.Error(), &req.ID).Object()
|
||||
if result == nil {
|
||||
// Special case null because it is decoded as an empty
|
||||
// raw message for some reason.
|
||||
resp.Set("result", otto.NullValue())
|
||||
} else {
|
||||
resultVal, callErr := JSON.Call("parse", string(result))
|
||||
if callErr != nil {
|
||||
resp = newErrorResponse(call.Otto, -32603, callErr.Error(), &req.ID).Object()
|
||||
} else {
|
||||
resp.Set("result", resultVal)
|
||||
}
|
||||
}
|
||||
|
||||
// do extra request post processing (setting back tx context)
|
||||
|
|
|
|||
|
|
@ -74,10 +74,7 @@ func makeSendHandler(jail *Jail) func(call otto.FunctionCall) (response otto.Val
|
|||
// makeJethIsConnectedHandler returns jeth.isConnected() handler
|
||||
func makeJethIsConnectedHandler(jail *Jail) func(call otto.FunctionCall) (response otto.Value) {
|
||||
return func(call otto.FunctionCall) otto.Value {
|
||||
client, err := jail.nodeManager.RPCClient()
|
||||
if err != nil {
|
||||
return newErrorResponse(call.Otto, -32603, err.Error(), nil)
|
||||
}
|
||||
client := jail.nodeManager.RPCClient()
|
||||
|
||||
var netListeningResult bool
|
||||
if err := client.Call(&netListeningResult, "net_listening"); err != nil {
|
||||
|
|
|
|||
|
|
@ -176,9 +176,7 @@ func (jail *Jail) Send(call otto.FunctionCall) (response otto.Value) {
|
|||
|
||||
// Execute the requests.
|
||||
for _, req := range reqs {
|
||||
log.Info("executing a request via jail.policy", "method", req.Method, "params", req.Params)
|
||||
res, err := jail.policy.Execute(req, call)
|
||||
log.Info("response from the request", "err", err)
|
||||
|
||||
if err != nil {
|
||||
switch err.(type) {
|
||||
|
|
@ -208,11 +206,16 @@ func (jail *Jail) Send(call otto.FunctionCall) (response otto.Value) {
|
|||
return response
|
||||
}
|
||||
|
||||
//==========================================================================================================
|
||||
|
||||
func newErrorResponse(vm *otto.Otto, code int, msg string, id interface{}) otto.Value {
|
||||
// Bundle the error into a JSON RPC call response
|
||||
m := map[string]interface{}{"jsonrpc": "2.0", "id": id, "error": map[string]interface{}{"code": code, msg: msg}}
|
||||
m := map[string]interface{}{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"error": map[string]interface{}{
|
||||
"code": code,
|
||||
"message": msg,
|
||||
},
|
||||
}
|
||||
res, _ := json.Marshal(m)
|
||||
val, _ := vm.Run("(" + string(res) + ")")
|
||||
return val
|
||||
|
|
|
|||
|
|
@ -1,234 +0,0 @@
|
|||
package jail
|
||||
|
||||
// import (
|
||||
// "context"
|
||||
|
||||
// gethcommon "github.com/ethereum/go-ethereum/common"
|
||||
// "github.com/ethereum/go-ethereum/common/hexutil"
|
||||
// "github.com/ethereum/go-ethereum/les/status"
|
||||
// "github.com/ethereum/go-ethereum/rpc"
|
||||
// "github.com/robertkrimen/otto"
|
||||
// "github.com/status-im/status-go/geth/common"
|
||||
// )
|
||||
|
||||
// const (
|
||||
// // SendTransactionRequest is triggered on send transaction request
|
||||
// SendTransactionRequest = "eth_sendTransaction"
|
||||
// )
|
||||
|
||||
// // RequestManager represents interface to manage jailed requests.
|
||||
// // Whenever some request passed to a Jail, needs to be pre/post processed,
|
||||
// // request manager is the right place for that.
|
||||
// type RequestManager struct {
|
||||
// nodeManager common.NodeManager
|
||||
// }
|
||||
|
||||
// func NewRequestManager(nodeManager common.NodeManager) *RequestManager {
|
||||
// return &RequestManager{
|
||||
// nodeManager: nodeManager,
|
||||
// }
|
||||
// }
|
||||
|
||||
// // PreProcessRequest pre-processes a given RPC call to a given Otto VM
|
||||
// func (m *RequestManager) PreProcessRequest(vm *otto.Otto, req RPCCall) (string, error) {
|
||||
// messageID := currentMessageID(vm.Context())
|
||||
|
||||
// return messageID, nil
|
||||
// }
|
||||
|
||||
// // PostProcessRequest post-processes a given RPC call to a given Otto VM
|
||||
// func (m *RequestManager) PostProcessRequest(vm *otto.Otto, req RPCCall, messageID string) {
|
||||
// if len(messageID) > 0 {
|
||||
// vm.Call("addContext", nil, messageID, common.MessageIDKey, messageID) // nolint: errcheck
|
||||
// }
|
||||
|
||||
// // set extra markers for queued transaction requests
|
||||
// if req.Method == SendTransactionRequest {
|
||||
// vm.Call("addContext", nil, messageID, SendTransactionRequest, true) // nolint: errcheck
|
||||
// }
|
||||
// }
|
||||
|
||||
// // ProcessSendTransactionRequest processes send transaction request.
|
||||
// // Both pre and post processing happens within this function. Pre-processing
|
||||
// // happens before transaction is send to backend, and post processing occurs
|
||||
// // when backend notifies that transaction sending is complete (either successfully
|
||||
// // or with error)
|
||||
// func (m *RequestManager) ProcessSendTransactionRequest(vm *otto.Otto, req RPCCall) (gethcommon.Hash, error) {
|
||||
// lightEthereum, err := m.nodeManager.LightEthereumService()
|
||||
// if err != nil {
|
||||
// return gethcommon.Hash{}, err
|
||||
// }
|
||||
|
||||
// backend := lightEthereum.StatusBackend
|
||||
|
||||
// messageID, err := m.PreProcessRequest(vm, req)
|
||||
// if err != nil {
|
||||
// return gethcommon.Hash{}, err
|
||||
// }
|
||||
|
||||
// // onSendTransactionRequest() will use context to obtain and release ticket
|
||||
// ctx := context.Background()
|
||||
// ctx = context.WithValue(ctx, common.MessageIDKey, messageID)
|
||||
|
||||
// // this call blocks, up until Complete Transaction is called
|
||||
// txHash, err := backend.SendTransaction(ctx, sendTxArgsFromRPCCall(req))
|
||||
// if err != nil {
|
||||
// return gethcommon.Hash{}, err
|
||||
// }
|
||||
|
||||
// // invoke post processing
|
||||
// m.PostProcessRequest(vm, req, messageID)
|
||||
|
||||
// return txHash, nil
|
||||
// }
|
||||
|
||||
// // RPCClient returns RPC client instance, creating it if necessary.
|
||||
// func (m *RequestManager) RPCClient() (*rpc.Client, error) {
|
||||
// return m.nodeManager.RPCClient()
|
||||
// }
|
||||
|
||||
// // RPCCall represents RPC call parameters
|
||||
// type RPCCall struct {
|
||||
// ID int64
|
||||
// Method string
|
||||
// Params []interface{}
|
||||
// }
|
||||
|
||||
// func sendTxArgsFromRPCCall(req RPCCall) status.SendTxArgs {
|
||||
// if req.Method != SendTransactionRequest { // no need to persist extra state for other requests
|
||||
// return status.SendTxArgs{}
|
||||
// }
|
||||
|
||||
// return status.SendTxArgs{
|
||||
// From: req.parseFromAddress(),
|
||||
// To: req.parseToAddress(),
|
||||
// Value: req.parseValue(),
|
||||
// Data: req.parseData(),
|
||||
// Gas: req.parseGas(),
|
||||
// GasPrice: req.parseGasPrice(),
|
||||
// }
|
||||
// }
|
||||
|
||||
// func (r RPCCall) parseFromAddress() gethcommon.Address {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return gethcommon.HexToAddress("0x")
|
||||
// }
|
||||
|
||||
// from, ok := params["from"].(string)
|
||||
// if !ok {
|
||||
// from = "0x"
|
||||
// }
|
||||
|
||||
// return gethcommon.HexToAddress(from)
|
||||
// }
|
||||
|
||||
// func (r RPCCall) parseToAddress() *gethcommon.Address {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// to, ok := params["to"].(string)
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// address := gethcommon.HexToAddress(to)
|
||||
// return &address
|
||||
// }
|
||||
|
||||
// func (r RPCCall) parseData() hexutil.Bytes {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return hexutil.Bytes("0x")
|
||||
// }
|
||||
|
||||
// data, ok := params["data"].(string)
|
||||
// if !ok {
|
||||
// data = "0x"
|
||||
// }
|
||||
|
||||
// byteCode, err := hexutil.Decode(data)
|
||||
// if err != nil {
|
||||
// byteCode = hexutil.Bytes(data)
|
||||
// }
|
||||
|
||||
// return byteCode
|
||||
// }
|
||||
|
||||
// // nolint: dupl
|
||||
// func (r RPCCall) parseValue() *hexutil.Big {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return nil
|
||||
// //return (*hexutil.Big)(big.NewInt("0x0"))
|
||||
// }
|
||||
|
||||
// inputValue, ok := params["value"].(string)
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// parsedValue, err := hexutil.DecodeBig(inputValue)
|
||||
// if err != nil {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// return (*hexutil.Big)(parsedValue)
|
||||
// }
|
||||
|
||||
// // nolint: dupl
|
||||
// func (r RPCCall) parseGas() *hexutil.Big {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// inputValue, ok := params["gas"].(string)
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// parsedValue, err := hexutil.DecodeBig(inputValue)
|
||||
// if err != nil {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// return (*hexutil.Big)(parsedValue)
|
||||
// }
|
||||
|
||||
// // nolint: dupl
|
||||
// func (r RPCCall) parseGasPrice() *hexutil.Big {
|
||||
// params, ok := r.Params[0].(map[string]interface{})
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// inputValue, ok := params["gasPrice"].(string)
|
||||
// if !ok {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// parsedValue, err := hexutil.DecodeBig(inputValue)
|
||||
// if err != nil {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
// return (*hexutil.Big)(parsedValue)
|
||||
// }
|
||||
|
||||
// // currentMessageID looks for `status.message_id` variable in current JS context
|
||||
// func currentMessageID(ctx otto.Context) string {
|
||||
// if statusObj, ok := ctx.Symbols["status"]; ok {
|
||||
// messageID, err := statusObj.Object().Get("message_id")
|
||||
// if err != nil {
|
||||
// return ""
|
||||
// }
|
||||
// if messageID, err := messageID.ToString(); err == nil {
|
||||
// return messageID
|
||||
// }
|
||||
// }
|
||||
|
||||
// return ""
|
||||
// }
|
||||
|
|
@ -12,10 +12,10 @@ import (
|
|||
"github.com/ethereum/go-ethereum/les"
|
||||
"github.com/ethereum/go-ethereum/node"
|
||||
"github.com/ethereum/go-ethereum/p2p/discover"
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
|
||||
"github.com/status-im/status-go/geth/log"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
"github.com/status-im/status-go/geth/rpc"
|
||||
)
|
||||
|
||||
// errors
|
||||
|
|
@ -27,8 +27,7 @@ var (
|
|||
ErrInvalidLightEthereumService = errors.New("LES service is unavailable")
|
||||
ErrInvalidAccountManager = errors.New("could not retrieve account manager")
|
||||
ErrAccountKeyStoreMissing = errors.New("account key store is not set")
|
||||
ErrInvalidRPCClient = errors.New("RPC client is unavailable")
|
||||
ErrInvalidRPCServer = errors.New("RPC server is unavailable")
|
||||
ErrRPCClient = errors.New("failed to init RPC client")
|
||||
)
|
||||
|
||||
// NodeManager manages Status node (which abstracts contained geth node)
|
||||
|
|
@ -41,7 +40,6 @@ type NodeManager struct {
|
|||
whisperService *whisper.Whisper // reference to Whisper service
|
||||
lesService *les.LightEthereum // reference to LES service
|
||||
rpcClient *rpc.Client // reference to RPC client
|
||||
rpcServer *rpc.Server // reference to RPC server
|
||||
}
|
||||
|
||||
// NewNodeManager makes new instance of node manager
|
||||
|
|
@ -97,6 +95,20 @@ func (m *NodeManager) startNode(config *params.NodeConfig) (<-chan struct{}, err
|
|||
m.node = ethNode
|
||||
m.nodeStopped = make(chan struct{}, 1)
|
||||
m.config = config
|
||||
|
||||
// init RPC client for this node
|
||||
m.rpcClient, err = rpc.NewClient(m.node, m.config.UpstreamConfig)
|
||||
if err != nil {
|
||||
log.Error("Init RPC client failed:", "error", err)
|
||||
m.Unlock()
|
||||
SendSignal(SignalEnvelope{
|
||||
Type: EventNodeCrashed,
|
||||
Event: NodeCrashEvent{
|
||||
Error: ErrRPCClient.Error(),
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
m.Unlock()
|
||||
|
||||
// underlying node is started, every method can use it, we use it immediately
|
||||
|
|
@ -159,7 +171,6 @@ func (m *NodeManager) stopNode() (<-chan struct{}, error) {
|
|||
m.lesService = nil
|
||||
m.whisperService = nil
|
||||
m.rpcClient = nil
|
||||
m.rpcServer = nil
|
||||
m.nodeStarted = nil
|
||||
m.node = nil
|
||||
m.Unlock()
|
||||
|
|
@ -456,90 +467,12 @@ func (m *NodeManager) AccountKeyStore() (*keystore.KeyStore, error) {
|
|||
return keyStore, nil
|
||||
}
|
||||
|
||||
// RPCLocalClient exposes reference to RPC client connected to the running node.
|
||||
func (m *NodeManager) RPCLocalClient() (*rpc.Client, error) {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
if err := m.isNodeAvailable(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
<-m.nodeStarted
|
||||
|
||||
if m.rpcClient == nil {
|
||||
var err error
|
||||
m.rpcClient, err = m.node.Attach()
|
||||
if err != nil {
|
||||
log.Error("Cannot attach RPC client to node", "error", err)
|
||||
return nil, ErrInvalidRPCClient
|
||||
}
|
||||
}
|
||||
|
||||
return m.rpcClient, nil
|
||||
}
|
||||
|
||||
// RPCUpstreamClient exposes reference to RPC client connected to the running node.
|
||||
func (m *NodeManager) RPCUpstreamClient() (*rpc.Client, error) {
|
||||
config, err := m.NodeConfig()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
if m.rpcClient == nil {
|
||||
m.rpcClient, err = rpc.Dial(config.UpstreamConfig.URL)
|
||||
if err != nil {
|
||||
log.Error("Failed to connect to upstream RPC server", "error", err)
|
||||
return nil, ErrInvalidRPCClient
|
||||
}
|
||||
}
|
||||
|
||||
return m.rpcClient, nil
|
||||
}
|
||||
|
||||
// RPCClient exposes reference to RPC client connected to the running node.
|
||||
func (m *NodeManager) RPCClient() (*rpc.Client, error) {
|
||||
config, err := m.NodeConfig()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
func (m *NodeManager) RPCClient() *rpc.Client {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
// Connect to upstream RPC server with new client and cache instance.
|
||||
if config.UpstreamConfig.Enabled {
|
||||
return m.RPCUpstreamClient()
|
||||
}
|
||||
|
||||
return m.RPCLocalClient()
|
||||
}
|
||||
|
||||
// RPCServer exposes reference to running node's in-proc RPC server/handler
|
||||
func (m *NodeManager) RPCServer() (*rpc.Server, error) {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
|
||||
if err := m.isNodeAvailable(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
<-m.nodeStarted
|
||||
|
||||
if m.rpcServer == nil {
|
||||
var err error
|
||||
m.rpcServer, err = m.node.InProcRPC()
|
||||
if err != nil {
|
||||
log.Error("Cannot expose on-proc RPC server", "error", err)
|
||||
return nil, ErrInvalidRPCServer
|
||||
}
|
||||
}
|
||||
|
||||
if m.rpcServer == nil {
|
||||
return nil, ErrInvalidRPCServer
|
||||
}
|
||||
|
||||
return m.rpcServer, nil
|
||||
return m.rpcClient
|
||||
}
|
||||
|
||||
// initLog initializes global logger parameters based on
|
||||
|
|
|
|||
|
|
@ -116,14 +116,7 @@ func (s *ManagerTestSuite) TestReferences() {
|
|||
{
|
||||
"non-null manager, no running node, get RPC Client",
|
||||
func() (interface{}, error) {
|
||||
return s.NodeManager.RPCClient()
|
||||
},
|
||||
node.ErrNoRunningNode,
|
||||
},
|
||||
{
|
||||
"non-null manager, no running node, get RPC Server",
|
||||
func() (interface{}, error) {
|
||||
return s.NodeManager.RPCServer()
|
||||
return s.NodeManager.RPCClient(), nil
|
||||
},
|
||||
node.ErrNoRunningNode,
|
||||
},
|
||||
|
|
@ -188,17 +181,10 @@ func (s *ManagerTestSuite) TestReferences() {
|
|||
{
|
||||
"node is running, get RPC Client",
|
||||
func() (interface{}, error) {
|
||||
return s.NodeManager.RPCClient()
|
||||
return s.NodeManager.RPCClient(), nil
|
||||
},
|
||||
&rpc.Client{},
|
||||
},
|
||||
{
|
||||
"node is running, get RPC Server",
|
||||
func() (interface{}, error) {
|
||||
return s.NodeManager.RPCServer()
|
||||
},
|
||||
&rpc.Server{},
|
||||
},
|
||||
}
|
||||
for _, testCase := range nodeReadyTestCases {
|
||||
obj, err := testCase.initFn()
|
||||
|
|
@ -412,14 +398,7 @@ func (s *ManagerTestSuite) TestRaceConditions() {
|
|||
},
|
||||
func(config *params.NodeConfig) {
|
||||
log.Info("RPCClient()")
|
||||
_, err := s.NodeManager.RPCClient()
|
||||
s.T().Logf("RPCClient(), error: %v", err)
|
||||
progress <- struct{}{}
|
||||
},
|
||||
func(config *params.NodeConfig) {
|
||||
log.Info("RPCServer()")
|
||||
_, err := s.NodeManager.RPCServer()
|
||||
s.T().Logf("RPCServer(), error: %v", err)
|
||||
s.NodeManager.RPCClient()
|
||||
progress <- struct{}{}
|
||||
},
|
||||
}
|
||||
|
|
|
|||
179
geth/node/rpc.go
179
geth/node/rpc.go
|
|
@ -1,179 +0,0 @@
|
|||
package node
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"net/url"
|
||||
|
||||
"github.com/status-im/status-go/geth/common"
|
||||
"github.com/status-im/status-go/geth/log"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
)
|
||||
|
||||
const (
|
||||
jsonrpcVersion = "2.0"
|
||||
)
|
||||
|
||||
type jsonRequest struct {
|
||||
Method string `json:"method"`
|
||||
Version string `json:"jsonrpc"`
|
||||
ID int `json:"id,omitempty"`
|
||||
Payload json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type jsonError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Data interface{} `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
type jsonErrResponse struct {
|
||||
Version string `json:"jsonrpc"`
|
||||
ID interface{} `json:"id,omitempty"`
|
||||
Error jsonError `json:"error"`
|
||||
}
|
||||
|
||||
// RPCManager abstract RPC management API (for both client and server)
|
||||
type RPCManager struct {
|
||||
sync.Mutex
|
||||
requestID int
|
||||
nodeManager common.NodeManager
|
||||
}
|
||||
|
||||
// errors
|
||||
var (
|
||||
ErrInvalidMethod = errors.New("method does not exist")
|
||||
ErrRPCServerTimeout = errors.New("RPC server cancelled call due to timeout")
|
||||
ErrRPCServerCallFailed = errors.New("RPC server cannot complete request")
|
||||
)
|
||||
|
||||
// NewRPCManager returns new instance of RPC client
|
||||
func NewRPCManager(nodeManager common.NodeManager) *RPCManager {
|
||||
return &RPCManager{
|
||||
nodeManager: nodeManager,
|
||||
}
|
||||
}
|
||||
|
||||
// Call executes RPC request on node's in-proc RPC server
|
||||
func (c *RPCManager) Call(inputJSON string) string {
|
||||
config, err := c.nodeManager.NodeConfig()
|
||||
if err != nil {
|
||||
return c.makeJSONErrorResponse(err)
|
||||
}
|
||||
|
||||
// allow HTTP requests to block w/o
|
||||
outputJSON := make(chan string, 1)
|
||||
|
||||
go func() {
|
||||
body := bytes.NewBufferString(inputJSON)
|
||||
|
||||
var err error
|
||||
var res []byte
|
||||
|
||||
if config.UpstreamConfig.Enabled {
|
||||
log.Info("Making RPC JSON Request to upstream RPCServer")
|
||||
res, err = c.callUpstreamStream(config, body)
|
||||
} else {
|
||||
log.Info("Making RPC JSON Request to internal RPCServer")
|
||||
res, err = c.callNodeStream(body)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
outputJSON <- c.makeJSONErrorResponse(err)
|
||||
return
|
||||
}
|
||||
|
||||
outputJSON <- string(res)
|
||||
return
|
||||
}()
|
||||
|
||||
// wait till call is complete
|
||||
select {
|
||||
case out := <-outputJSON:
|
||||
return out
|
||||
case <-time.After((DefaultTxSendCompletionTimeout + 10) * time.Minute): // give up eventually
|
||||
// pass
|
||||
}
|
||||
|
||||
return c.makeJSONErrorResponse(ErrRPCServerTimeout)
|
||||
}
|
||||
|
||||
// callNodeStream delivers giving request and body content to the external ethereum
|
||||
// (infura) RPCServer to process the request and returns response.
|
||||
func (c *RPCManager) callUpstreamStream(config *params.NodeConfig, body io.Reader) ([]byte, error) {
|
||||
upstreamURL, err := url.Parse(config.UpstreamConfig.URL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
httpReq, err := http.NewRequest("POST", upstreamURL.String(), body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
httpClient := http.Client{
|
||||
Timeout: 20 * time.Second,
|
||||
}
|
||||
|
||||
res, err := httpClient.Do(httpReq)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer res.Body.Close()
|
||||
|
||||
if respStatusCode := res.StatusCode; respStatusCode != http.StatusOK {
|
||||
log.Error("handler returned wrong status code", "got", respStatusCode, "want", http.StatusOK)
|
||||
return nil, ErrRPCServerCallFailed
|
||||
}
|
||||
|
||||
return ioutil.ReadAll(res.Body)
|
||||
}
|
||||
|
||||
// callNodeStream delivers giving request and body content to the internal ethereum
|
||||
// RPCServer to process the request.
|
||||
func (c *RPCManager) callNodeStream(body io.Reader) ([]byte, error) {
|
||||
server, err := c.nodeManager.RPCServer()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
httpReq, err := http.NewRequest("POST", "/", body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rr := httptest.NewRecorder()
|
||||
|
||||
server.ServeHTTP(rr, httpReq)
|
||||
|
||||
// Check the status code is what we expect.
|
||||
if respStatus := rr.Code; respStatus != http.StatusOK {
|
||||
log.Error("handler returned wrong status code", "got", respStatus, "want", http.StatusOK)
|
||||
// outputJSON <- c.makeJSONErrorResponse(ErrRPCServerCallFailed)
|
||||
return nil, ErrRPCServerCallFailed
|
||||
}
|
||||
|
||||
return rr.Body.Bytes(), nil
|
||||
}
|
||||
|
||||
// makeJSONErrorResponse returns error as JSON response
|
||||
func (c *RPCManager) makeJSONErrorResponse(err error) string {
|
||||
response := jsonErrResponse{
|
||||
Version: jsonrpcVersion,
|
||||
Error: jsonError{
|
||||
Message: err.Error(),
|
||||
},
|
||||
}
|
||||
|
||||
outBytes, _ := json.Marshal(&response)
|
||||
return string(outBytes)
|
||||
}
|
||||
|
|
@ -210,18 +210,14 @@ func (m *TxQueueManager) completeRemoteTransaction(queuedTx *common.QueuedTx, pa
|
|||
return emptyHash, err
|
||||
}
|
||||
|
||||
client, err := m.nodeManager.RPCUpstreamClient()
|
||||
if err != nil {
|
||||
return emptyHash, err
|
||||
}
|
||||
|
||||
// We need to request a new transaction nounce from upstream node.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
|
||||
var txCount hexutil.Uint
|
||||
if callErr := client.CallContext(ctx, &txCount, "eth_getTransactionCount", queuedTx.Args.From, "pending"); callErr != nil {
|
||||
return emptyHash, callErr
|
||||
client := m.nodeManager.RPCClient()
|
||||
if err := client.CallContext(ctx, &txCount, "eth_getTransactionCount", queuedTx.Args.From, "pending"); err != nil {
|
||||
return emptyHash, err
|
||||
}
|
||||
|
||||
chainID := big.NewInt(int64(config.NetworkID))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,143 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
gethrpc "github.com/ethereum/go-ethereum/rpc"
|
||||
"github.com/status-im/status-go/geth/log"
|
||||
)
|
||||
|
||||
const (
|
||||
jsonrpcVersion = "2.0"
|
||||
errInvalidMessageCode = -32700 // from go-ethereum/rpc/errors.go
|
||||
)
|
||||
|
||||
// for JSON-RPC responses obtained via CallRaw(), we have no way
|
||||
// to know ID field from actual response. web3.js (primary and
|
||||
// only user of CallRaw()) will validate response by checking
|
||||
// ID field for being a number:
|
||||
// https://github.com/ethereum/web3.js/blob/develop/lib/web3/jsonrpc.js#L66
|
||||
// thus, we will use zero ID as a workaround of this limitation
|
||||
var defaultMsgID = json.RawMessage(`0`)
|
||||
|
||||
// CallRaw performs a JSON-RPC call with already crafted JSON-RPC body. It
|
||||
// returns string in JSON format with response (successul or error).
|
||||
func (c *Client) CallRaw(body string) string {
|
||||
ctx := context.Background()
|
||||
return c.callRawContext(ctx, body)
|
||||
}
|
||||
|
||||
// jsonrpcMessage represents JSON-RPC request, notification, successful response or
|
||||
// error response.
|
||||
type jsonrpcMessage struct {
|
||||
Version string `json:"jsonrpc"`
|
||||
ID json.RawMessage `json:"id,omitempty"`
|
||||
Method string `json:"method,omitempty"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
Error *jsonError `json:"error,omitempty"`
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
}
|
||||
|
||||
// jsonError represents Error message for JSON-RPC responses.
|
||||
type jsonError struct {
|
||||
Code int `json:"code,omitempty"`
|
||||
Message string `json:"message"`
|
||||
Data interface{} `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
// callRawContext performs a JSON-RPC call with already crafted JSON-RPC body and
|
||||
// given context. It returns string in JSON format with response (successul or error).
|
||||
//
|
||||
// TODO(divan): this function exists for compatibility and uses default
|
||||
// go-ethereum's RPC client under the hood. It adds some unnecessary overhead
|
||||
// by first marshalling JSON string into object to use with normal Call,
|
||||
// which is then umarshalled back to the same JSON. The same goes with response.
|
||||
// This is waste of CPU and memory and should be avoided if possible,
|
||||
// either by changing exported API (provide only Call, not CallRaw) or
|
||||
// refactoring go-ethereum's client to allow using raw JSON directly.
|
||||
func (c *Client) callRawContext(ctx context.Context, body string) string {
|
||||
// unmarshal JSON body into json-rpc request
|
||||
method, params, id, err := methodAndParamsFromBody(body)
|
||||
if err != nil {
|
||||
return newErrorResponse(errInvalidMessageCode, err, id)
|
||||
}
|
||||
|
||||
// route and execute
|
||||
var result json.RawMessage
|
||||
err = c.CallContext(ctx, &result, method, params...)
|
||||
|
||||
// as we have to return original JSON, we have to
|
||||
// analyze returned error and reconstruct original
|
||||
// JSON error response.
|
||||
if err != nil && err != gethrpc.ErrNoResult {
|
||||
if er, ok := err.(gethrpc.Error); ok {
|
||||
return newErrorResponse(er.ErrorCode(), err, id)
|
||||
}
|
||||
|
||||
return newErrorResponse(errInvalidMessageCode, err, id)
|
||||
}
|
||||
|
||||
// finally, marshal answer
|
||||
return newSuccessResponse(result, id)
|
||||
}
|
||||
|
||||
// methodAndParamsFromBody extracts Method and Params of
|
||||
// JSON-RPC body into values ready to use with ethereum-go's
|
||||
// RPC client Call() function. A lot of empty interface usage is
|
||||
// due to the underlying code design :/
|
||||
func methodAndParamsFromBody(body string) (string, []interface{}, json.RawMessage, error) {
|
||||
msg, err := unmarshalMessage(body)
|
||||
if err != nil {
|
||||
return "", nil, nil, err
|
||||
}
|
||||
|
||||
params := []interface{}{}
|
||||
if msg.Params != nil {
|
||||
err = json.Unmarshal(msg.Params, ¶ms)
|
||||
if err != nil {
|
||||
log.Error("unmarshal params", "error", err)
|
||||
return "", nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return msg.Method, params, msg.ID, nil
|
||||
}
|
||||
|
||||
func unmarshalMessage(body string) (*jsonrpcMessage, error) {
|
||||
var msg jsonrpcMessage
|
||||
err := json.Unmarshal([]byte(body), &msg)
|
||||
return &msg, err
|
||||
}
|
||||
|
||||
func newSuccessResponse(result json.RawMessage, id json.RawMessage) string {
|
||||
if id == nil {
|
||||
id = defaultMsgID
|
||||
}
|
||||
|
||||
msg := &jsonrpcMessage{
|
||||
ID: id,
|
||||
Version: jsonrpcVersion,
|
||||
Result: result,
|
||||
}
|
||||
data, _ := json.Marshal(msg)
|
||||
return string(data)
|
||||
}
|
||||
|
||||
func newErrorResponse(code int, err error, id json.RawMessage) string {
|
||||
if id == nil {
|
||||
id = defaultMsgID
|
||||
}
|
||||
|
||||
errMsg := &jsonrpcMessage{
|
||||
Version: jsonrpcVersion,
|
||||
ID: id,
|
||||
Error: &jsonError{
|
||||
Code: code,
|
||||
Message: err.Error(),
|
||||
},
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(errMsg)
|
||||
return string(data)
|
||||
}
|
||||
|
|
@ -0,0 +1,96 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNewSuccessResponse(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
result json.RawMessage
|
||||
id json.RawMessage
|
||||
expected string
|
||||
}{
|
||||
{"string", json.RawMessage(`"3434=done"`), nil, `{"jsonrpc":"2.0","id":0,"result":"3434=done"}`},
|
||||
{"struct_nil_id", json.RawMessage(`{"field": "value"}`), nil, `{"jsonrpc":"2.0","id":0,"result":{"field":"value"}}`},
|
||||
{"struct_non_nil_id", json.RawMessage(`{"field": "value"}`), json.RawMessage(`42`), `{"jsonrpc":"2.0","id":42,"result":{"field":"value"}}`},
|
||||
}
|
||||
|
||||
for _, test := range cases {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
got := newSuccessResponse(test.result, test.id)
|
||||
require.Equal(t, test.expected, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewErrorResponse(t *testing.T) {
|
||||
got := newErrorResponse(-32601, errors.New("Method not found"), json.RawMessage(`42`))
|
||||
|
||||
expected := `{"jsonrpc":"2.0","id":42,"error":{"code":-32601,"message":"Method not found"}}`
|
||||
require.Equal(t, expected, got)
|
||||
}
|
||||
|
||||
func TestUnmarshalMessage(t *testing.T) {
|
||||
body := `{"jsonrpc": "2.0", "method": "subtract", "params": {"subtrahend": 23, "minuend": 42}}`
|
||||
got, err := unmarshalMessage(body)
|
||||
require.NoError(t, err)
|
||||
|
||||
expected := &jsonrpcMessage{
|
||||
Version: "2.0",
|
||||
Method: "subtract",
|
||||
Params: json.RawMessage(`{"subtrahend": 23, "minuend": 42}`),
|
||||
}
|
||||
require.Equal(t, expected, got)
|
||||
}
|
||||
|
||||
func TestMethodAndParamsFromBody(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
body string
|
||||
params []interface{}
|
||||
method string
|
||||
id json.RawMessage
|
||||
}{
|
||||
{
|
||||
"params_array",
|
||||
`{"jsonrpc": "2.0", "id": 42, "method": "subtract", "params": [{"subtrahend": 23, "minuend": 42}]}`,
|
||||
[]interface{}{
|
||||
map[string]interface{}{
|
||||
"subtrahend": float64(23),
|
||||
"minuend": float64(42),
|
||||
},
|
||||
},
|
||||
"subtract",
|
||||
json.RawMessage(`42`),
|
||||
},
|
||||
{
|
||||
"params_empty_array",
|
||||
`{"jsonrpc": "2.0", "method": "test", "params": []}`,
|
||||
[]interface{}{},
|
||||
"test",
|
||||
json.RawMessage(nil),
|
||||
},
|
||||
{
|
||||
"params_none",
|
||||
`{"jsonrpc": "2.0", "method": "test"}`,
|
||||
[]interface{}{},
|
||||
"test",
|
||||
json.RawMessage(nil),
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range cases {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
method, params, id, err := methodAndParamsFromBody(test.body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, test.method, method)
|
||||
require.Equal(t, test.params, params)
|
||||
require.Equal(t, test.id, id)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,80 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/ethereum/go-ethereum/node"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
|
||||
gethrpc "github.com/ethereum/go-ethereum/rpc"
|
||||
)
|
||||
|
||||
// Client represents RPC client with custom routing
|
||||
// scheme. It automatically decides where RPC call
|
||||
// goes - Upstream or Local node.
|
||||
type Client struct {
|
||||
upstreamEnabled bool
|
||||
upstreamURL string
|
||||
|
||||
local *gethrpc.Client
|
||||
upstream *gethrpc.Client
|
||||
|
||||
router *router
|
||||
}
|
||||
|
||||
// NewClient initializes Client and tries to connect to both,
|
||||
// upstream and local node.
|
||||
//
|
||||
// Client is safe for concurrent use and will automatically
|
||||
// reconnect to the server if connection is lost.
|
||||
func NewClient(node *node.Node, upstream params.UpstreamRPCConfig) (*Client, error) {
|
||||
c := &Client{}
|
||||
|
||||
var err error
|
||||
c.local, err = node.Attach()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("attach to local node: %s", err)
|
||||
}
|
||||
|
||||
if upstream.Enabled {
|
||||
c.upstreamEnabled = upstream.Enabled
|
||||
c.upstreamURL = upstream.URL
|
||||
|
||||
c.upstream, err = gethrpc.Dial(c.upstreamURL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dial upstream server: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
c.router = newRouter(c.upstreamEnabled)
|
||||
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Call performs a JSON-RPC call with the given arguments and unmarshals into
|
||||
// result if no error occurred.
|
||||
//
|
||||
// The result must be a pointer so that package json can unmarshal into it. You
|
||||
// can also pass nil, in which case the result is ignored.
|
||||
//
|
||||
// It uses custom routing scheme for calls.
|
||||
func (c *Client) Call(result interface{}, method string, args ...interface{}) error {
|
||||
ctx := context.Background()
|
||||
return c.CallContext(ctx, result, method, args...)
|
||||
}
|
||||
|
||||
// CallContext performs a JSON-RPC call with the given arguments. If the context is
|
||||
// canceled before the call has successfully returned, CallContext returns immediately.
|
||||
//
|
||||
// The result must be a pointer so that package json can unmarshal into it. You
|
||||
// can also pass nil, in which case the result is ignored.
|
||||
//
|
||||
// It uses custom routing scheme for calls.
|
||||
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
|
||||
if c.router.routeLocally(method) {
|
||||
return c.local.CallContext(ctx, result, method, args...)
|
||||
}
|
||||
|
||||
return c.upstream.CallContext(ctx, result, method, args...)
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package node_test
|
||||
package rpc_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
|
@ -9,6 +9,7 @@ import (
|
|||
"github.com/status-im/status-go/geth/log"
|
||||
"github.com/status-im/status-go/geth/node"
|
||||
"github.com/status-im/status-go/geth/params"
|
||||
"github.com/status-im/status-go/geth/rpc"
|
||||
. "github.com/status-im/status-go/geth/testing"
|
||||
"github.com/stretchr/testify/suite"
|
||||
)
|
||||
|
|
@ -44,9 +45,52 @@ func (s *RPCTestSuite) SetupTest() {
|
|||
s.NodeManager = nodeManager
|
||||
}
|
||||
|
||||
func (s *RPCTestSuite) TestNewClient() {
|
||||
require := s.Require()
|
||||
|
||||
config, err := MakeTestNodeConfig(params.RinkebyNetworkID)
|
||||
require.NoError(err)
|
||||
|
||||
nodeStarted, err := s.NodeManager.StartNode(config)
|
||||
require.NoError(err)
|
||||
require.NotNil(config)
|
||||
|
||||
<-nodeStarted
|
||||
|
||||
node, err := s.NodeManager.Node()
|
||||
require.NoError(err)
|
||||
|
||||
// upstream disabled, local node ok
|
||||
_, err = rpc.NewClient(node, config.UpstreamConfig)
|
||||
require.NoError(err)
|
||||
|
||||
// upstream enabled with incorrect URL, local node ok
|
||||
upstreamBad := config.UpstreamConfig
|
||||
upstreamBad.Enabled = true
|
||||
upstreamBad.URL = "///__httphh://///incorrect_urlxxx"
|
||||
_, err = rpc.NewClient(node, upstreamBad)
|
||||
require.NotNil(err)
|
||||
|
||||
// upstream enabled with correct URL, local node ok
|
||||
upstreamGood := config.UpstreamConfig
|
||||
upstreamGood.Enabled = true
|
||||
upstreamGood.URL = "http://example.com/rpc"
|
||||
_, err = rpc.NewClient(node, upstreamGood)
|
||||
require.Nil(err)
|
||||
|
||||
// upstream disabled, local node failed (stopped)
|
||||
nodeStopped, err := s.NodeManager.StopNode()
|
||||
require.NoError(err)
|
||||
|
||||
<-nodeStopped
|
||||
|
||||
_, err = rpc.NewClient(node, config.UpstreamConfig)
|
||||
require.NotNil(err)
|
||||
}
|
||||
|
||||
func (s *RPCTestSuite) TestRPCSendTransaction() {
|
||||
require := s.Require()
|
||||
expectedResponse := []byte(`{"jsonrpc": "2.0", "status":200, "result": "3434=done"}`)
|
||||
expectedResponse := []byte(`{"jsonrpc":"2.0","id":10,"result":"3434=done"}`)
|
||||
|
||||
// httpRPCServer will serve as an upstream server accepting transactions.
|
||||
httpRPCServer := httptest.NewServer(service{
|
||||
|
|
@ -59,7 +103,7 @@ func (s *RPCTestSuite) TestRPCSendTransaction() {
|
|||
|
||||
if txReq.Method == "eth_getTransactionCount" {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(`{"jsonrpc": "2.0", "status":200, "result": "0x434"}`))
|
||||
w.Write([]byte(`{"jsonrpc": "2.0", "result": "0x434"}`))
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -79,10 +123,10 @@ func (s *RPCTestSuite) TestRPCSendTransaction() {
|
|||
s.StartTestNode(params.RopstenNetworkID, WithUpstream(httpRPCServer.URL))
|
||||
defer s.StopTestNode()
|
||||
|
||||
rpcClient := node.NewRPCManager(s.NodeManager)
|
||||
rpcClient := s.NodeManager.RPCClient()
|
||||
require.NotNil(rpcClient)
|
||||
|
||||
response := rpcClient.Call(`{
|
||||
response := rpcClient.CallRaw(`{
|
||||
"jsonrpc": "2.0",
|
||||
"id":10,
|
||||
"method": "eth_sendTransaction",
|
||||
|
|
@ -104,9 +148,6 @@ func (s *RPCTestSuite) TestCallRPC() {
|
|||
require := s.Require()
|
||||
require.NotNil(s.NodeManager)
|
||||
|
||||
rpcClient := node.NewRPCManager(s.NodeManager)
|
||||
require.NotNil(rpcClient)
|
||||
|
||||
nodeConfig, err := MakeTestNodeConfig(params.RinkebyNetworkID)
|
||||
require.NoError(err)
|
||||
|
||||
|
|
@ -122,6 +163,9 @@ func (s *RPCTestSuite) TestCallRPC() {
|
|||
|
||||
<-nodeStarted
|
||||
|
||||
rpcClient := s.NodeManager.RPCClient()
|
||||
require.NotNil(rpcClient)
|
||||
|
||||
progress := make(chan struct{}, 25)
|
||||
type rpcCall struct {
|
||||
inputJSON string
|
||||
|
|
@ -135,37 +179,35 @@ func (s *RPCTestSuite) TestCallRPC() {
|
|||
"gas": "0x76c0",
|
||||
"gasPrice": "0x9184e72a000",
|
||||
"value": "0x9184e72a",
|
||||
"data": "0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675"}],"id":1}`,
|
||||
"data": "0xd46e8dd67c5d32be8d46e8dd67c5d32be8058bb8eb970870f072445675058bb8eb970870f072445675"}],
|
||||
"id": 1
|
||||
}`,
|
||||
func(resultJSON string) {
|
||||
log.Info("eth_sendTransaction")
|
||||
s.T().Log("GOT: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
},
|
||||
},
|
||||
{
|
||||
`{"jsonrpc":"2.0","method":"shh_version","params":[],"id":67}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"5.0"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"5.0"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("shh_version: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
},
|
||||
},
|
||||
{
|
||||
`{"jsonrpc":"2.0","method":"web3_sha3","params":["0x68656c6c6f20776f726c64"],"id":64}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":64,"result":"0x47173285a8d7341e5e972fc677286384f802f8ef42a5ec5f03bbfa254cb01fad"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("web3_sha3: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
},
|
||||
},
|
||||
{
|
||||
`{"jsonrpc":"2.0","method":"net_version","params":[],"id":67}`,
|
||||
func(resultJSON string) {
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"4"}` + "\n"
|
||||
expected := `{"jsonrpc":"2.0","id":67,"result":"4"}`
|
||||
s.Equal(expected, resultJSON)
|
||||
s.T().Log("net_version: ", resultJSON)
|
||||
progress <- struct{}{}
|
||||
},
|
||||
},
|
||||
|
|
@ -174,8 +216,7 @@ func (s *RPCTestSuite) TestCallRPC() {
|
|||
cnt := len(rpcCalls) - 1 // send transaction blocks up until complete/discarded/times out
|
||||
for _, r := range rpcCalls {
|
||||
go func(r rpcCall) {
|
||||
s.T().Logf("Run test: %v", r.inputJSON)
|
||||
resultJSON := rpcClient.Call(r.inputJSON)
|
||||
resultJSON := rpcClient.CallRaw(r.inputJSON)
|
||||
r.validator(resultJSON)
|
||||
}(r)
|
||||
}
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
package rpc
|
||||
|
||||
// router implements logic for routing
|
||||
// JSON-RPC requests either to Upstream or
|
||||
// Local node.
|
||||
type router struct {
|
||||
methods map[string]bool
|
||||
upstreamEnabled bool
|
||||
}
|
||||
|
||||
// newRouter inits new router.
|
||||
func newRouter(upstreamEnabled bool) *router {
|
||||
r := &router{
|
||||
methods: make(map[string]bool),
|
||||
upstreamEnabled: upstreamEnabled,
|
||||
}
|
||||
|
||||
for _, m := range localMethods {
|
||||
r.methods[m] = true
|
||||
}
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// routeLocally returns true if given method should be routed to
|
||||
// the local node
|
||||
func (r *router) routeLocally(method string) bool {
|
||||
// if upstream is disabled, always route to
|
||||
// the local node
|
||||
if !r.upstreamEnabled {
|
||||
return true
|
||||
}
|
||||
|
||||
// else check route using the methods list
|
||||
return r.methods[method]
|
||||
}
|
||||
|
||||
// localMethods contains methods that should be routed to
|
||||
// the local node; the rest is considered to be routed to
|
||||
// the upstream node.
|
||||
// TODO: in the future, default option may be "local node",
|
||||
// so it'd be convinient to keep track of "remoteMethods" here.
|
||||
var localMethods = [...]string{
|
||||
//Whisper commands
|
||||
"shh_post",
|
||||
"shh_version",
|
||||
"shh_newIdentity",
|
||||
"shh_hasIdentity",
|
||||
"shh_newGroup",
|
||||
"shh_addToGroup",
|
||||
"shh_newFilter",
|
||||
"shh_uninstallFilter",
|
||||
"shh_getFilterChanges",
|
||||
"shh_getMessages",
|
||||
|
||||
// DB commands
|
||||
"db_putString",
|
||||
"db_getString",
|
||||
"db_putHex",
|
||||
"db_getHex",
|
||||
|
||||
// Other commands
|
||||
"net_version",
|
||||
"net_peerCount",
|
||||
"net_listening",
|
||||
|
||||
// blockchain commands
|
||||
"eth_sign",
|
||||
"eth_accounts",
|
||||
"eth_getCompilers",
|
||||
"eth_compileLLL",
|
||||
"eth_compileSolidity",
|
||||
"eth_compileSerpent",
|
||||
}
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"github.com/stretchr/testify/require"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// some of the upstream examples
|
||||
var upstreamMethods = []string{"some_weirdo_method", "eth_syncing", "eth_getBalance", "eth_call", "eth_getTransactionReceipt"}
|
||||
|
||||
func TestRouteWithUpstream(t *testing.T) {
|
||||
router := newRouter(true)
|
||||
|
||||
for _, method := range localMethods {
|
||||
require.True(t, router.routeLocally(method))
|
||||
}
|
||||
|
||||
for _, method := range upstreamMethods {
|
||||
require.False(t, router.routeLocally(method))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouteWithoutUpstream(t *testing.T) {
|
||||
router := newRouter(false)
|
||||
|
||||
for _, method := range localMethods {
|
||||
require.True(t, router.routeLocally(method))
|
||||
}
|
||||
|
||||
for _, method := range upstreamMethods {
|
||||
require.True(t, router.routeLocally(method))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue