pubsub

package module

Go to main page

Versions in this module

v1

v1.0.7

Feb 21, 2025

v1.0.6

Feb 11, 2025

v1.0.5

Feb 10, 2025

v1.0.4

Jan 23, 2025

v1.0.3

Jan 15, 2025

Changes in this version

type NodePubSub

+ func (pubsub *NodePubSub) Host() host.Host

v1.0.2

Jan 13, 2025

v1.0.1

Jan 10, 2025

v1.0.0

Jan 9, 2025

Changes in this version

+ func SetLog(filename string, stderr ...bool)

v0

v0.0.9

Jan 7, 2025

v0.0.8

Dec 27, 2024

v0.0.7

Dec 24, 2024

v0.0.6

Dec 20, 2024

v0.0.5

Dec 19, 2024

v0.0.4

Dec 19, 2024

Changes in this version

type NodePubSub

+ func (pubsub *NodePubSub) NotifyNewPeer(peer peer.ID) error

type PubSub

+ func (ps *PubSub) NotifyNewPeer(peer peer.ID) error

v0.0.3

Dec 19, 2024

Changes in this version

type NodeOption

+ func WithNodeDiscovery(d discovery.Discovery) NodeOption

type Options

+ func (opt *Options) GetNodeDiscovery() discovery.Discovery

v0.0.2

Dec 18, 2024

Changes in this version

+ const BackoffCleanupInterval

+ const BackoffMultiplier

+ const DefaultDecayInterval

+ const DefaultDecayToZero

+ const DefaultMaxMessageSize

+ const DefaultPubsubProtocol

+ const FloodSubID

+ const FloodSubTopicSearchSize

+ const GossipSubFeatureMesh

+ const GossipSubFeaturePX

+ const GossipSubID_v10

+ const GossipSubID_v11

+ const LaxNoSign

+ const LaxSign

+ const MaxBackoffAttempts

+ const MaxBackoffDelay

+ const MaxBackoffJitterCoff

+ const MinBackoffDelay

+ const RandomSubID

+ const RejectBlacklistedSource

+ const RejectBlacklstedPeer

+ const RejectInvalidSignature

+ const RejectMissingSignature

+ const RejectSelfOrigin

+ const RejectUnexpectedAuthInfo

+ const RejectUnexpectedSignature

+ const RejectValidationFailed

+ const RejectValidationIgnored

+ const RejectValidationQueueFull

+ const RejectValidationThrottled

+ const RemoteTracerProtoID

+ const SignPrefix

+ const StrictNoSign

+ const StrictSign

+ const TimeToLive

+ const ValidationAccept

+ const ValidationIgnore

+ const ValidationReject

+ 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 DiscoveryPollInitialDelay = 0 * time.Millisecond

+ var DiscoveryPollInterval = 1 * time.Second

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

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

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

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

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

+ var GossipSubConnTagBumpMessageDelivery = 1

+ var GossipSubConnTagDecayAmount = 1

+ var GossipSubConnTagDecayInterval = 10 * time.Minute

+ var GossipSubConnTagMessageDeliveryCap = 15

+ var GossipSubConnectionTimeout = 30 * time.Second

+ var GossipSubConnectors = 8

+ var GossipSubD = 6

+ var GossipSubDefaultProtocols = []protocol.ID

+ var GossipSubDhi = 12

+ var GossipSubDirectConnectInitialDelay = time.Second

+ var GossipSubDirectConnectTicks uint64 = 300

+ var GossipSubDlazy = 6

+ var GossipSubDlo = 5

+ var GossipSubDout = 2

+ var GossipSubDscore = 4

+ var GossipSubFanoutTTL = 60 * time.Second

+ var GossipSubGossipFactor = 0.25

+ var GossipSubGossipRetransmission = 3

+ var GossipSubGraftFloodThreshold = 10 * time.Second

+ var GossipSubHeartbeatInitialDelay = 100 * time.Millisecond

+ var GossipSubHeartbeatInterval = 1 * time.Second

+ var GossipSubHistoryGossip = 3

+ var GossipSubHistoryLength = 5

+ 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

+ var GossipSubUnsubscribeBackoff = 10 * time.Second

+ var MinTraceBatchSize = 16

+ var RandomSubD = 6

+ var TimeCacheDuration = 120 * time.Second

+ var TimeCacheStrategy = timecache.Strategy_FirstSeen

+ var TraceBufferSize = 1 << 16

+ func Debug(args ...interface{})

+ func Debugf(format string, args ...interface{})

+ func DefaultMsgIdFn(pmsg *pb.Message) string

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

