package p2p import ( "bytes" "context" "reflect" "github.com/gogo/protobuf/proto" "github.com/pkg/errors" "github.com/prysmaticlabs/prysm/shared/hashutil" "github.com/prysmaticlabs/prysm/shared/traceutil" "go.opencensus.io/trace" ) // ErrMessageNotMapped occurs on a Broadcast attempt when a message has not been defined in the // GossipTypeMapping. var ErrMessageNotMapped = errors.New("message type is not mapped to a PubSub topic") // Broadcast a message to the p2p network. func (s *Service) Broadcast(ctx context.Context, msg proto.Message) error { ctx, span := trace.StartSpan(ctx, "p2p.Broadcast") defer span.End() topic, ok := GossipTypeMapping[reflect.TypeOf(msg)] if !ok { traceutil.AnnotateError(span, ErrMessageNotMapped) return ErrMessageNotMapped } span.AddAttributes(trace.StringAttribute("topic", topic)) buf := new(bytes.Buffer) if _, err := s.Encoding().Encode(buf, msg); err != nil { err := errors.Wrap(err, "could not encode message") traceutil.AnnotateError(span, err) return err } if span.IsRecordingEvents() { id := hashutil.FastSum64(buf.Bytes()) messageLen := int64(buf.Len()) span.AddMessageSendEvent(int64(id), messageLen /*uncompressed*/, messageLen /*compressed*/) } if err := s.pubsub.Publish(topic+s.Encoding().ProtocolSuffix(), buf.Bytes()); err != nil { err := errors.Wrap(err, "could not publish message") traceutil.AnnotateError(span, err) return err } return nil }