pubsub

Aug 16, 2024

Changes in this version

+ const GossipSubFeatureIdontwant

+ const GossipSubID_v12

+ var ErrQueueCancelled = errors.New("rpc queue operation cancelled")

+ var ErrQueueClosed = errors.New("rpc queue closed")

+ var ErrQueueFull = errors.New("rpc queue full")

+ var ErrQueuePushOnClosed = errors.New("push on closed rpc queue")

+ var GossipSubIDontWantMessageTTL = 3

+ var GossipSubIDontWantMessageThreshold = 1024

+ var GossipSubMaxIDontWantMessages = 1000

type FloodSubRouter

+ func (fs *FloodSubRouter) PreValidation([]*Message)

type GossipSubParams

+ IDontWantMessageTTL int

+ IDontWantMessageThreshold int

+ MaxIDontWantMessages int

type GossipSubRouter

+ func (gs *GossipSubRouter) PreValidation(msgs []*Message)

type PubSubRouter

+ PreValidation func([]*Message)

type RandomSubRouter

+ func (rs *RandomSubRouter) PreValidation([]*Message)

Mar 1, 2023

Changes in this version

+ type BasicSeqnoValidator struct

type Option

+ func WithDefaultValidator(val interface{}, opts ...ValidatorOpt) Option

+ type PeerMetadataStore interface

+ Get func(context.Context, peer.ID) ([]byte, error)

+ Put func(context.Context, peer.ID, []byte) error

type ValidatorEx

+ func NewBasicSeqnoValidator(meta PeerMetadataStore) ValidatorEx

Jan 24, 2023

Changes in this version

+ var TimeCacheStrategy = timecache.Strategy_FirstSeen

type Option

+ func WithSeenMessagesStrategy(strategy timecache.Strategy) Option

Dec 1, 2022

Changes in this version

type GossipSubRouter

+ func DefaultGossipSubRouter(h host.Host) *GossipSubRouter

+ func (gs *GossipSubRouter) WithDefaultTagTracer() Option

type Option

+ func WithAppSpecificRpcInspector(inspector func(peer.ID, *RPC) error) Option

type PubSub

+ func NewGossipSubWithRouter(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option) (*PubSub, error)

Sep 8, 2022

Changes in this version

type PeerScoreParams

+ SkipAtomicValidation bool

type PeerScoreThresholds

+ SkipAtomicValidation bool

type TopicScoreParams

+ SkipAtomicValidation bool

Aug 26, 2022

Changes in this version

+ var GossipSubUnsubscribeBackoff = 10 * time.Second

type GossipSubParams

+ UnsubscribeBackoff time.Duration

type Message

+ Local bool

type PubOpt

+ func WithLocalPublication(local bool) PubOpt

May 30, 2022

Changes in this version

+ const BackoffCleanupInterval

+ const BackoffMultiplier

+ const MaxBackoffAttempts

+ const MaxBackoffDelay

+ const MaxBackoffJitterCoff

+ const MinBackoffDelay

+ const TimeToLive

+ var ErrEmptyPeerID = errors.New("empty peer ID")

+ var ErrNilSignKey = errors.New("nil sign key")

type Message

+ ID string

type Option

+ func WithSeenMessagesTTL(ttl time.Duration) Option

type PeerScoreParams

+ SeenMsgTTL time.Duration

+ type ProvideKey func() (crypto.PrivKey, peer.ID)

type PubOpt

+ func WithSecretKeyAndPeerId(key crypto.PrivKey, pid peer.ID) PubOpt

type TopicOpt

+ func WithTopicMessageIdFn(msgId MsgIdFunction) TopicOpt

Sep 22, 2021

Changes in this version

+ func DefaultPeerFilter(pid peer.ID, topic string) bool

type Option

+ func WithPeerFilter(filter PeerFilter) Option

+ type PeerFilter func(pid peer.ID, topic string) bool

