mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-25 12:57:18 +00:00
5569a68452
* Value assigned to a variable is never read before being overwritten * The result of append is not used anywhere * Suspicious assignment of range-loop vars detected * Unused method receiver detected * Revert "Auxiliary commit to revert individual files from 54edcb445484a2e5d79612e19af8e949b8861253" This reverts commit bbd1e1beabf7b0c5cfc4f514dcc820062ad6c063. * Method modifies receiver * Fix test * Duplicate imports detected * Incorrectly formatted error string * Types of function parameters can be combined * One more "Unused method receiver detected" * Unused parameter detected in function
72 lines
2.2 KiB
Go
72 lines
2.2 KiB
Go
package sync
|
|
|
|
import (
|
|
"reflect"
|
|
"strings"
|
|
|
|
ssz "github.com/ferranbt/fastssz"
|
|
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
|
"github.com/pkg/errors"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
|
|
ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
var errNilPubsubMessage = errors.New("nil pubsub message")
|
|
var errInvalidTopic = errors.New("invalid topic format")
|
|
|
|
func (s *Service) decodePubsubMessage(msg *pubsub.Message) (ssz.Unmarshaler, error) {
|
|
if msg == nil || msg.Topic == nil || *msg.Topic == "" {
|
|
return nil, errNilPubsubMessage
|
|
}
|
|
topic := *msg.Topic
|
|
fDigest, err := p2p.ExtractGossipDigest(topic)
|
|
if err != nil {
|
|
return nil, errors.Wrapf(err, "extraction failed for topic: %s", topic)
|
|
}
|
|
topic = strings.TrimSuffix(topic, s.cfg.p2p.Encoding().ProtocolSuffix())
|
|
topic, err = s.replaceForkDigest(topic)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Specially handle subnet messages.
|
|
switch {
|
|
case strings.Contains(topic, p2p.GossipAttestationMessage):
|
|
topic = p2p.GossipTypeMapping[reflect.TypeOf(ðpb.Attestation{})]
|
|
// Given that both sync message related subnets have the same message name, we have to
|
|
// differentiate them below.
|
|
case strings.Contains(topic, p2p.GossipSyncCommitteeMessage) && !strings.Contains(topic, p2p.SyncContributionAndProofSubnetTopicFormat):
|
|
topic = p2p.GossipTypeMapping[reflect.TypeOf(ðpb.SyncCommitteeMessage{})]
|
|
}
|
|
|
|
base := p2p.GossipTopicMappings(topic, 0)
|
|
if base == nil {
|
|
return nil, p2p.ErrMessageNotMapped
|
|
}
|
|
m, ok := proto.Clone(base).(ssz.Unmarshaler)
|
|
if !ok {
|
|
return nil, errors.Errorf("message of %T does not support marshaller interface", base)
|
|
}
|
|
// Handle different message types across forks.
|
|
if topic == p2p.BlockSubnetTopicFormat {
|
|
m, err = extractBlockDataType(fDigest[:], s.cfg.chain)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
if err := s.cfg.p2p.Encoding().DecodeGossip(msg.Data, m); err != nil {
|
|
return nil, err
|
|
}
|
|
return m, nil
|
|
}
|
|
|
|
// Replaces our fork digest with the formatter.
|
|
func (_ *Service) replaceForkDigest(topic string) (string, error) {
|
|
subStrings := strings.Split(topic, "/")
|
|
if len(subStrings) != 4 {
|
|
return "", errInvalidTopic
|
|
}
|
|
subStrings[2] = "%x"
|
|
return strings.Join(subStrings, "/"), nil
|
|
}
|