+ func Error(args ...interface{})

+ func Errorf(format string, args ...interface{})

+ func Fatal(args ...interface{})

+ func Fatalf(format string, args ...interface{})

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

+ func GetLevel() logrus.Level

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

+ func Info(args ...interface{})

+ func Infof(format string, args ...interface{})

+ func LogMessage(level logrus.Level, args ...interface{})

+ func Logf(level logrus.Level, format string, args ...interface{})

+ func Panic(args ...interface{})

+ func Panicf(format string, args ...interface{})

+ func ParseLevel(level string) (logrus.Level, error)

+ func Print(args ...interface{})

+ func Printf(format string, args ...interface{})

+ func Println(args ...interface{})

+ func ScoreParameterDecay(decay time.Duration) float64

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

+ func SetLevel(level logrus.Level)

+ func SetOutput(output *os.File)

+ func SetReportCaller(reportCaller bool)

+ func Trace(args ...interface{})

+ func Tracef(format string, args ...interface{})

+ func Warn(args ...interface{})

+ func Warnf(format string, args ...interface{})

+ func WithError(err error) *logrus.Entry

+ func WithField(key string, value interface{}) *logrus.Entry

+ func WithFields(fields logrus.Fields) *logrus.Entry

+ type AcceptStatus int

+ const AcceptAll

+ const AcceptControl

+ const AcceptNone

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

+ type BasicSeqnoValidator struct

+ type Blacklist interface

+ Add func(peer.ID) bool

+ Contains func(peer.ID) bool

+ func NewMapBlacklist() Blacklist

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

+ type CacheEntry struct

+ type DefaultLogger struct

+ type DiscoverOpt func(*discoverOptions) error

+ func WithDiscoverConnector(connFactory BackoffConnectorFactory) DiscoverOpt

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

+ type EventTracer interface

+ Trace func(evt *pb.TraceEvent)

+ type EventType int

+ const PeerJoin

+ const PeerLeave

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

+ type FloodSubRouter struct

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

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

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

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

+ 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(msg *Message)

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

+ 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

+ SlowHeartbeatWarning float64

+ UnsubscribeBackoff time.Duration

+ func DefaultGossipSubParams() GossipSubParams

+ type GossipSubRouter struct

+ func DefaultGossipSubRouter(h host.Host) *GossipSubRouter

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

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

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

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

+ 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(msg *Message)

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

+ func (gs *GossipSubRouter) WithDefaultTagTracer() Option

+ 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 MapBlacklist map[peer.ID]struct

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

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

+ type Message struct

+ ID string

+ Local bool

+ ReceivedFrom peer.ID

+ ValidatorData interface{}

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

+ type MessageCache struct

+ func NewMessageCache(gossip, history int) *MessageCache

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

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

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

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

+ func (mc *MessageCache) SetMsgIdFn(msgID func(*Message) string)

+ func (mc *MessageCache) Shift()

+ type MessageMetadataOpt struct

+ type MessageSignaturePolicy uint8

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

+ type Node struct

+ CheckInterval time.Duration

+ ConnectionQuality float64

+ FailedAttempts int

+ History []StatusChange

+ ID peer.ID

+ LastCheckTime time.Time

+ LastSeen time.Time

+ NetworkCondition []time.Duration

+ Score float64

+ Status NodeStatus

+ SuccessiveSuccesses int

+ type NodeOption func(*Options) error

+ func WithSetD(d int) NodeOption

+ func WithSetDirectPeers(peers []peer.AddrInfo) NodeOption

+ func WithSetDlo(dlo int) NodeOption

+ func WithSetFollowupTime(t time.Duration) NodeOption

+ func WithSetGossipFactor(f float64) NodeOption

+ func WithSetHeartbeatInterval(interval time.Duration) NodeOption

+ func WithSetLoadConfig(load bool) NodeOption

+ func WithSetMaxMessageSize(size int) NodeOption

+ func WithSetMaxPendingConns(n int) NodeOption

+ func WithSetMaxTransmissionSize(size int) NodeOption

+ func WithSetPubSubMode(mode PubSubType) NodeOption

+ func WithSetSignMessages(sign bool) NodeOption

+ func WithSetValidateMessages(validate bool) NodeOption

+ type NodePubSub struct

+ func NewNodePubSub(ctx context.Context, host host.Host, opts ...NodeOption) (*NodePubSub, error)

+ func (pubsub *NodePubSub) BroadcastWithTopic(topic string, data []byte) error

+ func (pubsub *NodePubSub) CancelPubsubWithTopic(name string) error

