Skip to content

Commit

Permalink
coreapi pubsub: better ctx for connectToPubSubPeers
Browse files Browse the repository at this point in the history
License: MIT
Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
  • Loading branch information
magik6k committed Oct 2, 2018
1 parent 1facb7b commit 9581396
Showing 1 changed file with 8 additions and 4 deletions.
12 changes: 8 additions & 4 deletions core/coreapi/pubsub.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
type PubSubAPI CoreAPI

type pubSubSubscription struct {
cancel context.CancelFunc
subscription *floodsub.Subscription
}

Expand Down Expand Up @@ -75,27 +76,29 @@ func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
return nil, err
}

pubctx, cancel := context.WithCancel(api.node.Context())

if options.Discover {
go func() {
blk, err := api.core().Block().Put(ctx, strings.NewReader("floodsub:"+topic))
blk, err := api.core().Block().Put(pubctx, strings.NewReader("floodsub:"+topic))
if err != nil {
log.Error("pubsub discovery: ", err)
return
}

connectToPubSubPeers(ctx, api.node, blk.Path().Cid())
connectToPubSubPeers(pubctx, api.node, blk.Path().Cid())
}()
}

return &pubSubSubscription{sub}, nil
return &pubSubSubscription{cancel, sub}, nil
}

func connectToPubSubPeers(ctx context.Context, n *core.IpfsNode, cid cid.Cid) {
ctx, cancel := context.WithCancel(ctx)
defer cancel()

provs := n.Routing.FindProvidersAsync(ctx, cid, 10)
wg := &sync.WaitGroup{}
var wg sync.WaitGroup
for p := range provs {
wg.Add(1)
go func(pi pstore.PeerInfo) {
Expand Down Expand Up @@ -127,6 +130,7 @@ func (api *PubSubAPI) checkNode() error {
}

func (sub *pubSubSubscription) Close() error {
sub.cancel()
sub.subscription.Cancel()
return nil
}
Expand Down

0 comments on commit 9581396

Please sign in to comment.