Jul 30, 2021

Changes in this version

type GossipSubParams

+ SlowHeartbeatWarning float64

type Option

+ func WithProtocolMatchFn(m ProtocolMatchFn) Option

+ type ProtocolMatchFn = func(string) func(string) bool

Jul 13, 2021

Changes in this version

+ const GossipSubFeatureMesh

+ const GossipSubFeaturePX

+ const RejectBlacklistedSource

+ const RejectBlacklstedPeer

+ const RejectInvalidSignature

+ const RejectMissingSignature

+ const RejectSelfOrigin

+ const RejectUnexpectedAuthInfo

+ const RejectUnexpectedSignature

+ const RejectValidationFailed

+ const RejectValidationIgnored

+ const RejectValidationQueueFull

+ const RejectValidationThrottled

+ var GossipSubDefaultProtocols = []protocol.ID

+ func GossipSubDefaultFeatures(feat GossipSubFeature, proto protocol.ID) bool

+ type GossipSubFeature int

+ type GossipSubFeatureTest = func(GossipSubFeature, protocol.ID) bool

+ type GossipSubParams struct

+ ConnectionTimeout time.Duration

+ Connectors int

+ D int

+ Dhi int

+ DirectConnectInitialDelay time.Duration

+ DirectConnectTicks uint64

+ Dlazy int

+ Dlo int

+ Dout int

+ Dscore int

+ FanoutTTL time.Duration

+ GossipFactor float64

+ GossipRetransmission int

+ GraftFloodThreshold time.Duration

+ HeartbeatInitialDelay time.Duration

+ HeartbeatInterval time.Duration

+ HistoryGossip int

+ HistoryLength int

+ IWantFollowupTime time.Duration

+ MaxIHaveLength int

+ MaxIHaveMessages int

+ MaxPendingConnections int

+ OpportunisticGraftPeers int

+ OpportunisticGraftTicks uint64

+ PruneBackoff time.Duration

+ PrunePeers int

+ func DefaultGossipSubParams() GossipSubParams

type Option

+ func WithGossipSubParams(cfg GossipSubParams) Option

+ func WithGossipSubProtocols(protos []protocol.ID, feature GossipSubFeatureTest) Option

+ func WithRawTracer(tracer RawTracer) Option

+ type RawTracer interface

+ AddPeer func(p peer.ID, proto protocol.ID)

+ DeliverMessage func(msg *Message)

+ DropRPC func(rpc *RPC, p peer.ID)

+ DuplicateMessage func(msg *Message)

+ Graft func(p peer.ID, topic string)

+ Join func(topic string)

+ Leave func(topic string)

+ Prune func(p peer.ID, topic string)

+ RecvRPC func(rpc *RPC)

+ RejectMessage func(msg *Message, reason string)

+ RemovePeer func(p peer.ID)

+ SendRPC func(rpc *RPC, p peer.ID)

+ ThrottlePeer func(p peer.ID)

+ UndeliverableMessage func(msg *Message)

+ ValidateMessage func(msg *Message)

type SubOpt

+ func WithBufferSize(size int) SubOpt

+ type ValidationError struct

+ Reason string

+ func (e ValidationError) Error() string

Jan 12, 2021

Changes in this version

type PubSubNotif

+ func (p *PubSubNotif) Initialize()

Oct 8, 2020

Changes in this version

+ var DefaultPeerGaterDuplicateWeight = 0.125

+ var DefaultPeerGaterGlobalDecay = ScoreParameterDecay(2 * time.Minute)

+ var DefaultPeerGaterIgnoreWeight = 1.0

+ var DefaultPeerGaterQuiet = time.Minute

+ var DefaultPeerGaterRejectWeight = 16.0

+ var DefaultPeerGaterRetainStats = 6 * time.Hour

+ var DefaultPeerGaterSourceDecay = ScoreParameterDecay(time.Hour)

