Commit 291d62bb authored by rht's avatar rht

Decompose maybeGzwriter

License: MIT
Signed-off-by: default avatarrht <rhtbot@gmail.com>
parent 78d19b2f
...@@ -36,67 +36,71 @@ func DagArchive(ctx cxt.Context, nd *mdag.Node, name string, dag mdag.DAGService ...@@ -36,67 +36,71 @@ func DagArchive(ctx cxt.Context, nd *mdag.Node, name string, dag mdag.DAGService
// need to connect a writer to a reader // need to connect a writer to a reader
piper, pipew := io.Pipe() piper, pipew := io.Pipe()
checkErrAndClosePipe := func(err error) bool {
if err != nil {
pipew.CloseWithError(err)
return true
}
return false
}
// use a buffered writer to parallelize task // use a buffered writer to parallelize task
bufw := bufio.NewWriterSize(pipew, DefaultBufSize) bufw := bufio.NewWriterSize(pipew, DefaultBufSize)
// compression determines whether to use gzip compression. // compression determines whether to use gzip compression.
var maybeGzw io.WriteCloser maybeGzw, err := newMaybeGzWriter(bufw, compression)
var err error if checkErrAndClosePipe(err) {
if compression != gzip.NoCompression { return nil, err
maybeGzw, err = gzip.NewWriterLevel(bufw, compression) }
if err != nil {
pipew.CloseWithError(err) closeGzwAndPipe := func() {
return nil, err if err := maybeGzw.Close(); checkErrAndClosePipe(err) {
return
} }
} else { if err := bufw.Flush(); checkErrAndClosePipe(err) {
maybeGzw = &identityWriteCloser{bufw} return
}
pipew.Close() // everything seems to be ok.
} }
if !archive && compression != gzip.NoCompression { if !archive && compression != gzip.NoCompression {
// the case when the node is a file // the case when the node is a file
dagr, err := uio.NewDagReader(ctx, nd, dag) dagr, err := uio.NewDagReader(ctx, nd, dag)
if err != nil { if checkErrAndClosePipe(err) {
pipew.CloseWithError(err)
return nil, err return nil, err
} }
go func() { go func() {
if _, err := dagr.WriteTo(maybeGzw); err != nil { if _, err := dagr.WriteTo(maybeGzw); checkErrAndClosePipe(err) {
pipew.CloseWithError(err)
return
}
maybeGzw.Close()
if err := bufw.Flush(); err != nil {
pipew.CloseWithError(err)
return return
} }
pipew.Close() // everything seems to be ok. closeGzwAndPipe() // everything seems to be ok
}() }()
} else { } else {
// the case for 1. archive, and 2. not archived and not compressed, in which tar is used anyway as a transport format // the case for 1. archive, and 2. not archived and not compressed, in which tar is used anyway as a transport format
// construct the tar writer // construct the tar writer
w, err := tar.NewWriter(ctx, dag, archive, compression, maybeGzw) w, err := tar.NewWriter(ctx, dag, archive, compression, maybeGzw)
if err != nil { if checkErrAndClosePipe(err) {
return nil, err return nil, err
} }
go func() { go func() {
// write all the nodes recursively // write all the nodes recursively
if err := w.WriteNode(nd, filename); err != nil { if err := w.WriteNode(nd, filename); checkErrAndClosePipe(err) {
pipew.CloseWithError(err)
return return
} }
w.Close() w.Close() // close tar writer
maybeGzw.Close() closeGzwAndPipe() // everything seems to be ok
if err := bufw.Flush(); err != nil {
pipew.CloseWithError(err)
return
}
pipew.Close() // everything seems to be ok.
}() }()
} }
return piper, nil return piper, nil
} }
func newMaybeGzWriter(w io.Writer, compression int) (io.WriteCloser, error) {
if compression != gzip.NoCompression {
return gzip.NewWriterLevel(w, compression)
}
return &identityWriteCloser{w}, nil
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment