/
handler.go
106 lines (97 loc) · 3.37 KB
/
handler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
package p2p
import (
"context"
"reflect"
"runtime/debug"
"strings"
libp2pcore "github.com/libp2p/go-libp2p/core"
corenet "github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/protocol"
ssz "github.com/prysmaticlabs/fastssz"
"github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p"
p2ptypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/types"
)
type rpcHandler func(context.Context, interface{}, libp2pcore.Stream) error
// registerRPC for a given topic with an expected protobuf message type.
func (c *client) registerRPCHandler(baseTopic string, handle rpcHandler) {
topic := baseTopic + c.Encoding().ProtocolSuffix()
c.host.SetStreamHandler(protocol.ID(topic), func(stream corenet.Stream) {
defer func() {
if r := recover(); r != nil {
log.WithField("error", r).Error("Panic occurred")
log.Errorf("%s", debug.Stack())
}
}()
// Resetting after closing is a no-op so defer a reset in case something goes wrong.
// It's up to the handler to Close the stream (send an EOF) if
// it successfully writes a response. We don't blindly call
// Close here because we may have only written a partial
// response.
defer func() {
_err := stream.Reset()
_ = _err
}()
log.WithField("peer", stream.Conn().RemotePeer().Pretty()).WithField("topic", string(stream.Protocol()))
base, ok := p2p.RPCTopicMappings[baseTopic]
if !ok {
log.Errorf("Could not retrieve base message for topic %s", baseTopic)
return
}
t := reflect.TypeOf(base)
// Copy Base
base = reflect.New(t)
// since metadata requests do not have any data in the payload, we
// do not decode anything.
if baseTopic == p2p.RPCMetaDataTopicV1 || baseTopic == p2p.RPCMetaDataTopicV2 {
if err := handle(context.Background(), base, stream); err != nil {
if err != p2ptypes.ErrWrongForkDigestVersion {
log.WithError(err).Debug("Could not handle p2p RPC")
}
}
return
}
// Given we have an input argument that can be pointer or the actual object, this gives us
// a way to check for its reflect.Kind and based on the result, we can decode
// accordingly.
if t.Kind() == reflect.Ptr {
msg, ok := reflect.New(t.Elem()).Interface().(ssz.Unmarshaler)
if !ok {
log.Errorf("message of %T does not support marshaller interface", msg)
return
}
if err := c.Encoding().DecodeWithMaxLength(stream, msg); err != nil {
// Trace logs for goodbye errors
logStreamErrors(err, topic)
return
}
if err := handle(context.Background(), msg, stream); err != nil {
if err != p2ptypes.ErrWrongForkDigestVersion {
log.WithError(err).Debug("Could not handle p2p RPC")
}
}
} else {
nTyp := reflect.New(t)
msg, ok := nTyp.Interface().(ssz.Unmarshaler)
if !ok {
log.Errorf("message of %T does not support marshaller interface", msg)
return
}
if err := c.Encoding().DecodeWithMaxLength(stream, msg); err != nil {
logStreamErrors(err, topic)
return
}
if err := handle(context.Background(), nTyp.Elem().Interface(), stream); err != nil {
if err != p2ptypes.ErrWrongForkDigestVersion {
log.WithError(err).Debug("Could not handle p2p RPC")
}
}
}
})
}
func logStreamErrors(err error, topic string) {
if strings.Contains(topic, p2p.RPCGoodByeTopicV1) {
log.WithError(err).Trace("Could not decode goodbye stream message")
return
}
log.WithError(err).Debug("Could not decode stream message")
}