mirror of
https://github.com/logos-messaging/go-libp2p-pubsub.git
synced 2026-01-02 12:53:09 +00:00
9e5145fb29
See #610 We previously send IDONTWANT only when forwarding. This has us send IDONTWANT on our initial publish as well. Helps in the case that one or more peers may also publish the same thing at around the same time (see #610 for a longer explanation) and prevents "boomerang" duplicates where a peer sends you back the message you sent before you get a chance to send it to them. This also serves as a hint to a peer that you are about to send them a certain message.
111 lines
2.4 KiB
Go
111 lines
2.4 KiB
Go
package pubsub
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/libp2p/go-libp2p/core/host"
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
|
"github.com/libp2p/go-libp2p/core/protocol"
|
|
)
|
|
|
|
const (
|
|
FloodSubID = protocol.ID("/floodsub/1.0.0")
|
|
FloodSubTopicSearchSize = 5
|
|
)
|
|
|
|
// NewFloodsubWithProtocols returns a new floodsub-enabled PubSub objecting using the protocols specified in ps.
|
|
func NewFloodsubWithProtocols(ctx context.Context, h host.Host, ps []protocol.ID, opts ...Option) (*PubSub, error) {
|
|
rt := &FloodSubRouter{
|
|
protocols: ps,
|
|
}
|
|
return NewPubSub(ctx, h, rt, opts...)
|
|
}
|
|
|
|
// NewFloodSub returns a new PubSub object using the FloodSubRouter.
|
|
func NewFloodSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error) {
|
|
return NewFloodsubWithProtocols(ctx, h, []protocol.ID{FloodSubID}, opts...)
|
|
}
|
|
|
|
type FloodSubRouter struct {
|
|
p *PubSub
|
|
protocols []protocol.ID
|
|
tracer *pubsubTracer
|
|
}
|
|
|
|
func (fs *FloodSubRouter) Protocols() []protocol.ID {
|
|
return fs.protocols
|
|
}
|
|
|
|
func (fs *FloodSubRouter) Attach(p *PubSub) {
|
|
fs.p = p
|
|
fs.tracer = p.tracer
|
|
}
|
|
|
|
func (fs *FloodSubRouter) AddPeer(p peer.ID, proto protocol.ID) {
|
|
fs.tracer.AddPeer(p, proto)
|
|
}
|
|
|
|
func (fs *FloodSubRouter) RemovePeer(p peer.ID) {
|
|
fs.tracer.RemovePeer(p)
|
|
}
|
|
|
|
func (fs *FloodSubRouter) EnoughPeers(topic string, suggested int) bool {
|
|
// check all peers in the topic
|
|
tmap, ok := fs.p.topics[topic]
|
|
if !ok {
|
|
return false
|
|
}
|
|
|
|
if suggested == 0 {
|
|
suggested = FloodSubTopicSearchSize
|
|
}
|
|
|
|
if len(tmap) >= suggested {
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func (fs *FloodSubRouter) AcceptFrom(peer.ID) AcceptStatus {
|
|
return AcceptAll
|
|
}
|
|
|
|
func (fs *FloodSubRouter) Preprocess(from peer.ID, msgs []*Message) {}
|
|
|
|
func (fs *FloodSubRouter) HandleRPC(rpc *RPC) {}
|
|
|
|
func (fs *FloodSubRouter) Publish(msg *Message) {
|
|
from := msg.ReceivedFrom
|
|
topic := msg.GetTopic()
|
|
|
|
out := rpcWithMessages(msg.Message)
|
|
for pid := range fs.p.topics[topic] {
|
|
if pid == from || pid == peer.ID(msg.GetFrom()) {
|
|
continue
|
|
}
|
|
|
|
q, ok := fs.p.peers[pid]
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
err := q.Push(out, false)
|
|
if err != nil {
|
|
log.Infof("dropping message to peer %s: queue full", pid)
|
|
fs.tracer.DropRPC(out, pid)
|
|
// Drop it. The peer is too slow.
|
|
continue
|
|
}
|
|
fs.tracer.SendRPC(out, pid)
|
|
}
|
|
}
|
|
|
|
func (fs *FloodSubRouter) Join(topic string) {
|
|
fs.tracer.Join(topic)
|
|
}
|
|
|
|
func (fs *FloodSubRouter) Leave(topic string) {
|
|
fs.tracer.Leave(topic)
|
|
}
|