forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

les: add talk request handler

Browse source
This commit is contained in:
Zsolt Felfoldi 2019-11-15 21:08:18 +01:00
parent 2d11d7455b
commit 2b96b4f13d
3 changed files with 21 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
les/server.go
View file

@ -42,6 +42,7 @@ type LesServer struct {
handler *serverHandler
lesTopics []discv5.Topic
privateKey *ecdsa.PrivateKey
srvr *p2p.Server
// Flow control and capacity management
fcManager *flowcontrol.ClientManager
@ -165,6 +166,7 @@ func (s *LesServer) Protocols() []p2p.Protocol {
// Start starts the LES server
func (s *LesServer) Start(srvr *p2p.Server) {
s.srvr = srvr
s.privateKey = srvr.PrivateKey
s.handler.start()
@ -172,6 +174,7 @@ func (s *LesServer) Start(srvr *p2p.Server) {
go s.capacityManagement()
if srvr.DiscV5 != nil {
srvr.DiscV5.RegisterTalkHandler("les", s.handler.talkRequestHandler)
for _, topic := range s.lesTopics {
topic := topic
go func() {
@ -189,6 +192,10 @@ func (s *LesServer) Start(srvr *p2p.Server) {
func (s *LesServer) Stop() {
close(s.closeCh)
if s.srvr.DiscV5 != nil {
s.srvr.DiscV5.RemoveTalkHandler("les")
}
// Disconnect existing sessions.
// This also closes the gate for any new registrations on the peer set.
// sessions which are already established but not added to pm.peers yet

6
les/server_handler.go
View file

@ -20,6 +20,7 @@ import (
"encoding/binary"
"encoding/json"
"errors"
"net"
"sync"
"sync/atomic"
"time"
@ -35,6 +36,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/trie"
)
@ -953,3 +955,7 @@ func (h *serverHandler) broadcastHeaders() {
}
}
}
func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload rlp.RawValue) (rlp.RawValue, bool) {
return payload, true
}

10
p2p/discv5/net.go
View file

@ -87,7 +87,7 @@ type Network struct {
}
type (
TalkRequestHandler func(enode.ID, rlp.RawValue) (rlp.RawValue, bool)
TalkRequestHandler func(enode.ID, *net.UDPAddr, rlp.RawValue) (rlp.RawValue, bool)
TalkResponseHandler func(rlp.RawValue) bool
)
@ -1220,7 +1220,7 @@ func (net *Network) handleQueryEvent(n *Node, ev nodeEvent, pkt *ingressPacket)
subFn := net.talkRequestSubs[string(p.TalkID)]
net.talkRequestSubLock.RUnlock()
if subFn != nil {
resp, ok := subFn(enode.ID(n.sha), p.Payload)
resp, ok := subFn(enode.ID(n.sha), n.addr(), p.Payload)
if ok {
net.conn.send(n, talkResponsePacket, talkResponse{ReplyTok: pkt.hash, Payload: resp})
} else {
@ -1329,6 +1329,12 @@ func (net *Network) RegisterTalkHandler(talkID string, handler TalkRequestHandle
net.talkRequestSubLock.Unlock()
}
func (net *Network) RemoveTalkHandler(talkID string) {
net.talkRequestSubLock.Lock()
delete(net.talkRequestSubs, talkID)
net.talkRequestSubLock.Unlock()
}
func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload rlp.RawValue, handler TalkResponseHandler) func() bool {
var nodeID NodeID
copy(nodeID[:], crypto.FromECDSAPub(to.Pubkey())[1:])