以太坊源码深入分析（3）-- 以太坊RPC通信实例和原理代码分析（上）

时间：2021-02-02 17:05:45
上一节提到，以太坊在node start的时候启动了RPC服务，以太坊通过Rpc服务来实现以太坊相关接口的远程调用。这节我们用个实例来看看以太坊 RPC是如何工作的，以及以太坊RPC的源码的实现

一，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_blockNumber 
curl -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（）

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
}
startRPC方法  收集Node里面所有service的 APIs。然后分别启动了
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()请求，获取到请求的返回值resp
json.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":[]}}

标签：

相关文章