+ var DefaultPeerGaterThreshold = 0.33

+ var ErrTooManySubscriptions = errors.New("too many subscriptions")

+ func FilterSubscriptions(subs []*pb.RPC_SubOpts, filter func(string) bool) []*pb.RPC_SubOpts

+ type AcceptStatus int

+ const AcceptAll

+ const AcceptControl

+ const AcceptNone

type Option

+ func WithPeerGater(params *PeerGaterParams) Option

+ func WithSubscriptionFilter(subFilter SubscriptionFilter) Option

+ type PeerGaterParams struct

+ DecayInterval time.Duration

+ DecayToZero float64

+ DuplicateWeight float64

+ GlobalDecay float64

+ IgnoreWeight float64

+ Quiet time.Duration

+ RejectWeight float64

+ RetainStats time.Duration

+ SourceDecay float64

+ Threshold float64

+ TopicDeliveryWeights map[string]float64

+ func DefaultPeerGaterParams() *PeerGaterParams

+ func NewPeerGaterParams(threshold, globalDecay, sourceDecay float64) *PeerGaterParams

+ func (p *PeerGaterParams) WithTopicDeliveryWeights(w map[string]float64) *PeerGaterParams

+ type SubscriptionFilter interface

+ CanSubscribe func(topic string) bool

+ FilterIncomingSubscriptions func(peer.ID, []*pb.RPC_SubOpts) ([]*pb.RPC_SubOpts, error)

+ func NewAllowlistSubscriptionFilter(topics ...string) SubscriptionFilter

+ func NewRegexpSubscriptionFilter(rx *regexp.Regexp) SubscriptionFilter

+ func WrapLimitSubscriptionFilter(filter SubscriptionFilter, limit int) SubscriptionFilter

type Topic

+ func (t *Topic) SetScoreParams(p *TopicScoreParams) error

Aug 25, 2020

Changes in this version

type PeerScoreParams

+ BehaviourPenaltyThreshold float64

type Topic

+ func (t *Topic) String() string

Jul 30, 2020

Changes in this version

+ const LaxNoSign

+ const LaxSign

+ const StrictNoSign

+ const StrictSign

+ var ErrSubscriptionCancelled = errors.New("subscription cancelled")

+ type ExtendedPeerScoreInspectFn = func(map[peer.ID]*PeerScoreSnapshot)

+ type MessageSignaturePolicy uint8

type Option

+ func WithMessageSignaturePolicy(policy MessageSignaturePolicy) Option

+ func WithNoAuthor() Option

+ type PeerScoreSnapshot struct

+ AppSpecificScore float64

+ BehaviourPenalty float64

+ IPColocationFactor float64

+ Score float64

+ Topics map[string]*TopicScoreSnapshot

+ type TopicScoreSnapshot struct

+ FirstMessageDeliveries float64

+ InvalidMessageDeliveries float64

+ MeshMessageDeliveries float64

+ TimeInMesh time.Duration

Jun 3, 2020

Changes in this version

type Option

+ func WithDirectConnectTicks(t uint64) Option

May 20, 2020

Changes in this version

+ const DefaultDecayInterval

+ const DefaultDecayToZero

+ const DefaultMaxMessageSize

+ const GossipSubID_v10

+ const GossipSubID_v11

+ const ValidationAccept

+ const ValidationIgnore

+ const ValidationReject

+ var GossipSubConnTagBumpMessageDelivery = 1

+ var GossipSubConnTagDecayAmount = 1

+ var GossipSubConnTagDecayInterval = 10 * time.Minute

+ var GossipSubConnTagMessageDeliveryCap = 15

+ var GossipSubConnTagValueDirectPeer = 1000

+ var GossipSubConnTagValueMeshPeer = 20

+ var GossipSubConnectionTimeout = 30 * time.Second

+ var GossipSubConnectors = 8

+ var GossipSubDirectConnectInitialDelay = time.Second

