Merge pull request #369 from ncw/fix-async

Rework ArchiveAsync interface to make it return any archiving errors

formats.go

@@ -235,6 +235,23 @@ func (caf CompressedArchive) Archive(ctx context.Context, output io.Writer, file
 	return caf.Archival.Archive(ctx, output, files)
 }
 
+// ArchiveAsync adds files to the output archive while compressing the result asynchronously.
+func (caf CompressedArchive) ArchiveAsync(ctx context.Context, output io.Writer, jobs <-chan ArchiveAsyncJob) error {
+	do, ok := caf.Archival.(ArchiverAsync)
+	if !ok {
+		return fmt.Errorf("%s archive does not support async writing", caf.Name())
+	}
+	if caf.Compression != nil {
+		wc, err := caf.Compression.OpenWriter(output)
+		if err != nil {
+			return err
+		}
+		defer wc.Close()
+		output = wc
+	}
+	return do.ArchiveAsync(ctx, output, jobs)
+}
+
 // Extract reads files out of an archive while decompressing the results.
 func (caf CompressedArchive) Extract(ctx context.Context, sourceArchive io.Reader, pathsInArchive []string, handleFile FileHandler) error {
 	if caf.Compression != nil {
@@ -358,7 +375,8 @@ var formats = make(map[string]Format)
 
 // Interface guards
 var (
-	_ Format    = (*CompressedArchive)(nil)
-	_ Archiver  = (*CompressedArchive)(nil)
-	_ Extractor = (*CompressedArchive)(nil)
+	_ Format        = (*CompressedArchive)(nil)
+	_ Archiver      = (*CompressedArchive)(nil)
+	_ ArchiverAsync = (*CompressedArchive)(nil)
+	_ Extractor     = (*CompressedArchive)(nil)
 )

interfaces.go

@@ -60,16 +60,25 @@ type Archiver interface {
 	Archive(ctx context.Context, output io.Writer, files []File) error
 }
 
+// ArchiveAsyncJob contains a File to be archived and a channel that
+// the result of the archiving should be returned on.
+type ArchiveAsyncJob struct {
+	File   File
+	Result chan<- error
+}
+
 // ArchiverAsync is an Archiver that can also create archives
 // asynchronously by pumping files into a channel as they are
 // discovered.
 type ArchiverAsync interface {
 	Archiver
 
 	// Use ArchiveAsync if you can't pre-assemble a list of all
-	// the files for the archive. Close the files channel after
+	// the files for the archive. Close the jobs channel after
 	// all the files have been sent.
-	ArchiveAsync(ctx context.Context, output io.Writer, files <-chan File) error
+	//
+	// This won't return until the channel is closed.
+	ArchiveAsync(ctx context.Context, output io.Writer, jobs <-chan ArchiveAsyncJob) error
 }
 
 // Extractor can extract files from an archive.

tar.go

@@ -60,18 +60,12 @@ func (t Tar) Archive(ctx context.Context, output io.Writer, files []File) error
 	return nil
 }
 
-func (t Tar) ArchiveAsync(ctx context.Context, output io.Writer, files <-chan File) error {
+func (t Tar) ArchiveAsync(ctx context.Context, output io.Writer, jobs <-chan ArchiveAsyncJob) error {
 	tw := tar.NewWriter(output)
 	defer tw.Close()
 
-	for file := range files {
-		if err := t.writeFileToArchive(ctx, tw, file); err != nil {
-			if t.ContinueOnError && ctx.Err() == nil { // context errors should always abort
-				log.Printf("[ERROR] %v", err)
-				continue
-			}
-			return err
-		}
+	for job := range jobs {
+		job.Result <- t.writeFileToArchive(ctx, tw, job.File)
 	}
 
 	return nil
@@ -234,7 +228,8 @@ func (t Tar) Extract(ctx context.Context, sourceArchive io.Reader, pathsInArchiv
 
 // Interface guards
 var (
-	_ Archiver  = (*Tar)(nil)
-	_ Extractor = (*Tar)(nil)
-	_ Inserter  = (*Tar)(nil)
+	_ Archiver      = (*Tar)(nil)
+	_ ArchiverAsync = (*Tar)(nil)
+	_ Extractor     = (*Tar)(nil)
+	_ Inserter      = (*Tar)(nil)
 )

zip.go

@@ -114,19 +114,13 @@ func (z Zip) Archive(ctx context.Context, output io.Writer, files []File) error
 	return nil
 }
 
-func (z Zip) ArchiveAsync(ctx context.Context, output io.Writer, files <-chan File) error {
+func (z Zip) ArchiveAsync(ctx context.Context, output io.Writer, jobs <-chan ArchiveAsyncJob) error {
 	zw := zip.NewWriter(output)
 	defer zw.Close()
 
 	var i int
-	for file := range files {
-		if err := z.archiveOneFile(ctx, zw, i, file); err != nil {
-			if z.ContinueOnError && ctx.Err() == nil { // context errors should always abort
-				log.Printf("[ERROR] %v", err)
-				continue
-			}
-			return err
-		}
+	for job := range jobs {
+		job.Result <- z.archiveOneFile(ctx, zw, i, job.File)
 		i++
 	}
 
@@ -202,7 +196,7 @@ func (z Zip) Extract(ctx context.Context, sourceArchive io.Reader, pathsInArchiv
 	skipDirs := skipList{}
 
 	for i, f := range zr.File {
-		f := f		// make a copy for the Open closure
+		f := f // make a copy for the Open closure
 		if err := ctx.Err(); err != nil {
 			return err // honor context cancellation
 		}
@@ -383,3 +377,10 @@ func decodeText(input, charset string) (string, error) {
 }
 
 var zipHeader = []byte("PK\x03\x04") // NOTE: headers of empty zip files might end with 0x05,0x06 or 0x06,0x06 instead of 0x03,0x04
+
+// Interface guards
+var (
+	_ Archiver      = Zip{}
+	_ ArchiverAsync = Zip{}
+	_ Extractor     = Zip{}
+)