feat:rpc part2

Signed-off-by: Chen Kai <281165273grape@gmail.com>
This commit is contained in:
Chen Kai 2023-12-10 21:35:22 +08:00
parent 088af7da90
commit 555b58c26c
3 changed files with 155 additions and 15 deletions

View file

@ -3,7 +3,9 @@ package discover
import ( import (
"errors" "errors"
"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/holiman/uint256"
) )
type DiscV5API struct { type DiscV5API struct {
@ -25,6 +27,17 @@ type RoutingTableInfo struct {
LocalNodeId string `json:"localNodeId"` LocalNodeId string `json:"localNodeId"`
} }
type DiscV5PongResp struct {
EnrSeq uint64 `json:"enrSeq"`
RecipientIP string `json:"recipientIP"`
RecipientPort uint16 `json:"recipientPort"`
}
type PortalPongResp struct {
EnrSeq uint64 `json:"enrSeq"`
DataRadius string `json:"dataRadius"`
}
func (d *DiscV5API) NodeInfo() *NodeInfo { func (d *DiscV5API) NodeInfo() *NodeInfo {
n := d.DiscV5.LocalNode().Node() n := d.DiscV5.LocalNode().Node()
@ -73,6 +86,54 @@ func (d *DiscV5API) GetEnr(nodeId string) (bool, error) {
return true, nil return true, nil
} }
func (d *DiscV5API) DeleteEnr(nodeId string) (bool, error) {
id, err := enode.ParseID(nodeId)
if err != nil {
return false, err
}
n := d.DiscV5.tab.getNode(id)
if n == nil {
return false, errors.New("record not in local routing table")
}
d.DiscV5.tab.delete(wrapNode(n))
return true, nil
}
func (d *DiscV5API) LookupEnr(nodeId string) (string, error) {
id, err := enode.ParseID(nodeId)
if err != nil {
return "", err
}
enr := d.DiscV5.ResolveNodeId(id)
if enr == nil {
return "", errors.New("record not found in DHT lookup")
}
return enr.String(), nil
}
func (d *DiscV5API) Ping(enr string) (*DiscV5PongResp, error) {
n, err := enode.Parse(enode.ValidSchemes, enr)
if err != nil {
return nil, err
}
pong, err := d.DiscV5.pingInner(n)
if err != nil {
return nil, err
}
return &DiscV5PongResp{
EnrSeq: pong.ENRSeq,
RecipientIP: pong.ToIP.String(),
RecipientPort: pong.ToPort,
}, nil
}
type PortalAPI struct { type PortalAPI struct {
*DiscV5API *DiscV5API
portalProtocol *PortalProtocol portalProtocol *PortalProtocol
@ -152,19 +213,19 @@ func (p *PortalAPI) GetEnr(nodeId string) (string, error) {
return n.String(), nil return n.String(), nil
} }
func (p *PortalAPI) DeleteEnr(nodeId string) bool { func (p *PortalAPI) DeleteEnr(nodeId string) (bool, error) {
id, err := enode.ParseID(nodeId) id, err := enode.ParseID(nodeId)
if err != nil { if err != nil {
return false return false, err
} }
n := p.portalProtocol.table.getNode(id) n := p.portalProtocol.table.getNode(id)
if n == nil { if n == nil {
return false return false, errors.New("record not in local routing table")
} }
p.portalProtocol.table.delete(wrapNode(n)) p.portalProtocol.table.delete(wrapNode(n))
return true return true, nil
} }
func (p *PortalAPI) LookupEnr(nodeId string) (string, error) { func (p *PortalAPI) LookupEnr(nodeId string) (string, error) {
@ -181,3 +242,32 @@ func (p *PortalAPI) LookupEnr(nodeId string) (string, error) {
return enr.String(), nil return enr.String(), nil
} }
func (p *PortalAPI) Ping(enr string) (*PortalPongResp, error) {
n, err := enode.Parse(enode.ValidSchemes, enr)
if err != nil {
return nil, err
}
pong, err := p.portalProtocol.pingInner(n)
if err != nil {
return nil, err
}
customPayload := &portalwire.PingPongCustomData{}
err = customPayload.UnmarshalSSZ(pong.CustomPayload)
if err != nil {
return nil, err
}
nodeRadius := new(uint256.Int)
err = nodeRadius.UnmarshalSSZ(customPayload.Radius)
if err != nil {
return nil, err
}
return &PortalPongResp{
EnrSeq: pong.EnrSeq,
DataRadius: nodeRadius.Hex(),
}, nil
}

View file

@ -273,10 +273,19 @@ func (p *PortalProtocol) setupDiscV5AndTable() error {
} }
func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) { func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
pong, err := p.pingInner(node)
if err != nil {
return 0, err
}
return pong.EnrSeq, nil
}
func (p *PortalProtocol) pingInner(node *enode.Node) (*portalwire.Pong, error) {
enrSeq := p.Self().Seq() enrSeq := p.Self().Seq()
radiusBytes, err := p.nodeRadius.MarshalSSZ() radiusBytes, err := p.nodeRadius.MarshalSSZ()
if err != nil { if err != nil {
return 0, err return nil, err
} }
customPayload := &portalwire.PingPongCustomData{ customPayload := &portalwire.PingPongCustomData{
Radius: radiusBytes, Radius: radiusBytes,
@ -284,7 +293,7 @@ func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
customPayloadBytes, err := customPayload.MarshalSSZ() customPayloadBytes, err := customPayload.MarshalSSZ()
if err != nil { if err != nil {
return 0, err return nil, err
} }
pingRequest := &portalwire.Ping{ pingRequest := &portalwire.Ping{
@ -295,7 +304,7 @@ func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest) p.log.Trace("Sending ping request", "protocol", p.protocolId, "source", p.Self().ID(), "target", node.ID(), "ping", pingRequest)
pingRequestBytes, err := pingRequest.MarshalSSZ() pingRequestBytes, err := pingRequest.MarshalSSZ()
if err != nil { if err != nil {
return 0, err return nil, err
} }
talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1) talkRequestBytes := make([]byte, 0, len(pingRequestBytes)+1)
@ -306,8 +315,9 @@ func (p *PortalProtocol) ping(node *enode.Node) (uint64, error) {
if err != nil { if err != nil {
p.replaceNode(node) p.replaceNode(node)
return 0, err return nil, err
} }
return p.processPong(node, talkResp) return p.processPong(node, talkResp)
} }
@ -630,15 +640,15 @@ func (p *PortalProtocol) filterNodes(target *enode.Node, enrs [][]byte, distance
return nodes return nodes
} }
func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, error) { func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (*portalwire.Pong, error) {
if resp[0] != portalwire.PONG { if resp[0] != portalwire.PONG {
return 0, fmt.Errorf("invalid pong response") return nil, fmt.Errorf("invalid pong response")
} }
pong := &portalwire.Pong{} pong := &portalwire.Pong{}
err := pong.UnmarshalSSZ(resp[1:]) err := pong.UnmarshalSSZ(resp[1:])
if err != nil { if err != nil {
p.replaceNode(target) p.replaceNode(target)
return 0, err return nil, err
} }
p.log.Trace("Received pong response", "id", target.ID(), "pong", pong) p.log.Trace("Received pong response", "id", target.ID(), "pong", pong)
@ -647,14 +657,14 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (uint64, e
err = customPayload.UnmarshalSSZ(pong.CustomPayload) err = customPayload.UnmarshalSSZ(pong.CustomPayload)
if err != nil { if err != nil {
p.replaceNode(target) p.replaceNode(target)
return 0, err return nil, err
} }
p.log.Trace("Received pong response", "id", target.ID(), "pong", pong, "customPayload", customPayload) p.log.Trace("Received pong response", "id", target.ID(), "pong", pong, "customPayload", customPayload)
p.radiusCache.Set([]byte(target.ID().String()), customPayload.Radius) p.radiusCache.Set([]byte(target.ID().String()), customPayload.Radius)
p.table.addVerifiedNode(wrapNode(target)) p.table.addVerifiedNode(wrapNode(target))
return pong.EnrSeq, nil return pong, nil
} }
func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte { func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {

View file

@ -224,6 +224,36 @@ func (t *UDPv5) Resolve(n *enode.Node) *enode.Node {
return n return n
} }
// ResolveNodeId searches for a specific Node with the given ID.
// It returns nil if the nodeId could not be resolved.
func (t *UDPv5) ResolveNodeId(id enode.ID) *enode.Node {
if id == t.Self().ID() {
return t.Self()
}
n := t.tab.getNode(id)
if n != nil {
// Try asking directly. This works if the Node is still responding on the endpoint we have.
if resp, err := t.RequestENR(n); err == nil {
return resp
}
}
// Otherwise do a network lookup.
result := t.Lookup(n.ID())
for _, rn := range result {
if rn.ID() == id {
if n != nil && rn.Seq() <= n.Seq() {
return n
} else {
return rn
}
}
}
return n
}
// AllNodes returns all the nodes stored in the local table. // AllNodes returns all the nodes stored in the local table.
func (t *UDPv5) AllNodes() []*enode.Node { func (t *UDPv5) AllNodes() []*enode.Node {
t.tab.mutex.Lock() t.tab.mutex.Lock()
@ -357,15 +387,25 @@ func lookupDistances(target, dest enode.ID) (dists []uint) {
// ping calls PING on a node and waits for a PONG response. // ping calls PING on a node and waits for a PONG response.
func (t *UDPv5) ping(n *enode.Node) (uint64, error) { func (t *UDPv5) ping(n *enode.Node) (uint64, error) {
pong, err := t.pingInner(n)
if err != nil {
return 0, err
}
return pong.ENRSeq, nil
}
// pingInner calls PING on a node and waits for a PONG response.
func (t *UDPv5) pingInner(n *enode.Node) (*v5wire.Pong, error) {
req := &v5wire.Ping{ENRSeq: t.localNode.Node().Seq()} req := &v5wire.Ping{ENRSeq: t.localNode.Node().Seq()}
resp := t.callToNode(n, v5wire.PongMsg, req) resp := t.callToNode(n, v5wire.PongMsg, req)
defer t.callDone(resp) defer t.callDone(resp)
select { select {
case pong := <-resp.ch: case pong := <-resp.ch:
return pong.(*v5wire.Pong).ENRSeq, nil return pong.(*v5wire.Pong), nil
case err := <-resp.err: case err := <-resp.err:
return 0, err return nil, err
} }
} }