+ var GossipSubDirectConnectTicks uint64 = 300

+ var GossipSubDlazy = 6

+ var GossipSubDout = 2

+ var GossipSubDscore = 4

+ var GossipSubGossipFactor = 0.25

+ var GossipSubGossipRetransmission = 3

+ var GossipSubGraftFloodThreshold = 10 * time.Second

+ var GossipSubIWantFollowupTime = 3 * time.Second

+ var GossipSubMaxIHaveLength = 5000

+ var GossipSubMaxIHaveMessages = 10

+ var GossipSubMaxPendingConnections = 128

+ var GossipSubOpportunisticGraftPeers = 2

+ var GossipSubOpportunisticGraftTicks uint64 = 60

+ var GossipSubPruneBackoff = time.Minute

+ var GossipSubPrunePeers = 16

+ func ScoreParameterDecay(decay time.Duration) float64

+ func ScoreParameterDecayWithBase(decay time.Duration, base time.Duration, decayToZero float64) float64

type FloodSubRouter

+ func (fs *FloodSubRouter) AcceptFrom(peer.ID) bool

type GossipSubRouter

+ D int

+ Dhi int

+ Dlazy int

+ Dlo int

+ Dout int

+ Dscore int

+ func (gs *GossipSubRouter) AcceptFrom(p peer.ID) bool

type MessageCache

+ func (mc *MessageCache) GetForPeer(mid string, p peer.ID) (*pb.Message, int, bool)

type Option

+ func WithDirectPeers(pis []peer.AddrInfo) Option

+ func WithFloodPublish(floodPublish bool) Option

+ func WithMaxMessageSize(maxMessageSize int) Option

+ func WithPeerExchange(doPX bool) Option

+ func WithPeerScore(params *PeerScoreParams, thresholds *PeerScoreThresholds) Option

+ func WithPeerScoreInspect(inspect PeerScoreInspectFn, period time.Duration) Option

+ type PeerScoreInspectFn func(map[peer.ID]float64)

+ type PeerScoreParams struct

+ AppSpecificScore func(p peer.ID) float64

+ AppSpecificWeight float64

+ BehaviourPenaltyDecay float64

+ BehaviourPenaltyWeight float64

+ DecayInterval time.Duration

+ DecayToZero float64

+ IPColocationFactorThreshold int

+ IPColocationFactorWeight float64

+ IPColocationFactorWhitelist map[string]struct{}

+ RetainScore time.Duration

+ TopicScoreCap float64

+ Topics map[string]*TopicScoreParams

+ type PeerScoreThresholds struct

+ AcceptPXThreshold float64

+ GossipThreshold float64

+ GraylistThreshold float64

+ OpportunisticGraftThreshold float64

+ PublishThreshold float64

type PubSubRouter

+ AcceptFrom func(peer.ID) bool

type RandomSubRouter

+ func (rs *RandomSubRouter) AcceptFrom(peer.ID) bool

+ type RelayCancelFunc func()

type Topic

+ func (t *Topic) Relay() (RelayCancelFunc, error)

+ type TopicScoreParams struct

+ FirstMessageDeliveriesCap float64

+ FirstMessageDeliveriesDecay float64

+ FirstMessageDeliveriesWeight float64

+ InvalidMessageDeliveriesDecay float64

+ InvalidMessageDeliveriesWeight float64

+ MeshFailurePenaltyDecay float64

+ MeshFailurePenaltyWeight float64

+ MeshMessageDeliveriesActivation time.Duration

+ MeshMessageDeliveriesCap float64

+ MeshMessageDeliveriesDecay float64

+ MeshMessageDeliveriesThreshold float64

+ MeshMessageDeliveriesWeight float64

+ MeshMessageDeliveriesWindow time.Duration

+ TimeInMeshCap float64

+ TimeInMeshQuantum time.Duration

+ TimeInMeshWeight float64

+ TopicWeight float64

