commands/get: move FileArchive here
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Jan 30, 2019 at 17:20 UTC
e71e700cc48d0c1842c5c52b5aa903f18dbe0d44
1 file changed
+95
-2
core/commands/get.go
+95
-2
@@ -1,11 +1,13 @@
1
package commands
2
3
import (
4
+ "bufio"
5
"compress/gzip"
6
"errors"
7
"fmt"
8
"io"
9
"os"
10
+ "path"
11
"path/filepath"
12
"strings"
13
@@ -14,8 +16,8 @@ import (
16
"github.com/ipfs/go-ipfs/core/coreapi/interface"
17
18
"gx/ipfs/QmQine7gvHncNevKtG9QXxf3nXcwSj6aDDmMm52mHofEEp/tar-utils"
17
- uarchive "gx/ipfs/QmSMJ4rZbCJaih3y82Ebq7BZqK6vU2FHsKcWKQiE1DPTpS/go-unixfs/archive"
19
"gx/ipfs/QmWGm4AbZEbnmdgVTza52MSNpEmBdFVqzmAysRbjrRyGbH/go-ipfs-cmds"
20
+ "gx/ipfs/QmXWZCd8jfaHmt4UDSnjKmGcrQMw95bDGWqEeVLVJjoANX/go-ipfs-files"
21
"gx/ipfs/QmYWB8oH6o7qftxoyqTTZhzLrhKCVT7NYahECQTwTtqbgj/pb"
22
"gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
23
)
@@ -87,7 +89,7 @@ may also specify the level of compression by specifying '-l=<1-9>'.
89
res.SetLength(uint64(size))
90
91
archive, _ := req.Options[archiveOptionName].(bool)
90
- reader, err := uarchive.FileArchive(file, p.String(), archive, cmplvl)
92
+ reader, err := fileArchive(file, p.String(), archive, cmplvl)
93
if err != nil {
94
return err
95
}
@@ -247,3 +249,94 @@ func getCompressOptions(req *cmds.Request) (int, error) {
249
}
250
return cmplvl, nil
251
}
252
+
253
+// DefaultBufSize is the buffer size for gets. for now, 1MB, which is ~4 blocks.
254
+// TODO: does this need to be configurable?
255
+var DefaultBufSize = 1048576
256
+
257
+type identityWriteCloser struct {
258
+ w io.Writer
259
+}
260
+
261
+func (i *identityWriteCloser) Write(p []byte) (int, error) {
262
+ return i.w.Write(p)
263
+}
264
+
265
+func (i *identityWriteCloser) Close() error {
266
+ return nil
267
+}
268
+
269
+func fileArchive(f files.Node, name string, archive bool, compression int) (io.Reader, error) {
270
+ cleaned := path.Clean(name)
271
+ _, filename := path.Split(cleaned)
272
+
273
+ // need to connect a writer to a reader
274
+ piper, pipew := io.Pipe()
275
+ checkErrAndClosePipe := func(err error) bool {
276
+ if err != nil {
277
+ pipew.CloseWithError(err)
278
+ return true
279
+ }
280
+ return false
281
+ }
282
+
283
+ // use a buffered writer to parallelize task
284
+ bufw := bufio.NewWriterSize(pipew, DefaultBufSize)
285
+
286
+ // compression determines whether to use gzip compression.
287
+ maybeGzw, err := newMaybeGzWriter(bufw, compression)
288
+ if checkErrAndClosePipe(err) {
289
+ return nil, err
290
+ }
291
+
292
+ closeGzwAndPipe := func() {
293
+ if err := maybeGzw.Close(); checkErrAndClosePipe(err) {
294
+ return
295
+ }
296
+ if err := bufw.Flush(); checkErrAndClosePipe(err) {
297
+ return
298
+ }
299
+ pipew.Close() // everything seems to be ok.
300
+ }
301
+
302
+ if !archive && compression != gzip.NoCompression {
303
+ // the case when the node is a file
304
+ r := files.ToFile(f)
305
+ if r == nil {
306
+ return nil, errors.New("file is not regular")
307
+ }
308
+
309
+ go func() {
310
+ if _, err := io.Copy(maybeGzw, r); checkErrAndClosePipe(err) {
311
+ return
312
+ }
313
+ closeGzwAndPipe() // everything seems to be ok
314
+ }()
315
+ } else {
316
+ // the case for 1. archive, and 2. not archived and not compressed, in which tar is used anyway as a transport format
317
+
318
+ // construct the tar writer
319
+ w, err := files.NewTarWriter(maybeGzw)
320
+ if checkErrAndClosePipe(err) {
321
+ return nil, err
322
+ }
323
+
324
+ go func() {
325
+ // write all the nodes recursively
326
+ if err := w.WriteFile(f, filename); checkErrAndClosePipe(err) {
327
+ return
328
+ }
329
+ w.Close() // close tar writer
330
+ closeGzwAndPipe() // everything seems to be ok
331
+ }()
332
+ }
333
+
334
+ return piper, nil
335
+}
336
+
337
+func newMaybeGzWriter(w io.Writer, compression int) (io.WriteCloser, error) {
338
+ if compression != gzip.NoCompression {
339
+ return gzip.NewWriterLevel(w, compression)
340
+ }
341
+ return &identityWriteCloser{w}, nil
342
+}