Unixfs.Add progress events
This commit was moved from ipfs/go-ipfs-http-client@2f3a77b686ee8c4194ee6c21cbdecd57f2081519
Łukasz Magiera committed
Jan 9, 2019 at 21:55 UTC
c213e2654235ffbdb6ce2f3a7df3c027ba53e02f
10 files changed
+71
-32
client/httpapi/api.go
+2
-2
@@ -99,8 +99,8 @@ func NewApiWithClient(a ma.Multiaddr, c *gohttp.Client) *HttpApi {
99
}
100
101
return &HttpApi{
102
- url: url,
103
- httpcli: c,
102
+ url: url,
103
+ httpcli: c,
104
applyGlobal: func(*RequestBuilder) {},
105
}
106
}
client/httpapi/api_test.go
+3
-3
@@ -53,7 +53,7 @@ func (NodeProvider) MakeAPISwarm(ctx context.Context, fullIdentity bool, n int)
53
}
54
55
if n > 1 {
56
- connectArgs := []string{"iptb", "--IPTB_ROOT", dir, "connect", fmt.Sprintf("[1-%d]", n - 1), "0"}
56
+ connectArgs := []string{"iptb", "--IPTB_ROOT", dir, "connect", fmt.Sprintf("[1-%d]", n-1), "0"}
57
if err := c.Run(connectArgs); err != nil {
58
return nil, err
59
}
@@ -95,8 +95,8 @@ func (NodeProvider) MakeAPISwarm(ctx context.Context, fullIdentity bool, n int)
95
}
96
c := &gohttp.Client{
97
Transport: &gohttp.Transport{
98
- Proxy: gohttp.ProxyFromEnvironment,
99
- DisableKeepAlives: true,
98
+ Proxy: gohttp.ProxyFromEnvironment,
99
+ DisableKeepAlives: true,
100
DisableCompression: true,
101
},
102
}
client/httpapi/apifile.go
+12
-12
@@ -22,12 +22,12 @@ func (api *UnixfsAPI) Get(ctx context.Context, p iface.Path) (files.Node, error)
22
}
23
}
24
25
- var stat struct{
25
+ var stat struct {
26
Hash string
27
Type string
28
Size int64 // unixfs size
29
}
30
- err := api.core().request("files/stat", p.String()). Exec(ctx, &stat)
30
+ err := api.core().request("files/stat", p.String()).Exec(ctx, &stat)
31
if err != nil {
32
return nil, err
33
}
@@ -43,12 +43,12 @@ func (api *UnixfsAPI) Get(ctx context.Context, p iface.Path) (files.Node, error)
43
}
44
45
type apiFile struct {
46
- ctx context.Context
46
+ ctx context.Context
47
core *HttpApi
48
size int64
49
path iface.Path
50
51
- r io.ReadCloser
51
+ r io.ReadCloser
52
at int64
53
}
54
@@ -96,7 +96,7 @@ func (f *apiFile) Size() (int64, error) {
96
97
func (api *UnixfsAPI) getFile(ctx context.Context, p iface.Path, size int64) (files.Node, error) {
98
f := &apiFile{
99
- ctx: ctx,
99
+ ctx: ctx,
100
core: api.core(),
101
size: size,
102
path: p,
@@ -106,14 +106,14 @@ func (api *UnixfsAPI) getFile(ctx context.Context, p iface.Path, size int64) (fi
106
}
107
108
type apiIter struct {
109
- ctx context.Context
109
+ ctx context.Context
110
core *UnixfsAPI
111
112
err error
113
114
- dec *json.Decoder
114
+ dec *json.Decoder
115
curFile files.Node
116
- cur lsLink
116
+ cur lsLink
117
}
118
119
func (it *apiIter) Err() error {
@@ -179,7 +179,7 @@ func (it *apiIter) Node() files.Node {
179
}
180
181
type apiDir struct {
182
- ctx context.Context
182
+ ctx context.Context
183
core *UnixfsAPI
184
size int64
185
path iface.Path
@@ -197,9 +197,9 @@ func (d *apiDir) Size() (int64, error) {
197
198
func (d *apiDir) Entries() files.DirIterator {
199
return &apiIter{
200
- ctx: d.ctx,
200
+ ctx: d.ctx,
201
core: d.core,
202
- dec: d.dec,
202
+ dec: d.dec,
203
}
204
}
205
@@ -216,7 +216,7 @@ func (api *UnixfsAPI) getDir(ctx context.Context, p iface.Path, size int64) (fil
216
}
217
218
d := &apiDir{
219
- ctx: ctx,
219
+ ctx: ctx,
220
core: api,
221
size: size,
222
path: p,
client/httpapi/block.go
+1
-1
@@ -16,7 +16,7 @@ import (
16
type BlockAPI HttpApi
17
18
type blockStat struct {
19
- Key string
19
+ Key string
20
BSize int `json:"Size"`
21
}
22
client/httpapi/dag.go
+1
-1
@@ -31,7 +31,7 @@ func (api *DagAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPut
31
return nil, fmt.Errorf("setting hash len is not supported yet")
32
}
33
34
- var out struct{
34
+ var out struct {
35
Cid cid.Cid
36
}
37
req := api.core().request("dag/put").
client/httpapi/ipldnode.go
+2
-2
@@ -22,9 +22,9 @@ type ipldNode struct {
22
23
func (a *HttpApi) nodeFromPath(ctx context.Context, p iface.ResolvedPath) ipld.Node {
24
return &ipldNode{
25
- ctx: ctx,
25
+ ctx: ctx,
26
path: p,
27
- api: a,
27
+ api: a,
28
}
29
}
30
client/httpapi/key.go
+2
-3
@@ -13,7 +13,7 @@ type KeyAPI HttpApi
13
14
type keyOutput struct {
15
JName string `json:"Name"`
16
- Id string
16
+ Id string
17
}
18
19
func (k *keyOutput) Name() string {
@@ -35,7 +35,6 @@ func (k *keyOutput) valid() error {
35
return err
36
}
37
38
-
38
func (api *KeyAPI) Generate(ctx context.Context, name string, opts ...caopts.KeyGenerateOption) (iface.Key, error) {
39
options, err := caopts.KeyGenerateOptions(opts...)
40
if err != nil {
@@ -65,7 +64,7 @@ func (api *KeyAPI) List(ctx context.Context) ([]iface.Key, error) {
64
}
65
66
func (api *KeyAPI) Self(ctx context.Context) (iface.Key, error) {
68
- var id struct{ID string}
67
+ var id struct{ ID string }
68
if err := api.core().request("id").Exec(ctx, &id); err != nil {
69
return nil, err
70
}
client/httpapi/object.go
+1
-1
@@ -14,7 +14,7 @@ import (
14
type ObjectAPI HttpApi
15
16
type objectOut struct {
17
- Hash string
17
+ Hash string
18
}
19
20
func (api *ObjectAPI) New(ctx context.Context, opts ...caopts.ObjectNewOption) (format.Node, error) {
client/httpapi/pin.go
+1
-2
@@ -20,7 +20,7 @@ type pinRefKeyList struct {
20
21
type pin struct {
22
path iface.ResolvedPath
23
- typ string
23
+ typ string
24
}
25
26
func (p *pin) Path() iface.ResolvedPath {
@@ -31,7 +31,6 @@ func (p *pin) Type() string {
31
return p.typ
32
}
33
34
-
34
func (api *PinAPI) Add(context.Context, iface.Path, ...caopts.PinAddOption) error {
35
panic("implement me")
36
}
client/httpapi/unixfs.go
+46
-5
@@ -2,9 +2,11 @@ package httpapi
2
3
import (
4
"context"
5
+ "encoding/json"
6
"fmt"
7
"github.com/ipfs/go-cid"
8
"github.com/pkg/errors"
9
+ "io"
10
11
"github.com/ipfs/go-ipfs/core/coreapi/interface"
12
caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
@@ -39,7 +41,6 @@ func (api *UnixfsAPI) Add(ctx context.Context, f files.Node, opts ...caopts.Unix
41
Option("hash", mht).
42
Option("chunker", options.Chunker).
43
Option("cid-version", options.CidVersion).
42
- //Option("", options.Events).
44
Option("fscache", options.FsCache).
45
Option("hidden", options.Hidden).
46
Option("inline", options.Inline).
@@ -47,11 +48,10 @@ func (api *UnixfsAPI) Add(ctx context.Context, f files.Node, opts ...caopts.Unix
48
Option("nocopy", options.NoCopy).
49
Option("only-hash", options.OnlyHash).
50
Option("pin", options.Pin).
50
- //Option("", options.Progress).
51
Option("silent", options.Silent).
52
Option("stdin-name", options.StdinName).
53
Option("wrap-with-directory", options.Wrap).
54
- Option("quieter", true) // TODO: rm after event impl
54
+ Option("progress", options.Progress)
55
56
if options.RawLeavesSet {
57
req.Option("raw-leaves", options.RawLeaves)
@@ -73,9 +73,50 @@ func (api *UnixfsAPI) Add(ctx context.Context, f files.Node, opts ...caopts.Unix
73
}
74
75
var out addEvent
76
- if err := req.Exec(ctx, &out); err != nil { //TODO: ndjson events
76
+ resp, err := req.Send(ctx)
77
+ if err != nil {
78
return nil, err
79
}
80
+ if resp.Error != nil {
81
+ return nil, resp.Error
82
+ }
83
+ defer resp.Output.Close()
84
+ dec := json.NewDecoder(resp.Output)
85
+loop:
86
+ for {
87
+ var evt addEvent
88
+ switch err := dec.Decode(&evt); err {
89
+ case nil:
90
+ case io.EOF:
91
+ break loop
92
+ default:
93
+ return nil, err
94
+ }
95
+ out = evt
96
+
97
+ if options.Events != nil {
98
+ ifevt := &iface.AddEvent{
99
+ Name: out.Name,
100
+ Size: out.Size,
101
+ Bytes: out.Bytes,
102
+ }
103
+
104
+ if out.Hash != "" {
105
+ c, err := cid.Parse(out.Hash)
106
+ if err != nil {
107
+ return nil, err
108
+ }
109
+
110
+ ifevt.Path = iface.IpfsPath(c)
111
+ }
112
+
113
+ select {
114
+ case options.Events <- ifevt:
115
+ case <-ctx.Done():
116
+ return nil, ctx.Err()
117
+ }
118
+ }
119
+ }
120
121
c, err := cid.Parse(out.Hash)
122
if err != nil {
@@ -120,7 +161,7 @@ func (api *UnixfsAPI) Ls(ctx context.Context, p iface.Path) ([]*format.Link, err
161
links[i] = &format.Link{
162
Name: l.Name,
163
Size: l.Size,
123
- Cid: c,
164
+ Cid: c,
165
}
166
}
167
return links, nil