+ func (pubsub *NodePubSub) CancelSubscribeWithTopic(topic string) error

+ func (pubsub *NodePubSub) GetTopic(name string) (*Topic, error)

+ func (pubsub *NodePubSub) IsSubscribed(topic string) bool

+ func (pubsub *NodePubSub) ListPeers(topic string) []peer.ID

+ func (pubsub *NodePubSub) Publish(topic string, data []byte) error

+ func (pubsub *NodePubSub) Pubsub() *PubSub

+ func (pubsub *NodePubSub) Subscribe(topic string, subscribe bool) (*Subscription, error)

+ func (pubsub *NodePubSub) SubscribeWithTopic(topic string, handler PubSubMsgHandler, subscribe bool) error

+ type NodeStatus int

+ const Offline

+ const Online

+ const Suspicious

+ type NodeStatusTracker struct

+ func NewNodeStatusTracker(h host.Host, defaultCheckInterval time.Duration, offlineThreshold int, ...) *NodeStatusTracker

+ func (nst *NodeStatusTracker) Start()

+ func (nst *NodeStatusTracker) Stop()

+ func (nst *NodeStatusTracker) SubscribeStatusChanges() chan StatusChange

+ func (nst *NodeStatusTracker) UnsubscribeStatusChanges(ch chan StatusChange)

+ func (nst *NodeStatusTracker) UpdateNodeStatus(pid peer.ID, status NodeStatus)

+ type Option func(*PubSub) error

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

+ func WithBlacklist(b Blacklist) Option

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

+ func WithDirectConnectTicks(t uint64) Option

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

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

+ func WithEventTracer(tracer EventTracer) Option

+ func WithFloodPublish(floodPublish bool) Option

+ func WithGossipSubParams(cfg GossipSubParams) Option

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

+ func WithMaxMessageSize(maxMessageSize int) Option

+ func WithMessageAuthor(author peer.ID) Option

+ func WithMessageIdFn(fn MsgIdFunction) Option

+ func WithMessageSignaturePolicy(policy MessageSignaturePolicy) Option

+ func WithMessageSigning(enabled bool) Option

+ func WithNoAuthor() Option

+ func WithPeerExchange(doPX bool) Option

+ func WithPeerFilter(filter PeerFilter) Option

+ func WithPeerGater(params *PeerGaterParams) Option

+ func WithPeerOutboundQueueSize(size int) Option

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

+ func WithPeerScoreInspect(inspect interface{}, period time.Duration) Option

+ func WithProtocolMatchFn(m ProtocolMatchFn) Option

+ func WithRawTracer(tracer RawTracer) Option

+ func WithRetry(retries int) Option

+ func WithSeenMessagesStrategy(strategy timecache.Strategy) Option

+ func WithSeenMessagesTTL(ttl time.Duration) Option

+ func WithStrictSignatureVerification(required bool) Option

+ func WithSubscriptionFilter(subFilter SubscriptionFilter) Option

+ func WithTimeout(d time.Duration) Option

+ func WithValidateQueueSize(n int) Option

+ func WithValidateThrottle(n int) Option

+ func WithValidateWorkers(n int) Option

+ type Options struct

+ D int

+ DirectPeers []peer.AddrInfo

+ Dlo int

+ FollowupTime time.Duration

+ GossipFactor float64

+ HeartbeatInterval time.Duration

+ LoadConfig bool

+ MaxMessageSize int

+ MaxPendingConns int

+ MaxTransmissionSize int

+ PubSubMode PubSubType

+ SignMessages bool

+ ValidateMessages bool

+ func DefaultOptions() *Options

+ func (o *Options) GetD() int

+ func (o *Options) GetDirectPeers() []peer.AddrInfo

+ func (o *Options) GetDlo() int

+ func (o *Options) GetFollowupTime() time.Duration

+ func (o *Options) GetGossipFactor() float64

+ func (o *Options) GetHeartbeatInterval() time.Duration

+ func (o *Options) GetLoadConfig() bool

+ func (o *Options) GetMaxMessageSize() int

+ func (o *Options) GetMaxPendingConns() int

+ func (o *Options) GetMaxTransmissionSize() int

+ func (o *Options) GetPubSubMode() PubSubType

+ func (o *Options) GetSignMessages() bool

+ func (o *Options) GetValidateMessages() bool

+ func (opt *Options) ApplyOptions(opts ...NodeOption) error

+ 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 PeerEvent struct

+ Peer peer.ID

+ Type EventType

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

+ 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 PeerMetadataStore interface

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

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

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