+ type ValidationResult int

+ type ValidatorEx func(context.Context, peer.ID, *Message) ValidationResult

Feb 11, 2020

Changes in this version

type Blacklist

+ func NewTimeCachedBlacklist(expiry time.Duration) (Blacklist, error)

type Message

+ ValidatorData interface{}

type Option

+ func WithValidateQueueSize(n int) Option

+ type TimeCachedBlacklist struct

+ func (b *TimeCachedBlacklist) Add(p peer.ID)

+ func (b *TimeCachedBlacklist) Contains(p peer.ID) bool

Dec 16, 2019

Changes in this version

+ func DefaultMsgIdFn(pmsg *pb.Message) string

type MessageCache

+ func (mc *MessageCache) SetMsgIdFn(msgID MsgIdFunction)

+ type MsgIdFunction func(pmsg *pb.Message) string

type Option

+ func WithMessageIdFn(fn MsgIdFunction) Option

Nov 19, 2019

Changes in this version

+ const RemoteTracerProtoID

+ var MinTraceBatchSize = 16

+ var TraceBufferSize = 1 << 16

+ type EventTracer interface

+ Trace func(evt *pb.TraceEvent)

+ type JSONTracer struct

+ func NewJSONTracer(file string) (*JSONTracer, error)

+ func OpenJSONTracer(file string, flags int, perm os.FileMode) (*JSONTracer, error)

+ func (t *JSONTracer) Close()

+ func (t *JSONTracer) Trace(evt *pb.TraceEvent)

type Option

+ func WithEventTracer(tracer EventTracer) Option

+ type PBTracer struct

+ func NewPBTracer(file string) (*PBTracer, error)

+ func OpenPBTracer(file string, flags int, perm os.FileMode) (*PBTracer, error)

+ func (t *PBTracer) Close()

+ func (t *PBTracer) Trace(evt *pb.TraceEvent)

+ type RemoteTracer struct

+ func NewRemoteTracer(ctx context.Context, host host.Host, pi peer.AddrInfo) (*RemoteTracer, error)

+ func (t *RemoteTracer) Close()

+ func (t *RemoteTracer) Trace(evt *pb.TraceEvent)

Nov 15, 2019

Changes in this version

type Message

+ VaidatorData interface{}

type Option

+ func WithPeerOutboundQueueSize(size int) Option

Nov 4, 2019

Changes in this version

+ var ErrTopicClosed = errors.New("this Topic is closed, try opening a new one")

Oct 31, 2019

Changes in this version

+ const FloodSubTopicSearchSize

+ var DiscoveryPollInitialDelay = 0 * time.Millisecond

+ var DiscoveryPollInterval = 1 * time.Second

+ type BackoffConnectorFactory func(host host.Host) (*discimpl.BackoffConnector, error)

+ type DiscoverOpt func(*discoverOptions) error

+ func WithDiscoverConnector(connFactory BackoffConnectorFactory) DiscoverOpt

+ func WithDiscoveryOpts(opts ...discovery.Option) DiscoverOpt

type FloodSubRouter

+ func (fs *FloodSubRouter) EnoughPeers(topic string, suggested int) bool

type GossipSubRouter

+ func (gs *GossipSubRouter) EnoughPeers(topic string, suggested int) bool

type Message

+ ReceivedFrom peer.ID

type Option

+ func WithDiscovery(d discovery.Discovery, opts ...DiscoverOpt) Option

+ type PubOpt func(pub *PublishOptions) error

+ func WithReadiness(ready RouterReady) PubOpt

type PubSub

+ func (p *PubSub) Join(topic string, opts ...TopicOpt) (*Topic, error)

type PubSubRouter

+ EnoughPeers func(topic string, suggested int) bool

+ type PublishOptions struct

type RandomSubRouter

+ func (rs *RandomSubRouter) EnoughPeers(topic string, suggested int) bool

