package p2p import ( "context" "encoding/base64" "time" pubsub "github.com/libp2p/go-libp2p-pubsub" pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb" "github.com/prysmaticlabs/prysm/shared/featureconfig" "github.com/prysmaticlabs/prysm/shared/hashutil" ) // JoinTopic will join PubSub topic, if not already joined. func (s *Service) JoinTopic(topic string, opts ...pubsub.TopicOpt) (*pubsub.Topic, error) { s.joinedTopicsLock.Lock() defer s.joinedTopicsLock.Unlock() if _, ok := s.joinedTopics[topic]; !ok { topicHandle, err := s.pubsub.Join(topic, opts...) if err != nil { return nil, err } s.joinedTopics[topic] = topicHandle } return s.joinedTopics[topic], nil } // LeaveTopic closes topic and removes corresponding handler from list of joined topics. // This method will return error if there are outstanding event handlers or subscriptions. func (s *Service) LeaveTopic(topic string) error { s.joinedTopicsLock.Lock() defer s.joinedTopicsLock.Unlock() if t, ok := s.joinedTopics[topic]; ok { if err := t.Close(); err != nil { return err } delete(s.joinedTopics, topic) } return nil } // PublishToTopic joins (if necessary) and publishes a message to a PubSub topic. func (s *Service) PublishToTopic(ctx context.Context, topic string, data []byte, opts ...pubsub.PubOpt) error { topicHandle, err := s.JoinTopic(topic) if err != nil { return err } // If feature flag isn't enabled, don't wait for peers to be present. if !featureconfig.Get().EnableAttBroadcastDiscoveryAttempts { return topicHandle.Publish(ctx, data, opts...) } // Wait for at least 1 peer to be available to receive the published message. for { if len(topicHandle.ListPeers()) > 0 { return topicHandle.Publish(ctx, data, opts...) } select { case <-ctx.Done(): return ctx.Err() default: time.Sleep(100 * time.Millisecond) } } } // SubscribeToTopic joins (if necessary) and subscribes to PubSub topic. func (s *Service) SubscribeToTopic(topic string, opts ...pubsub.SubOpt) (*pubsub.Subscription, error) { topicHandle, err := s.JoinTopic(topic) if err != nil { return nil, err } return topicHandle.Subscribe(opts...) } // Content addressable ID function. // // ETH2 spec defines the message ID as: // message-id: base64(SHA256(message.data)) // where base64 is the URL-safe base64 alphabet with // padding characters omitted. func msgIDFunction(pmsg *pubsub_pb.Message) string { h := hashutil.Hash(pmsg.Data) return base64.RawURLEncoding.EncodeToString(h[:]) } func setPubSubParameters() { pubsub.GossipSubDlo = 5 pubsub.GossipSubHeartbeatInterval = 700 * time.Millisecond pubsub.GossipSubHistoryLength = 6 pubsub.GossipSubHistoryGossip = 3 }