以太坊原始碼深入分析(3)-- 以太坊RPC通訊例項和原理程式碼分析(上)
阿新 • • 發佈:2018-12-31
上一節提到,以太坊在node start的時候啟動了RPC服務,以太坊通過Rpc服務來實現以太坊相關介面的遠端呼叫。這節我們用個例項來看看以太坊 RPC是如何工作的,以及以太坊RPC的原始碼的實現
通過以下命令修改預設地址和埠:
同源策略的限制,請求將失敗。
InProc IPC Http Ws這些RPC endpoint,並把收集的APIs傳給這些RPC endpoint。
如果任何一個RPC啟動失敗,結束所有RPC endpoint,並返回err。
我們先看看比較常用的HTTP RPC的實現。
和以太坊的http Rpc server建立流程大致一樣,不過以太坊的APIs並不是按規範的RPC介面寫的,因此以太坊的RPC做了註冊方法的時候做了些特殊處理。
過濾白名單的介面,白名單在defaultConfig裡面配置。
api的結構體:
suitableCallbacks方法獲取Service所有的的方法和符合訂閱標準的方法。
將Service的所有方法放入map s.services, service的name作為map key。
ethclient.go做了個go-ethereum 客戶端請求的client例項。
ethclient例項建立
根據url的scheme判斷呼叫撥號哪個RPC server
如果是http的話直接生成httpConn就返回newClient了,其他方式的話會複雜一些。
json.NewDecoder()對請求返回值進行json序列化,然後send進管道op.resp
一,RPC通訊例項
1,RPC啟動命令 :
geth --rpc
go-ethereum的RPC服務預設地址:http://localhost:8545/通過以下命令修改預設地址和埠:
geth --rpc --rpcaddr < ip > --rpcport < portnumber >
如果從瀏覽器訪問RPC,CORS將需要啟用相應的域集。否則,JavaScript呼叫受到同源策略的限制,請求將失敗。
geth --rpc --rpccorsdomain “ http:// localhost:3000 ”
也可以使用該命令在geth console 啟動admin.startRPC(addr, port)
2, 用curl模擬RPC請求
我們請求一個最簡單的一個eth模組的RPC介面:eth_blockNumbercurl -H "content-Type:application/json" -X POST --data '{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":83}' http://localhost:8545
返回結果:{"jsonrpc":"2.0","id":83,"result":"0x4eb2e8"}
二, Go-ethereum RPC的原始碼分析
1,Node.start 呼叫startRPC()
startRPC方法 收集Node裡面所有service的 APIs。然後分別啟動了func (n *Node) startRPC(services map[reflect.Type]Service) error { // Gather all the possible APIs to surface apis := n.apis() for _, service := range services { apis = append(apis, service.APIs()...) } // Start the various API endpoints, terminating all in case of errors if err := n.startInProc(apis); err != nil { return err } if err := n.startIPC(apis); err != nil { n.stopInProc() return err } if err := n.startHTTP(n.httpEndpoint, apis, n.config.HTTPModules, n.config.HTTPCors, n.config.HTTPVirtualHosts); err != nil { n.stopIPC() n.stopInProc() return err } if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil { n.stopHTTP() n.stopIPC() n.stopInProc() return err } // All API endpoints started successfully n.rpcAPIs = apis return nil }
InProc IPC Http Ws這些RPC endpoint,並把收集的APIs傳給這些RPC endpoint。
如果任何一個RPC啟動失敗,結束所有RPC endpoint,並返回err。
我們先看看比較常用的HTTP RPC的實現。
2,Http RPC server建立過程
有寫過go rpc經驗的同學大概都知道標準的go rpc server建立流程大概是:寫符合規範的RPC server介面-->new server(實現serverHttp()方法)-->RPC 註冊server-->RPC HandleHTTP()-->net.Listen 埠和地址-->http.server(listen )和以太坊的http Rpc server建立流程大致一樣,不過以太坊的APIs並不是按規範的RPC介面寫的,因此以太坊的RPC做了註冊方法的時候做了些特殊處理。
func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string) error { // Short circuit if the HTTP 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 } n.log.Debug("HTTP registered", "service", api.Service, "namespace", 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, vhosts, handler).Serve(listener) n.log.Info("HTTP endpoint opened", "url", fmt.Sprintf("http://%s", endpoint), "cors", strings.Join(cors, ","), "vhosts", strings.Join(vhosts, ",")) // All listeners booted successfully n.httpEndpoint = endpoint n.httpListener = listener n.httpHandler = handler return nil }
過濾白名單的介面,白名單在defaultConfig裡面配置。
api的結構體:
type API struct {
Namespace string // namespace under which the rpc methods of Service are exposed
Version string // api version for DApp's
Service interface{} // receiver instance which holds the methods
Public bool // indication if the methods must be considered safe for public use
}
將api的namespace和service傳入RegisterName()func (s *Server) RegisterName(name string, rcvr interface{}) error {
if s.services == nil {
s.services = make(serviceRegistry)
}
svc := new(service)
svc.typ = reflect.TypeOf(rcvr)
rcvrVal := reflect.ValueOf(rcvr)
if name == "" {
return fmt.Errorf("no service name for type %s", svc.typ.String())
}
if !isExported(reflect.Indirect(rcvrVal).Type().Name()) {
return fmt.Errorf("%s is not exported", reflect.Indirect(rcvrVal).Type().Name())
}
methods, subscriptions := suitableCallbacks(rcvrVal, svc.typ)
// already a previous service register under given sname, merge methods/subscriptions
if regsvc, present := s.services[name]; present {
if len(methods) == 0 && len(subscriptions) == 0 {
return fmt.Errorf("Service %T doesn't have any suitable methods/subscriptions to expose", rcvr)
}
for _, m := range methods {
regsvc.callbacks[formatName(m.method.Name)] = m
}
for _, s := range subscriptions {
regsvc.subscriptions[formatName(s.method.Name)] = s
}
return nil
}
svc.name = name
svc.callbacks, svc.subscriptions = methods, subscriptions
if len(svc.callbacks) == 0 && len(svc.subscriptions) == 0 {
return fmt.Errorf("Service %T doesn't have any suitable methods/subscriptions to expose", rcvr)
}
s.services[svc.name] = svc
return nil
}
用go的反射方法獲取到service的型別和所持有的值。suitableCallbacks方法獲取Service所有的的方法和符合訂閱標準的方法。
將Service的所有方法放入map s.services, service的name作為map key。
3,Http RPC Client的呼叫過程
ethclient.go做了個go-ethereum 客戶端請求的client例項。
ethclient例項建立
func NewEthereumClient(rawurl string) (client *EthereumClient, _ error) {
rawClient, err := ethclient.Dial(rawurl)
return &EthereumClient{rawClient}, err
}
ethclient根據url撥號func Dial(rawurl string) (*Client, error) {
c, err := rpc.Dial(rawurl)
if err != nil {
return nil, err
}
return NewClient(c), nil
}
呼叫rpc的Dial介面func Dial(rawurl string) (*Client, error) {
return DialContext(context.Background(), rawurl)
}
// DialContext creates a new RPC client, just like Dial.
//
// The context is used to cancel or time out the initial connection establishment. It does
// not affect subsequent interactions with the client.
func DialContext(ctx context.Context, rawurl string) (*Client, error) {
u, err := url.Parse(rawurl)
if err != nil {
return nil, err
}
switch u.Scheme {
case "http", "https":
return DialHTTP(rawurl)
case "ws", "wss":
return DialWebsocket(ctx, rawurl, "")
case "":
return DialIPC(ctx, rawurl)
default:
return nil, fmt.Errorf("no known transport for URL scheme %q", u.Scheme)
}
}
根據url的scheme判斷呼叫撥號哪個RPC server
func DialHTTPWithClient(endpoint string, client *http.Client) (*Client, error) {
req, err := http.NewRequest(http.MethodPost, endpoint, nil)
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", contentType)
req.Header.Set("Accept", contentType)
initctx := context.Background()
return newClient(initctx, func(context.Context) (net.Conn, error) {
return &httpConn{client: client, req: req, closed: make(chan struct{})}, nil
})
}
如果是http的話直接生成httpConn就返回newClient了,其他方式的話會複雜一些。
4,一個具體的介面請求例項
選一個簡單的ethClient.go的介面 HeaderByNumber(),請求引數是區塊的number值func (ec *Client) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
var head *types.Header
err := ec.c.CallContext(ctx, &head, "eth_getBlockByNumber", toBlockNumArg(number), false)
if err == nil && head == nil {
err = ethereum.NotFound
}
return head, err
}
呼叫RPC.Client的CallContext()方法,head引用作為出參,"eth_getBlockByNumber"是請求RPC的方法名,後面的兩個引數是eth_getBlockByNumber介面需要的引數。func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
msg, err := c.newMessage(method, args...)
if err != nil {
return err
}
op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)}
if c.isHTTP {
err = c.sendHTTP(ctx, op, msg)
} else {
err = c.send(ctx, op, msg)
}
if err != nil {
return err
}
// dispatch has accepted the request and will close the channel it when it quits.
switch resp, err := op.wait(ctx); {
case err != nil:
return err
case resp.Error != nil:
return resp.Error
case len(resp.Result) == 0:
return ErrNoResult
default:
return json.Unmarshal(resp.Result, &result)
}
}
newMessage()方法拼接了請求data, 如下:{"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["0x111",true],"id":83}
如果是httpRPC請求,進入方法sendHTTP()func (c *Client) sendHTTP(ctx context.Context, op *requestOp, msg interface{}) error {
hc := c.writeConn.(*httpConn)
respBody, err := hc.doRequest(ctx, msg)
if err != nil {
return err
}
defer respBody.Close()
var respmsg jsonrpcMessage
if err := json.NewDecoder(respBody).Decode(&respmsg); err != nil {
return err
}
op.resp <- &respmsg
return nil
}
c.writeConn 就是DialHTTPWithClient 裡面的 &httpConn{client: client, req: req, closed: make(chan struct{})func (hc *httpConn) doRequest(ctx context.Context, msg interface{}) (io.ReadCloser, error) {
body, err := json.Marshal(msg)
if err != nil {
return nil, err
}
req := hc.req.WithContext(ctx)
req.Body = ioutil.NopCloser(bytes.NewReader(body))
req.ContentLength = int64(len(body))
resp, err := hc.client.Do(req)
if err != nil {
return nil, err
}
return resp.Body, nil
}
拼接request請求,呼叫http的do()請求,獲取到請求的返回值respjson.NewDecoder()對請求返回值進行json序列化,然後send進管道op.resp
回到 CallContext()裡面的op.wait(ctx)方法:
func (op *requestOp) wait(ctx context.Context) (*jsonrpcMessage, error) {
select {
case <-ctx.Done():
return nil, ctx.Err()
case resp := <-op.resp:
return resp, op.err
}
}
resp recieve到op.resp管道的資料,然後對resp資料進行json序列化,返回結果。大致如下:{"jsonrpc":"2.0","id":83,"result":{"difficulty":"0x48a01d5ed","extraData":"0x476574682f76312e302e302f6c696e75782f676f312e342e32","gasLimit":"0x1388","gasUsed":"0x0","hash":"0x98dfac36f6e27ef7c538b32e97af049bff79179f20a077dd368be2e76766086d",,"miner":"0x28921e4e2c9d84f4c0f0c0ceb991f45751a0fe93","mixHash":"0x1eec103d7e9b11b78cd899753207105f99327c3611bcb1d99c66c55fb692e33c","nonce":"0x96704278e606f939","number":"0x111","parentHash":"0x8e9a27332450fcdc845e6e20c4a6bd2d2ffc13cf096d4fdd4090ea16b47add19","receiptsRoot":"0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421","sha3Uncles":"0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347","size":"0x21b","stateRoot":"0xa2320aa9611b97a96c6490b5cfd468e89468235c8e0de9a3649c157012577964","timestamp":"0x55ba453f","totalDifficulty":"0x48c8eb4c314","transactions":[],"transactionsRoot":"0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421","uncles":[]}}