+ type PeerScoreParams struct

+ AppSpecificScore func(p peer.ID) float64

+ AppSpecificWeight float64

+ BehaviourPenaltyDecay float64

+ BehaviourPenaltyThreshold float64

+ BehaviourPenaltyWeight float64

+ DecayInterval time.Duration

+ DecayToZero float64

+ IPColocationFactorThreshold int

+ IPColocationFactorWeight float64

+ IPColocationFactorWhitelist []*net.IPNet

+ RetainScore time.Duration

+ SeenMsgTTL time.Duration

+ SkipAtomicValidation bool

+ TopicScoreCap float64

+ Topics map[string]*TopicScoreParams

+ type PeerScoreSnapshot struct

+ AppSpecificScore float64

+ BehaviourPenalty float64

+ IPColocationFactor float64

+ Score float64

+ Topics map[string]*TopicScoreSnapshot

+ type PeerScoreThresholds struct

+ AcceptPXThreshold float64

+ GossipThreshold float64

+ GraylistThreshold float64

+ OpportunisticGraftThreshold float64

+ PublishThreshold float64

+ SkipAtomicValidation bool

+ type ProtocolMatchFn = func(protocol.ID) func(protocol.ID) bool

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

+ type PubOpt func(pub *PublishOptions) error

+ func WithLocalPublication(local bool) PubOpt

+ func WithMessageMetadata(messageID string, msgType pb.MessageMetadata_MessageType) PubOpt

+ func WithReadiness(ready RouterReady) PubOpt

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

+ func WithTargetMap(targets []peer.ID) PubOpt

+ 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 NewGossipSubWithRouter(ctx context.Context, h host.Host, rt PubSubRouter, 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, size int, opts ...Option) (*PubSub, error)

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

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

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

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

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

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

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

+ type PubSubMsgHandler func(*Message)

+ type PubSubRouter interface

+ AcceptFrom func(peer.ID) AcceptStatus

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

+ Attach func(*PubSub)

+ EnoughPeers func(topic string, suggested int) bool

+ HandleRPC func(*RPC)

+ Join func(topic string)

+ Leave func(topic string)

+ Protocols func() []protocol.ID

+ Publish func(*Message)

+ RemovePeer func(peer.ID)

+ type PubSubType int

+ const FloodSub

+ const GossipSub

+ const RandomSub

+ type PublishOptions struct

+ type RPC struct

+ type RandomSubRouter struct

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

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

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

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

+ 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(msg *Message)

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

+ 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 RelayCancelFunc func()

+ 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)

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

+ func MinTopicSize(size int) RouterReady

+ type StatusChange struct

+ CheckInterval time.Duration

+ ConnectionQuality float64

+ FailedAttempts int

+ NetworkLatency time.Duration

+ NewStatus NodeStatus

+ OldStatus NodeStatus

+ PeerID peer.ID

+ Score float64

+ Timestamp time.Time

+ type SubOpt func(sub *Subscription) error

+ func WithBufferSize(size int) SubOpt

+ type Subscription struct

+ func (sub *Subscription) Cancel()

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

+ func (sub *Subscription) Topic() string

+ type SubscriptionFilter interface

+ CanSubscribe func(topic string) bool

+ FilterIncomingSubscriptions func(from peer.ID, subs []*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 TimeCachedBlacklist struct

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

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

+ 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) PublishWithReply(ctx context.Context, data []byte, targetNodes ...peer.ID) ([]byte, error)

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

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

+ func (t *Topic) String() string

+ 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

+ func WithTopicMessageIdFn(msgId MsgIdFunction) TopicOpt

+ type TopicOptions struct

+ 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

+ SkipAtomicValidation bool

+ TimeInMeshCap float64

+ TimeInMeshQuantum time.Duration

+ TimeInMeshWeight float64

+ TopicWeight float64

+ type TopicScoreSnapshot struct

+ FirstMessageDeliveries float64

+ InvalidMessageDeliveries float64

+ MeshMessageDeliveries float64

+ TimeInMesh time.Duration

+ type ValidationError struct

+ Reason string

+ func (e ValidationError) Error() string

+ type ValidationResult int

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

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

+ func NewBasicSeqnoValidator(meta PeerMetadataStore) ValidatorEx

+ type ValidatorOpt func(addVal *addValReq) error

+ func WithValidatorConcurrency(n int) ValidatorOpt

+ func WithValidatorInline(inline bool) ValidatorOpt

+ func WithValidatorTimeout(timeout time.Duration) ValidatorOpt

v0.0.1

Oct 9, 2024

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL