-
-
Notifications
You must be signed in to change notification settings - Fork 3k
/
export.go
155 lines (131 loc) · 3.48 KB
/
export.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
package dagcmd
import (
"errors"
"fmt"
"io"
"os"
"time"
"github.com/cheggaaa/pb"
cid "github.com/ipfs/go-cid"
"github.com/ipfs/go-ipfs/core/commands/cmdenv"
ipld "github.com/ipfs/go-ipld-format"
mdag "github.com/ipfs/go-merkledag"
cmds "github.com/ipfs/go-ipfs-cmds"
gocar "github.com/ipld/go-car"
)
func dagExport(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
c, err := cid.Decode(req.Arguments[0])
if err != nil {
return fmt.Errorf(
"unable to parse root specification (currently only bare CIDs are supported): %s",
err,
)
}
api, err := cmdenv.GetApi(env, req)
if err != nil {
return err
}
// Code disabled until descent-issue in go-ipld-prime is fixed
// https://github.com/ribasushi/gip-muddle-up
//
// sb := gipselectorbuilder.NewSelectorSpecBuilder(gipfree.NodeBuilder())
// car := gocar.NewSelectiveCar(
// req.Context,
// <needs to be fixed to take format.NodeGetter as well>,
// []gocar.Dag{gocar.Dag{
// Root: c,
// Selector: sb.ExploreRecursive(
// gipselector.RecursionLimitNone(),
// sb.ExploreAll(sb.ExploreRecursiveEdge()),
// ).Node(),
// }},
// )
// ...
// if err := car.Write(pipeW); err != nil {}
pipeR, pipeW := io.Pipe()
errCh := make(chan error, 2) // we only report the 1st error
go func() {
defer func() {
if err := pipeW.Close(); err != nil {
errCh <- fmt.Errorf("stream flush failed: %s", err)
}
close(errCh)
}()
if err := gocar.WriteCar(
req.Context,
mdag.NewSession(
req.Context,
api.Dag(),
),
[]cid.Cid{c},
pipeW,
); err != nil {
errCh <- err
}
}()
if err := res.Emit(pipeR); err != nil {
pipeR.Close() // ignore the error if any
return err
}
err = <-errCh
// minimal user friendliness
if err != nil &&
err == ipld.ErrNotFound {
explicitOffline, _ := req.Options["offline"].(bool)
if explicitOffline {
err = fmt.Errorf("%s (currently offline, perhaps retry without the offline flag)", err)
} else {
node, envErr := cmdenv.GetNode(env)
if envErr == nil && !node.IsOnline {
err = fmt.Errorf("%s (currently offline, perhaps retry after attaching to the network)", err)
}
}
}
return err
}
func finishCLIExport(res cmds.Response, re cmds.ResponseEmitter) error {
var showProgress bool
val, specified := res.Request().Options[progressOptionName]
if !specified {
// default based on TTY availability
errStat, _ := os.Stderr.Stat()
if 0 != (errStat.Mode() & os.ModeCharDevice) {
showProgress = true
}
} else if val.(bool) {
showProgress = true
}
// simple passthrough, no progress
if !showProgress {
return cmds.Copy(re, res)
}
bar := pb.New64(0).SetUnits(pb.U_BYTES)
bar.Output = os.Stderr
bar.ShowSpeed = true
bar.ShowElapsedTime = true
bar.RefreshRate = 500 * time.Millisecond
bar.Start()
var processedOneResponse bool
for {
v, err := res.Next()
if err == io.EOF {
// We only write the final bar update on success
// On error it looks too weird
bar.Finish()
return re.Close()
} else if err != nil {
return re.CloseWithError(err)
} else if processedOneResponse {
return re.CloseWithError(errors.New("unexpected multipart response during emit, please file a bugreport"))
}
r, ok := v.(io.Reader)
if !ok {
// some sort of encoded response, this should not be happening
return errors.New("unexpected non-stream passed to PostRun: please file a bugreport")
}
processedOneResponse = true
if err := re.Emit(bar.NewProxyReader(r)); err != nil {
return err
}
}
}