master
go 368 lines 9.2 KB
Raw
1 package commands
2
3 import (
4 gotar "archive/tar"
5 "bufio"
6 "compress/gzip"
7 "errors"
8 "fmt"
9 "io"
10 "os"
11 gopath "path"
12 "path/filepath"
13 "strings"
14
15 "github.com/ipfs/kubo/core/commands/cmdenv"
16 "github.com/ipfs/kubo/core/commands/cmdutils"
17 "github.com/ipfs/kubo/core/commands/e"
18
19 "github.com/cheggaaa/pb/v3"
20 "github.com/ipfs/boxo/files"
21 "github.com/ipfs/boxo/tar"
22 cmds "github.com/ipfs/go-ipfs-cmds"
23 )
24
25 var ErrInvalidCompressionLevel = errors.New("compression level must be between 1 and 9")
26
27 const (
28 outputOptionName = "output"
29 archiveOptionName = "archive"
30 compressOptionName = "compress"
31 compressionLevelOptionName = "compression-level"
32 )
33
34 var GetCmd = &cmds.Command{
35 Helptext: cmds.HelpText{
36 Tagline: "Download IPFS objects.",
37 ShortDescription: `
38 Stores to disk the data contained an IPFS or IPNS object(s) at the given path.
39
40 By default, the output will be stored at './<ipfs-path>', but an alternate
41 path can be specified with '--output=<path>' or '-o=<path>'.
42
43 To output a TAR archive instead of unpacked files, use '--archive' or '-a'.
44
45 To compress the output with GZIP compression, use '--compress' or '-C'. You
46 may also specify the level of compression by specifying '-l=<1-9>'.
47 `,
48 HTTP: &cmds.HTTPHelpText{
49 ResponseContentType: "application/x-tar, or application/gzip when compress=true",
50 },
51 },
52
53 Arguments: []cmds.Argument{
54 cmds.StringArg("ipfs-path", true, false, "The path to the IPFS object(s) to be outputted.").EnableStdin(),
55 },
56 Options: []cmds.Option{
57 cmds.StringOption(outputOptionName, "o", "The path where the output should be stored."),
58 cmds.BoolOption(archiveOptionName, "a", "Output a TAR archive."),
59 cmds.BoolOption(compressOptionName, "C", "Compress the output with GZIP compression."),
60 cmds.IntOption(compressionLevelOptionName, "l", "The level of compression (1-9)."),
61 cmds.BoolOption(progressOptionName, "p", "Stream progress data. Defaults to true when stderr is a terminal."),
62 },
63 PreRun: func(req *cmds.Request, env cmds.Environment) error {
64 _, err := getCompressOptions(req)
65 return err
66 },
67 Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
68 ctx := req.Context
69 cmplvl, err := getCompressOptions(req)
70 if err != nil {
71 return err
72 }
73
74 api, err := cmdenv.GetApi(env, req)
75 if err != nil {
76 return err
77 }
78
79 p, err := cmdutils.PathOrCidPath(req.Arguments[0])
80 if err != nil {
81 return err
82 }
83
84 file, err := api.Unixfs().Get(ctx, p)
85 if err != nil {
86 return err
87 }
88
89 size, err := file.Size()
90 if err != nil {
91 return err
92 }
93
94 res.SetLength(uint64(size))
95
96 archive, _ := req.Options[archiveOptionName].(bool)
97 reader, err := fileArchive(file, p.String(), archive, cmplvl)
98 if err != nil {
99 return err
100 }
101 go func() {
102 // We cannot defer a close in the response writer (like we should)
103 // Because the cmd framework outsmart us and doesn't call response
104 // if the context is over.
105 <-ctx.Done()
106 reader.Close()
107 }()
108
109 // Set Content-Type based on output format.
110 // When compression is enabled, output is gzip (or tar.gz for directories).
111 // Otherwise, tar is used as the transport format.
112 res.SetEncodingType(cmds.OctetStream)
113 if cmplvl != gzip.NoCompression {
114 res.SetContentType("application/gzip")
115 } else {
116 res.SetContentType("application/x-tar")
117 }
118
119 return res.Emit(reader)
120 },
121 PostRun: cmds.PostRunMap{
122 cmds.CLI: func(res cmds.Response, re cmds.ResponseEmitter) error {
123 req := res.Request()
124
125 v, err := res.Next()
126 if err != nil {
127 return err
128 }
129
130 outReader, ok := v.(io.Reader)
131 if !ok {
132 return e.New(e.TypeErr(outReader, v))
133 }
134
135 outPath := getOutPath(req)
136
137 cmplvl, err := getCompressOptions(req)
138 if err != nil {
139 return err
140 }
141
142 archive, _ := req.Options[archiveOptionName].(bool)
143 showProgress := cmdenv.ShouldShowProgress(req, progressOptionName)
144
145 gw := getWriter{
146 Out: os.Stdout,
147 Err: os.Stderr,
148 Archive: archive,
149 Compression: cmplvl,
150 Size: int64(res.Length()),
151 Progress: showProgress,
152 }
153
154 return gw.Write(outReader, outPath)
155 },
156 },
157 }
158
159 type clearlineReader struct {
160 io.Reader
161 out io.Writer
162 }
163
164 func (r *clearlineReader) Read(p []byte) (n int, err error) {
165 n, err = r.Reader.Read(p)
166 if err == io.EOF {
167 // callback
168 fmt.Fprintf(r.out, "\033[2K\r") // clear progress bar line on EOF
169 }
170 return
171 }
172
173 func progressBarForReader(out io.Writer, r io.Reader, l int64) (*pb.ProgressBar, io.Reader) {
174 bar := makeProgressBar(out, l)
175 barR := bar.NewProxyReader(r)
176 return bar, &clearlineReader{barR, out}
177 }
178
179 func makeProgressBar(out io.Writer, l int64) *pb.ProgressBar {
180 return pb.New64(l).Set(pb.Bytes, true).SetTemplateString(cmdenv.ProgressBarFullTemplate).SetWriter(out)
181 }
182
183 func getOutPath(req *cmds.Request) string {
184 outPath, _ := req.Options[outputOptionName].(string)
185 if outPath == "" {
186 trimmed := strings.TrimRight(req.Arguments[0], "/")
187 _, outPath = filepath.Split(trimmed)
188 outPath = filepath.Clean(outPath)
189 }
190 return outPath
191 }
192
193 type getWriter struct {
194 Out io.Writer // for output to user
195 Err io.Writer // for progress bar output
196
197 Archive bool
198 Compression int
199 Size int64
200 Progress bool
201 }
202
203 func (gw *getWriter) Write(r io.Reader, fpath string) error {
204 if gw.Archive || gw.Compression != gzip.NoCompression {
205 return gw.writeArchive(r, fpath)
206 }
207 return gw.writeExtracted(r, fpath)
208 }
209
210 func (gw *getWriter) writeArchive(r io.Reader, fpath string) error {
211 // adjust file name if tar
212 if gw.Archive {
213 if !strings.HasSuffix(fpath, ".tar") && !strings.HasSuffix(fpath, ".tar.gz") {
214 fpath += ".tar"
215 }
216 }
217
218 // adjust file name if gz
219 if gw.Compression != gzip.NoCompression {
220 if !strings.HasSuffix(fpath, ".gz") {
221 fpath += ".gz"
222 }
223 }
224
225 // create file
226 file, err := os.Create(fpath)
227 if err != nil {
228 return err
229 }
230 defer file.Close()
231
232 fmt.Fprintf(gw.Out, "Saving archive to %s\n", fpath)
233 if gw.Progress {
234 var bar *pb.ProgressBar
235 bar, r = progressBarForReader(gw.Err, r, gw.Size)
236 bar.Start()
237 defer bar.Finish()
238 }
239
240 _, err = io.Copy(file, r)
241 return err
242 }
243
244 func (gw *getWriter) writeExtracted(r io.Reader, fpath string) error {
245 fmt.Fprintf(gw.Out, "Saving file(s) to %s\n", fpath)
246 var progressCb func(int64) int64
247 if gw.Progress {
248 bar := makeProgressBar(gw.Err, gw.Size)
249 bar.Start()
250 defer bar.Finish()
251 defer bar.SetCurrent(gw.Size)
252 progressCb = func(n int64) int64 { bar.Add64(n); return bar.Current() }
253 }
254
255 extractor := &tar.Extractor{Path: fpath, Progress: progressCb}
256 return extractor.Extract(r)
257 }
258
259 func getCompressOptions(req *cmds.Request) (int, error) {
260 cmprs, _ := req.Options[compressOptionName].(bool)
261 cmplvl, cmplvlFound := req.Options[compressionLevelOptionName].(int)
262 switch {
263 case !cmprs:
264 return gzip.NoCompression, nil
265 case cmprs && !cmplvlFound:
266 return gzip.DefaultCompression, nil
267 case cmprs && (cmplvl < 1 || cmplvl > 9):
268 return gzip.NoCompression, ErrInvalidCompressionLevel
269 }
270 return cmplvl, nil
271 }
272
273 // DefaultBufSize is the buffer size for gets. for now, 1MiB, which is ~4 blocks.
274 // TODO: does this need to be configurable?
275 var DefaultBufSize = 1048576
276
277 type identityWriteCloser struct {
278 w io.Writer
279 }
280
281 func (i *identityWriteCloser) Write(p []byte) (int, error) {
282 return i.w.Write(p)
283 }
284
285 func (i *identityWriteCloser) Close() error {
286 return nil
287 }
288
289 func fileArchive(f files.Node, name string, archive bool, compression int) (io.ReadCloser, error) {
290 cleaned := gopath.Clean(name)
291 _, filename := gopath.Split(cleaned)
292
293 // need to connect a writer to a reader
294 piper, pipew := io.Pipe()
295 checkErrAndClosePipe := func(err error) bool {
296 if err != nil {
297 _ = pipew.CloseWithError(err)
298 return true
299 }
300 return false
301 }
302
303 // use a buffered writer to parallelize task
304 bufw := bufio.NewWriterSize(pipew, DefaultBufSize)
305
306 // compression determines whether to use gzip compression.
307 maybeGzw, err := newMaybeGzWriter(bufw, compression)
308 if checkErrAndClosePipe(err) {
309 return nil, err
310 }
311
312 closeGzwAndPipe := func() {
313 if err := maybeGzw.Close(); checkErrAndClosePipe(err) {
314 return
315 }
316 if err := bufw.Flush(); checkErrAndClosePipe(err) {
317 return
318 }
319 pipew.Close() // everything seems to be ok.
320 }
321
322 if !archive && compression != gzip.NoCompression {
323 // the case when the node is a file
324 r := files.ToFile(f)
325 if r == nil {
326 return nil, errors.New("file is not regular")
327 }
328
329 go func() {
330 if _, err := io.Copy(maybeGzw, r); checkErrAndClosePipe(err) {
331 return
332 }
333 closeGzwAndPipe() // everything seems to be ok
334 }()
335 } else {
336 // the case for 1. archive, and 2. not archived and not compressed, in
337 // which tar is used anyway as a transport format
338
339 // construct the tar writer
340 w, err := files.NewTarWriter(maybeGzw)
341 if checkErrAndClosePipe(err) {
342 return nil, err
343 }
344
345 // if not creating an archive set the format to PAX in order to preserve nanoseconds
346 if !archive {
347 w.SetFormat(gotar.FormatPAX)
348 }
349
350 go func() {
351 // write all the nodes recursively
352 if err := w.WriteFile(f, filename); checkErrAndClosePipe(err) {
353 return
354 }
355 w.Close() // close tar writer
356 closeGzwAndPipe() // everything seems to be ok
357 }()
358 }
359
360 return piper, nil
361 }
362
363 func newMaybeGzWriter(w io.Writer, compression int) (io.WriteCloser, error) {
364 if compression != gzip.NoCompression {
365 return gzip.NewWriterLevel(w, compression)
366 }
367 return &identityWriteCloser{w}, nil
368 }