client/rpc: use streaming pin listing
This also fix a leaking goroutine bug on client/rpc.PinAPI.Ls, we would deadlock if context was canceled while writing the keys.
Jorropo committed
Jun 2, 2023 at 19:22 UTC
f8f4b83c9c473ebd01e9624a39097a429f65c572
1 file changed
+43
-11
client/rpc/pin.go
+43
-11
@@ -29,15 +29,15 @@ type pin struct {
29
err error
30
}
31
32
-func (p *pin) Err() error {
32
+func (p pin) Err() error {
33
return p.err
34
}
35
36
-func (p *pin) Path() path.Resolved {
36
+func (p pin) Path() path.Resolved {
37
return p.path
38
}
39
40
-func (p *pin) Type() string {
40
+func (p pin) Type() string {
41
return p.typ
42
}
43
@@ -51,29 +51,61 @@ func (api *PinAPI) Add(ctx context.Context, p path.Path, opts ...caopts.PinAddOp
51
Option("recursive", options.Recursive).Exec(ctx, nil)
52
}
53
54
+type pinLsObject struct {
55
+ Cid string
56
+ Type string
57
+}
58
+
59
func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) (<-chan iface.Pin, error) {
60
options, err := caopts.PinLsOptions(opts...)
61
if err != nil {
62
return nil, err
63
}
64
60
- var out pinRefKeyList
61
- err = api.core().Request("pin/ls").
62
- Option("type", options.Type).Exec(ctx, &out)
65
+ res, err := api.core().Request("pin/ls").
66
+ Option("type", options.Type).
67
+ Option("stream", true).
68
+ Send(ctx)
69
if err != nil {
70
return nil, err
71
}
72
73
pins := make(chan iface.Pin)
74
go func(ch chan<- iface.Pin) {
75
+ defer res.Output.Close()
76
defer close(ch)
70
- for hash, p := range out.Keys {
71
- c, e := cid.Parse(hash)
72
- if e != nil {
73
- ch <- &pin{typ: p.Type, err: e}
77
+
78
+ dec := json.NewDecoder(res.Output)
79
+ var out pinLsObject
80
+ for {
81
+ switch err := dec.Decode(&out); err {
82
+ case nil:
83
+ case io.EOF:
84
+ return
85
+ default:
86
+ select {
87
+ case ch <- pin{err: err}:
88
+ return
89
+ case <-ctx.Done():
90
+ return
91
+ }
92
+ }
93
+
94
+ c, err := cid.Parse(out.Cid)
95
+ if err != nil {
96
+ select {
97
+ case ch <- pin{err: err}:
98
+ return
99
+ case <-ctx.Done():
100
+ return
101
+ }
102
+ }
103
+
104
+ select {
105
+ case ch <- pin{typ: out.Type, path: path.IpldPath(c)}:
106
+ case <-ctx.Done():
107
return
108
}
76
- ch <- &pin{typ: p.Type, path: path.IpldPath(c), err: e}
109
}
110
}(pins)
111
return pins, nil