Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

pubsub

package module

Go to main page

Versions in this module

v0

v0.4.0-bdw.3

Dec 10, 2020

v0.4.0-bdw.2

Dec 10, 2020

v0.4.0-bdw.1

Nov 24, 2020

v0.3.1-bdw.4

Oct 15, 2020

v0.3.1-bdw.3

Aug 23, 2020

v0.3.1-bdw.2

Aug 23, 2020

v0.3.1-bdw.1

Jul 5, 2020

v0.2.6

Apr 6, 2020

Changes in this version

+ const DefaultMaxMessageSize

+ const FloodSubID

+ const FloodSubTopicSearchSize

+ const GossipSubID

+ const RandomSubID

+ const RemoteTracerProtoID

+ const SignPrefix

+ var DiscoveryPollInitialDelay = 0 * time.Millisecond

+ var DiscoveryPollInterval = 1 * time.Second

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

+ 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 MinTraceBatchSize = 16

+ var RandomSubD = 6

+ var TimeCacheDuration = 120 * time.Second

+ var TraceBufferSize = 1 << 16

+ func DefaultMsgIdFn(pmsg *pb.Message) string

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

+ type Blacklist interface

+ Add func(peer.ID)

+ Contains func(peer.ID) bool

+ func NewMapBlacklist() Blacklist

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

+ type CacheEntry 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 FloodSubRouter struct

+ 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(from peer.ID, msg *pb.Message)

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

+ type GossipSubRouter struct

+ 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(from peer.ID, msg *pb.Message)

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

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

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

+ type Message struct

+ 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) (*pb.Message, bool)

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

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

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

+ func (mc *MessageCache) Shift()

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

+ type Option func(*PubSub) error

+ func WithBlacklist(b Blacklist) Option

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

+ func WithEventTracer(tracer EventTracer) Option

+ func WithMaxMessageSize(maxMessageSize int) Option

+ func WithMessageAuthor(author peer.ID) Option

+ func WithMessageIdFn(fn MsgIdFunction) Option

+ func WithMessageSigning(enabled bool) Option

+ func WithPeerOutboundQueueSize(size int) Option

+ func WithStrictSignatureVerification(required bool) Option

+ func WithValidateQueueSize(n int) Option

+ func WithValidateThrottle(n int) Option

+ func WithValidateWorkers(n int) 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 PeerEvent struct

+ Peer peer.ID

+ Type EventType

+ type PubOpt func(pub *PublishOptions) error

+ func WithReadiness(ready RouterReady) 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 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) Join(topic string, opts ...TopicOpt) (*Topic, error)

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

+ func (p *PubSub) Publish(topic string, data []byte, opts ...PubOpt) 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 network.Network, s network.Stream)

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

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

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

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

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

+ type PubSubRouter interface

+ 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(peer.ID, *pb.Message)

+ RemovePeer func(peer.ID)

+ type PublishOptions struct

+ 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) 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(from peer.ID, msg *pb.Message)

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

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

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

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

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

+ type ValidatorOpt func(addVal *addValReq) error

+ func WithValidatorConcurrency(n int) ValidatorOpt

+ func WithValidatorInline(inline bool) ValidatorOpt

+ func WithValidatorTimeout(timeout time.Duration) ValidatorOpt