+ type RouterReady func(rt PubSubRouter, topic string) (bool, error)

+ func MinTopicSize(size int) RouterReady

+ type Topic struct

+ func (t *Topic) Close() error

+ func (t *Topic) EventHandler(opts ...TopicEventHandlerOpt) (*TopicEventHandler, error)

+ func (t *Topic) ListPeers() []peer.ID

+ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error

+ func (t *Topic) Subscribe(opts ...SubOpt) (*Subscription, error)

+ type TopicEventHandler struct

+ func (t *TopicEventHandler) Cancel()

+ func (t *TopicEventHandler) NextPeerEvent(ctx context.Context) (PeerEvent, error)

+ type TopicEventHandlerOpt func(t *TopicEventHandler) error

+ type TopicOpt func(t *Topic) error

+ type TopicOptions struct

Aug 7, 2019

Changes in this version

+ type EventType int

+ const PeerJoin

+ const PeerLeave

+ type PeerEvent struct

+ Peer peer.ID

+ Type EventType

type Subscription

+ func (sub *Subscription) NextPeerEvent(ctx context.Context) (PeerEvent, error)

May 15, 2019

Changes in this version

type Option

+ func WithValidateWorkers(n int) Option

type ValidatorOpt

+ func WithValidatorInline(inline bool) ValidatorOpt

Mar 2, 2019

Changes in this version

+ const FloodSubID

+ const GossipSubID

+ const RandomSubID

+ const SignPrefix

+ var GossipSubD = 6

+ var GossipSubDhi = 12

+ var GossipSubDlo = 4

+ var GossipSubFanoutTTL = 60 * time.Second

+ var GossipSubHeartbeatInitialDelay = 100 * time.Millisecond

+ var GossipSubHeartbeatInterval = 1 * time.Second

+ var GossipSubHistoryGossip = 3

+ var GossipSubHistoryLength = 5

+ var RandomSubD = 6

+ var TimeCacheDuration = 120 * time.Second

+ type Blacklist interface

+ Add func(peer.ID)

+ Contains func(peer.ID) bool

+ func NewLRUBlacklist(cap int) (Blacklist, error)

+ func NewMapBlacklist() Blacklist

+ type CacheEntry struct

+ type FloodSubRouter struct

+ func (fs *FloodSubRouter) AddPeer(peer.ID, protocol.ID)

+ func (fs *FloodSubRouter) Attach(p *PubSub)

+ func (fs *FloodSubRouter) HandleRPC(rpc *RPC)

+ func (fs *FloodSubRouter) Join(topic string)

+ func (fs *FloodSubRouter) Leave(topic string)

+ func (fs *FloodSubRouter) Protocols() []protocol.ID

+ func (fs *FloodSubRouter) Publish(from peer.ID, msg *pb.Message)

+ func (fs *FloodSubRouter) RemovePeer(peer.ID)

+ type GossipSubRouter struct

+ func (gs *GossipSubRouter) AddPeer(p peer.ID, proto protocol.ID)

+ func (gs *GossipSubRouter) Attach(p *PubSub)

+ func (gs *GossipSubRouter) HandleRPC(rpc *RPC)

+ func (gs *GossipSubRouter) Join(topic string)

+ func (gs *GossipSubRouter) Leave(topic string)

+ func (gs *GossipSubRouter) Protocols() []protocol.ID

+ func (gs *GossipSubRouter) Publish(from peer.ID, msg *pb.Message)

+ func (gs *GossipSubRouter) RemovePeer(p peer.ID)

+ type LRUBlacklist struct

+ func (b LRUBlacklist) Add(p peer.ID)

+ func (b LRUBlacklist) Contains(p peer.ID) bool

+ type MapBlacklist map[peer.ID]struct

+ func (b MapBlacklist) Add(p peer.ID)

+ func (b MapBlacklist) Contains(p peer.ID) bool

+ type Message struct

+ func (m *Message) GetFrom() peer.ID

+ type MessageCache struct

+ func NewMessageCache(gossip, history int) *MessageCache

+ func (mc *MessageCache) Get(mid string) (*pb.Message, bool)

+ func (mc *MessageCache) GetGossipIDs(topic string) []string

+ func (mc *MessageCache) Put(msg *pb.Message)

+ func (mc *MessageCache) Shift()

+ type Option func(*PubSub) error

+ func WithBlacklist(b Blacklist) Option

+ func WithMessageAuthor(author peer.ID) Option

+ func WithMessageSigning(enabled bool) Option

+ func WithStrictSignatureVerification(required bool) Option

+ func WithValidateThrottle(n int) Option

+ type PubSub struct

+ func NewFloodSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error)

+ func NewFloodsubWithProtocols(ctx context.Context, h host.Host, ps []protocol.ID, opts ...Option) (*PubSub, error)

+ func NewGossipSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error)

+ func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option) (*PubSub, error)

+ func NewRandomSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error)

+ func (p *PubSub) BlacklistPeer(pid peer.ID)

+ func (p *PubSub) GetTopics() []string

+ func (p *PubSub) ListPeers(topic string) []peer.ID

+ func (p *PubSub) Publish(topic string, data []byte) error

+ func (p *PubSub) RegisterTopicValidator(topic string, val Validator, opts ...ValidatorOpt) error

+ func (p *PubSub) Subscribe(topic string, opts ...SubOpt) (*Subscription, error)

+ func (p *PubSub) SubscribeByTopicDescriptor(td *pb.TopicDescriptor, opts ...SubOpt) (*Subscription, error)

+ func (p *PubSub) UnregisterTopicValidator(topic string) error

+ type PubSubNotif PubSub

+ func (p *PubSubNotif) ClosedStream(n inet.Network, s inet.Stream)

+ func (p *PubSubNotif) Connected(n inet.Network, c inet.Conn)

+ func (p *PubSubNotif) Disconnected(n inet.Network, c inet.Conn)

+ func (p *PubSubNotif) Listen(n inet.Network, _ ma.Multiaddr)

+ func (p *PubSubNotif) ListenClose(n inet.Network, _ ma.Multiaddr)

+ func (p *PubSubNotif) OpenedStream(n inet.Network, s inet.Stream)

+ type PubSubRouter interface

+ AddPeer func(peer.ID, protocol.ID)

+ Attach func(*PubSub)

+ HandleRPC func(*RPC)

+ Join func(topic string)

+ Leave func(topic string)

+ Protocols func() []protocol.ID

+ Publish func(peer.ID, *pb.Message)

+ RemovePeer func(peer.ID)

+ type RPC struct

+ type RandomSubRouter struct

+ func (rs *RandomSubRouter) AddPeer(p peer.ID, proto protocol.ID)

+ func (rs *RandomSubRouter) Attach(p *PubSub)

+ func (rs *RandomSubRouter) HandleRPC(rpc *RPC)

+ func (rs *RandomSubRouter) Join(topic string)

+ func (rs *RandomSubRouter) Leave(topic string)

+ func (rs *RandomSubRouter) Protocols() []protocol.ID

+ func (rs *RandomSubRouter) Publish(from peer.ID, msg *pb.Message)

+ func (rs *RandomSubRouter) RemovePeer(p peer.ID)

+ type SubOpt func(sub *Subscription) error

+ type Subscription struct

+ func (sub *Subscription) Cancel()

+ func (sub *Subscription) Next(ctx context.Context) (*Message, error)

+ func (sub *Subscription) Topic() string

+ type Validator func(context.Context, peer.ID, *Message) bool

+ type ValidatorOpt func(addVal *addValReq) error

+ func WithValidatorConcurrency(n int) ValidatorOpt

+ func WithValidatorTimeout(timeout time.Duration) ValidatorOpt