fix: remove pubsub discovery hack
Pubsub handles this internally now.
Steven Allen committed
May 29, 2020 at 15:54 UTC
c58e3e4c7d400c238846b42376e52c555b44229b
3 files changed
+9
-59
core/commands/pubsub.go
+2
-4
@@ -74,7 +74,7 @@ This command outputs data in the following encodings:
74
cmds.StringArg("topic", true, false, "String name of topic to subscribe to."),
75
},
76
Options: []cmds.Option{
77
- cmds.BoolOption(pubsubDiscoverOptionName, "try to discover other peers subscribed to the same topic"),
77
+ cmds.BoolOption(pubsubDiscoverOptionName, "Deprecated option to instruct pubsub to discovery peers for the topic. Discovery is now built into pubsub."),
78
},
79
Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
80
api, err := cmdenv.GetApi(env, req)
@@ -83,9 +83,7 @@ This command outputs data in the following encodings:
83
}
84
85
topic := req.Arguments[0]
86
- discover, _ := req.Options[pubsubDiscoverOptionName].(bool)
87
-
88
- sub, err := api.PubSub().Subscribe(req.Context, topic, options.PubSub.Discover(discover))
86
+ sub, err := api.PubSub().Subscribe(req.Context, topic)
87
if err != nil {
88
return err
89
}
core/coreapi/coreapi.go
-3
@@ -26,7 +26,6 @@ import (
26
"github.com/ipfs/go-ipfs-provider"
27
offlineroute "github.com/ipfs/go-ipfs-routing/offline"
28
ipld "github.com/ipfs/go-ipld-format"
29
- logging "github.com/ipfs/go-log"
29
dag "github.com/ipfs/go-merkledag"
30
coreiface "github.com/ipfs/interface-go-ipfs-core"
31
"github.com/ipfs/interface-go-ipfs-core/options"
@@ -44,8 +43,6 @@ import (
43
"github.com/ipfs/go-ipfs/repo"
44
)
45
47
-var log = logging.Logger("core/coreapi")
48
-
46
type CoreAPI struct {
47
nctx context.Context
48
core/coreapi/pubsub.go
+7
-52
@@ -3,14 +3,9 @@ package coreapi
3
import (
4
"context"
5
"errors"
6
- "strings"
7
- "sync"
8
- "time"
6
10
- cid "github.com/ipfs/go-cid"
7
coreiface "github.com/ipfs/interface-go-ipfs-core"
8
caopts "github.com/ipfs/interface-go-ipfs-core/options"
13
- p2phost "github.com/libp2p/go-libp2p-core/host"
9
peer "github.com/libp2p/go-libp2p-core/peer"
10
routing "github.com/libp2p/go-libp2p-core/routing"
11
pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -19,7 +14,6 @@ import (
14
type PubSubAPI CoreAPI
15
16
type pubSubSubscription struct {
22
- cancel context.CancelFunc
17
subscription *pubsub.Subscription
18
}
19
@@ -61,12 +55,16 @@ func (api *PubSubAPI) Publish(ctx context.Context, topic string, data []byte) er
55
}
56
57
func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (coreiface.PubSubSubscription, error) {
64
- options, err := caopts.PubSubSubscribeOptions(opts...)
58
+ // Parse the options to avoid introducing silent failures for invalid
59
+ // options. However, we don't currently have any use for them. The only
60
+ // subscription option, discovery, is now a no-op as it's handled by
61
+ // pubsub itself.
62
+ _, err := caopts.PubSubSubscribeOptions(opts...)
63
if err != nil {
64
return nil, err
65
}
66
69
- r, err := api.checkNode()
67
+ _, err = api.checkNode()
68
if err != nil {
69
return nil, err
70
}
@@ -77,45 +75,7 @@ func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
75
return nil, err
76
}
77
80
- pubctx, cancel := context.WithCancel(api.nctx)
81
-
82
- if options.Discover {
83
- go func() {
84
- blk, err := api.core().Block().Put(pubctx, strings.NewReader("floodsub:"+topic))
85
- if err != nil {
86
- log.Error("pubsub discovery: ", err)
87
- return
88
- }
89
-
90
- connectToPubSubPeers(pubctx, r, api.peerHost, blk.Path().Cid())
91
- }()
92
- }
93
-
94
- return &pubSubSubscription{cancel, sub}, nil
95
-}
96
-
97
-func connectToPubSubPeers(ctx context.Context, r routing.Routing, ph p2phost.Host, cid cid.Cid) {
98
- ctx, cancel := context.WithCancel(ctx)
99
- defer cancel()
100
-
101
- provs := r.FindProvidersAsync(ctx, cid, 10)
102
- var wg sync.WaitGroup
103
- for p := range provs {
104
- wg.Add(1)
105
- go func(pi peer.AddrInfo) {
106
- defer wg.Done()
107
- ctx, cancel := context.WithTimeout(ctx, time.Second*10)
108
- defer cancel()
109
- err := ph.Connect(ctx, pi)
110
- if err != nil {
111
- log.Info("pubsub discover: ", err)
112
- return
113
- }
114
- log.Info("connected to pubsub peer:", pi.ID)
115
- }(p)
116
- }
117
-
118
- wg.Wait()
78
+ return &pubSubSubscription{sub}, nil
79
}
80
81
func (api *PubSubAPI) checkNode() (routing.Routing, error) {
@@ -132,7 +92,6 @@ func (api *PubSubAPI) checkNode() (routing.Routing, error) {
92
}
93
94
func (sub *pubSubSubscription) Close() error {
135
- sub.cancel()
95
sub.subscription.Cancel()
96
return nil
97
}
@@ -161,7 +120,3 @@ func (msg *pubSubMessage) Seq() []byte {
120
func (msg *pubSubMessage) Topics() []string {
121
return msg.msg.TopicIDs
122
}
164
-
165
-func (api *PubSubAPI) core() coreiface.CoreAPI {
166
- return (*CoreAPI)(api)
167
-}