2018-08-28 03:01:08 +00:00
|
|
|
package pubsub
|
2016-09-10 00:15:39 +00:00
|
|
|
|
|
|
|
import (
|
2016-09-10 23:03:53 +00:00
|
|
|
"context"
|
2016-09-10 00:15:39 +00:00
|
|
|
|
2019-05-26 16:19:03 +00:00
|
|
|
"github.com/libp2p/go-libp2p-core/host"
|
|
|
|
"github.com/libp2p/go-libp2p-core/peer"
|
|
|
|
"github.com/libp2p/go-libp2p-core/protocol"
|
2016-09-10 00:15:39 +00:00
|
|
|
)
|
|
|
|
|
2017-11-22 17:40:45 +00:00
|
|
|
const (
|
2019-10-31 19:56:09 +00:00
|
|
|
FloodSubID = protocol.ID("/floodsub/1.0.0")
|
|
|
|
FloodSubTopicSearchSize = 5
|
2017-11-22 17:40:45 +00:00
|
|
|
)
|
2016-09-10 00:15:39 +00:00
|
|
|
|
2019-01-04 11:09:21 +00:00
|
|
|
// NewFloodsubWithProtocols returns a new floodsub-enabled PubSub objecting using the protocols specified in ps.
|
2018-06-04 21:14:54 +00:00
|
|
|
func NewFloodsubWithProtocols(ctx context.Context, h host.Host, ps []protocol.ID, opts ...Option) (*PubSub, error) {
|
2018-06-01 22:50:24 +00:00
|
|
|
rt := &FloodSubRouter{
|
|
|
|
protocols: ps,
|
|
|
|
}
|
|
|
|
return NewPubSub(ctx, h, rt, opts...)
|
|
|
|
}
|
|
|
|
|
2019-01-04 11:09:21 +00:00
|
|
|
// NewFloodSub returns a new PubSub object using the FloodSubRouter.
|
2017-12-16 12:12:23 +00:00
|
|
|
func NewFloodSub(ctx context.Context, h host.Host, opts ...Option) (*PubSub, error) {
|
2018-06-04 21:14:54 +00:00
|
|
|
return NewFloodsubWithProtocols(ctx, h, []protocol.ID{FloodSubID}, opts...)
|
2016-09-10 00:15:39 +00:00
|
|
|
}
|
2016-09-10 03:13:50 +00:00
|
|
|
|
2018-01-24 14:04:39 +00:00
|
|
|
type FloodSubRouter struct {
|
2018-06-01 22:50:24 +00:00
|
|
|
p *PubSub
|
|
|
|
protocols []protocol.ID
|
2019-11-04 18:43:48 +00:00
|
|
|
tracer *pubsubTracer
|
2016-10-19 23:01:06 +00:00
|
|
|
}
|
|
|
|
|
2018-01-24 14:04:39 +00:00
|
|
|
func (fs *FloodSubRouter) Protocols() []protocol.ID {
|
2018-06-01 22:50:24 +00:00
|
|
|
return fs.protocols
|
2016-09-10 03:13:50 +00:00
|
|
|
}
|
2016-09-10 00:15:39 +00:00
|
|
|
|
2018-01-24 14:04:39 +00:00
|
|
|
func (fs *FloodSubRouter) Attach(p *PubSub) {
|
|
|
|
fs.p = p
|
2019-11-04 18:47:19 +00:00
|
|
|
fs.tracer = p.tracer
|
2016-09-10 00:15:39 +00:00
|
|
|
}
|
|
|
|
|
2019-11-04 18:43:48 +00:00
|
|
|
func (fs *FloodSubRouter) AddPeer(p peer.ID, proto protocol.ID) {
|
|
|
|
fs.tracer.AddPeer(p, proto)
|
|
|
|
}
|
2017-12-16 12:12:23 +00:00
|
|
|
|
2019-11-04 18:43:48 +00:00
|
|
|
func (fs *FloodSubRouter) RemovePeer(p peer.ID) {
|
|
|
|
fs.tracer.RemovePeer(p)
|
|
|
|
}
|
2018-01-13 16:34:00 +00:00
|
|
|
|
2019-10-31 19:56:09 +00:00
|
|
|
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
|
|
|
|
}
|
|
|
|
|
2018-01-27 07:52:35 +00:00
|
|
|
func (fs *FloodSubRouter) HandleRPC(rpc *RPC) {}
|
2016-09-10 00:15:39 +00:00
|
|
|
|
2020-03-03 13:44:44 +00:00
|
|
|
func (fs *FloodSubRouter) Publish(from peer.ID, msg *Message) {
|
2016-09-12 20:22:16 +00:00
|
|
|
tosend := make(map[peer.ID]struct{})
|
|
|
|
for _, topic := range msg.GetTopicIDs() {
|
2018-01-24 14:04:39 +00:00
|
|
|
tmap, ok := fs.p.topics[topic]
|
2016-09-12 20:22:16 +00:00
|
|
|
if !ok {
|
|
|
|
continue
|
|
|
|
}
|
2016-09-11 03:47:12 +00:00
|
|
|
|
2017-11-23 13:39:14 +00:00
|
|
|
for p := range tmap {
|
2016-09-12 20:22:16 +00:00
|
|
|
tosend[p] = struct{}{}
|
|
|
|
}
|
2016-09-10 00:15:39 +00:00
|
|
|
}
|
|
|
|
|
2020-03-03 13:44:44 +00:00
|
|
|
out := rpcWithMessages(msg.Message)
|
2016-09-12 20:22:16 +00:00
|
|
|
for pid := range tosend {
|
2016-09-11 03:47:12 +00:00
|
|
|
if pid == from || pid == peer.ID(msg.GetFrom()) {
|
2016-09-10 00:15:39 +00:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2018-01-24 14:04:39 +00:00
|
|
|
mch, ok := fs.p.peers[pid]
|
2016-09-10 00:15:39 +00:00
|
|
|
if !ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2017-08-30 02:11:58 +00:00
|
|
|
select {
|
|
|
|
case mch <- out:
|
2019-11-04 18:43:48 +00:00
|
|
|
fs.tracer.SendRPC(out, pid)
|
2017-08-30 02:11:58 +00:00
|
|
|
default:
|
2017-08-30 02:42:33 +00:00
|
|
|
log.Infof("dropping message to peer %s: queue full", pid)
|
2019-11-04 18:43:48 +00:00
|
|
|
fs.tracer.DropRPC(out, pid)
|
2017-08-30 02:11:58 +00:00
|
|
|
// Drop it. The peer is too slow.
|
|
|
|
}
|
2016-09-10 00:15:39 +00:00
|
|
|
}
|
|
|
|
}
|
2018-02-03 09:30:19 +00:00
|
|
|
|
2019-11-04 18:43:48 +00:00
|
|
|
func (fs *FloodSubRouter) Join(topic string) {
|
|
|
|
fs.tracer.Join(topic)
|
|
|
|
}
|
2018-02-03 09:30:19 +00:00
|
|
|
|
2019-11-04 18:43:48 +00:00
|
|
|
func (fs *FloodSubRouter) Leave(topic string) {
|
|
|
|
fs.tracer.Join(topic)
|
|
|
|
}
|