Close provider on ipfs shutdown
License: MIT Signed-off-by: Michael Avila <davidmichaelavila@gmail.com>
Michael Avila committed
Apr 5, 2019 at 09:52 UTC
572844c2620c7063e8e05f339f6b8d01683620b8
4 files changed
+33
-1
core/core.go
+4
@@ -675,6 +675,10 @@ func (n *IpfsNode) teardown() error {
675
// needs to use another during its shutdown/cleanup process, it should be
676
// closed before that other object
677
678
+ if n.Provider != nil {
679
+ closers = append(closers, n.Provider)
680
+ }
681
+
682
if n.FilesRoot != nil {
683
closers = append(closers, n.FilesRoot)
684
}
provider/offline.go
+4
@@ -14,3 +14,7 @@ func (op *offlineProvider) Run() {}
14
func (op *offlineProvider) Provide(cid cid.Cid) error {
15
return nil
16
}
17
+
18
+func (op *offlineProvider) Close() error {
19
+ return nil
20
+}
provider/provider.go
+8
@@ -20,6 +20,8 @@ type Provider interface {
20
Run()
21
// Provide takes a cid and makes an attempt to announce it to the network
22
Provide(cid.Cid) error
23
+ // Close stops the provider
24
+ Close() error
25
}
26
27
type provider struct {
@@ -39,6 +41,12 @@ func NewProvider(ctx context.Context, queue *Queue, contentRouting routing.Conte
41
}
42
}
43
44
+// Close stops the provider
45
+func (p *provider) Close() error {
46
+ p.queue.Close()
47
+ return nil
48
+}
49
+
50
// Start workers to handle provide requests.
51
func (p *provider) Run() {
52
p.handleAnnouncements()
provider/queue.go
+17
-1
@@ -27,6 +27,8 @@ type Queue struct {
27
ds datastore.Datastore // Must be threadsafe
28
dequeue chan cid.Cid
29
enqueue chan cid.Cid
30
+ close context.CancelFunc
31
+ closed chan struct{}
32
}
33
34
// NewQueue creates a queue for cids
@@ -36,19 +38,29 @@ func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue,
38
if err != nil {
39
return nil, err
40
}
41
+ cancelCtx, cancel := context.WithCancel(ctx)
42
q := &Queue{
43
name: name,
41
- ctx: ctx,
44
+ ctx: cancelCtx,
45
head: head,
46
tail: tail,
47
ds: namespaced,
48
dequeue: make(chan cid.Cid),
49
enqueue: make(chan cid.Cid),
50
+ close: cancel,
51
+ closed: make(chan struct{}, 1),
52
}
53
q.work()
54
return q, nil
55
}
56
57
+// Close stops the queue
58
+func (q *Queue) Close() error {
59
+ q.close()
60
+ <-q.closed
61
+ return nil
62
+}
63
+
64
// Enqueue puts a cid in the queue
65
func (q *Queue) Enqueue(cid cid.Cid) {
66
select {
@@ -103,6 +115,10 @@ func (q *Queue) work() {
115
var k datastore.Key = datastore.Key{}
116
var c cid.Cid = cid.Undef
117
118
+ defer func() {
119
+ close(q.closed)
120
+ }()
121
+
122
for {
123
if c == cid.Undef {
124
k, c = q.nextEntry()