@@ -1,9 +1,14 @@

 package imageprogress

 import (

+	"bytes"

 	"encoding/json"

+	"errors"

+	"fmt"

 	"io"

 	"regexp"

+	"strings"

+	"sync"

 	"time"

 )

@@ -17,6 +22,7 @@ type progressLine struct {

 	ID     string          `json:"id"`

 	Status string          `json:"status"`

 	Detail *progressDetail `json:"progressDetail"`

+	Error  string          `json:"error"`

 }

 // progressDetail is the progressDetail structure in a Docker pull progress line

@@ -94,33 +100,47 @@ func (r report) totalCount() int {

 	return cnt

 }

+// String is used for test output

+func (r report) String() string {

+	result := &bytes.Buffer{}

+	fmt.Fprintf(result, "{")

+	for k := range r {

+		var status string

+		switch k {

+		case statusPending:

+			status = "pending"

+		case statusDownloading:

+			status = "downloading"

+		case statusExtracting:

+			status = "extracting"

+		case statusComplete:

+			status = "complete"

+		}

+		fmt.Fprintf(result, "%s:{Count: %d, Current: %d, Total: %d}, ", status, r[k].Count, r[k].Current, r[k].Total)

+	}

+	fmt.Fprintf(result, "}")

+	return result.String()

+}

+

 // newWriter creates a writer that periodically reports

 // on pull/push progress of a Docker image. It only reports when the state of the

 // different layers has changed and uses time thresholds to limit the

 // rate of the reports.

 func newWriter(reportFn func(report), layersChangedFn func(report, report) bool) io.Writer {

-	pipeIn, pipeOut := io.Pipe()

 	writer := &imageProgressWriter{

-		Writer:                 pipeOut,

-		decoder:                json.NewDecoder(pipeIn),

+		mutex:                  &sync.Mutex{},

 		layerStatus:            map[string]progressLine{},

 		reportFn:               reportFn,

 		layersChangedFn:        layersChangedFn,

 		progressTimeThreshhold: defaultProgressTimeThreshhold,

 		stableThreshhold:       defaultStableThreshhold,

 	}

-	go func() {

-		err := writer.readProgress()

-		if err != nil {

-			pipeIn.CloseWithError(err)

-		}

-	}()

 	return writer

 }

 type imageProgressWriter struct {

-	io.Writer

-	decoder                *json.Decoder

+	mutex                  *sync.Mutex

+	internalWriter         io.Writer

 	layerStatus            map[string]progressLine

 	lastLayerCount         int

 	stableLines            int

@@ -132,10 +152,32 @@ type imageProgressWriter struct {

 	layersChangedFn        func(report, report) bool

 }

-func (w *imageProgressWriter) readProgress() error {

+func (w *imageProgressWriter) ReadFrom(reader io.Reader) (int64, error) {

+	decoder := json.NewDecoder(reader)

+	return 0, w.readProgress(decoder)

+}

+

+func (w *imageProgressWriter) Write(data []byte) (int, error) {

+	w.mutex.Lock()

+	defer w.mutex.Unlock()

+	if w.internalWriter == nil {

+		var pipeIn *io.PipeReader

+		pipeIn, w.internalWriter = io.Pipe()

+		decoder := json.NewDecoder(pipeIn)

+		go func() {

+			err := w.readProgress(decoder)

+			if err != nil {

+				pipeIn.CloseWithError(err)

+			}

+		}()

+	}

+	return w.internalWriter.Write(data)

+}

+

+func (w *imageProgressWriter) readProgress(decoder *json.Decoder) error {

 	for {

 		line := &progressLine{}

-		err := w.decoder.Decode(line)

+		err := decoder.Decode(line)

 		if err == io.EOF {

 			break

 		}

@@ -151,6 +193,11 @@ func (w *imageProgressWriter) readProgress() error {

 }

 func (w *imageProgressWriter) processLine(line *progressLine) error {

+

+	if err := getError(line); err != nil {

+		return err

+	}

+

 	// determine if it's a line we want to process

 	if !islayerStatus(line) {

 		return nil

@@ -212,9 +259,20 @@ func islayerStatus(line *progressLine) bool {

 	if !layerIDRegexp.MatchString(line.ID) {

 		return false

 	}

+	// ignore retrying status

+	if strings.HasPrefix(line.Status, "Retrying") {

+		return false

+	}

 	return true

 }

+func getError(line *progressLine) error {

+	if len(line.Error) > 0 {

+		return errors.New(line.Error)

+	}

+	return nil

+}

+

 func createReport(dockerProgress map[string]progressLine) report {

 	r := report{}

 	for _, line := range dockerProgress {

new file mode 100644

@@ -0,0 +1,183 @@

+package imageprogress

+

+import (

+	"encoding/json"

+	"io"

+	"reflect"

+	"testing"

+)

+

+func TestReports(t *testing.T) {

+	tests := []struct {

+		name        string

+		gen         func(*progressGenerator)

+		errExpected bool

+		expected    report

+	}{

+		{

+			name: "simple report",

+			gen: func(p *progressGenerator) {

+				p.status("1", "Extracting")

+				p.status("2", "Downloading")

+				p.status("1", "Downloading")

+				p.status("2", "Pull complete")

+			},

+			expected: report{

+				statusDownloading: &layerDetail{Count: 1},

+				statusComplete:    &layerDetail{Count: 1},

+			},

+		},

+		{

+			name: "ignore invalid layer id",

+			gen: func(p *progressGenerator) {

+				p.status("1", "Downloading")

+				p.status("hello", "testing")

+				p.status("1", "Downloading")

+			},

+			expected: report{

+				statusDownloading: &layerDetail{Count: 1},

+			},

+		},

+		{

+			name: "ignore retrying status",

+			gen: func(p *progressGenerator) {

+				p.status("1", "Downloading")

+				p.status("2", "Pull complete")

+				p.status("1", "Downloading")

+				p.status("3", "Retrying")

+			},

+			expected: report{

+				statusDownloading: &layerDetail{Count: 1},

+				statusComplete:    &layerDetail{Count: 1},

+			},

+		},

+		{

+			name: "detect error",

+			gen: func(p *progressGenerator) {

+				p.status("1", "Downloading")

+				p.err("an error")

+			},

+			errExpected: true,

+		},

+	}

+

+	for _, test := range tests {

+		pipeIn, pipeOut := io.Pipe()

+		go func() {

+			p := newProgressGenerator(pipeOut)

+			test.gen(p)

+			pipeOut.Close()

+		}()

+		var lastReport report

+		w := newWriter(

+			func(r report) {

+				lastReport = r

+			},

+			func(a report, b report) bool {

+				return true

+			},

+		)

+		w.(*imageProgressWriter).stableThreshhold = 0

+		_, err := io.Copy(w, pipeIn)

+		if err != nil {

+			if !test.errExpected {

+				t.Errorf("%s: unexpected: %v", test.name, err)

+			}

+			continue

+		}

+		if test.errExpected {

+			t.Errorf("%s: did not get expected error", test.name)

+			continue

+		}

+		if !compareReport(lastReport, test.expected) {

+			t.Errorf("%s: unexpected report, got: %v, expected: %v", test.name, lastReport, test.expected)

+		}

+	}

+}

+

+func TestErrorOnCopy(t *testing.T) {

+	// Producer pipe

+	genIn, genOut := io.Pipe()

+	p := newProgressGenerator(genOut)

+

+	// generate some data

+	go func() {

+		for i := 0; i < 100; i++ {

+			p.status("1", "Downloading")

+			p.status("2", "Downloading")

+			p.status("3", "Downloading")

+		}

+		p.err("data error")

+		genOut.Close()

+	}()

+

+	w := newWriter(func(r report) {}, func(a, b report) bool { return true })

+

+	// Ensure that the error is propagated to the copy

+	_, err := io.Copy(w, genIn)

+	if err == nil {

+		t.Errorf("Did not get an error when copying to writer")

+	}

+	if err.Error() != "data error" {

+		t.Errorf("Did not get expected error: %v", err)

+	}

+}

+

+func TestStableLayerCount(t *testing.T) {

+	var result report

+	w := newWriter(func(r report) { result = r }, func(a, b report) bool { return true })

+	w.(*imageProgressWriter).stableThreshhold = 3 // This means that the number of layers must be stable for at least 3 lines

+	p := newProgressGenerator(w)

+

+	// Increase number of layers by one each time

+	p.status("1", "one")

+	p.status("2", "two")

+	p.status("3", "three")

+	if result != nil {

+		t.Errorf("do not expect any reports at this point")

+		return

+	}

+

+	// Report on the same layers as before, keeping the

+	// the number stable

+	p.status("1", "one-a")

+	p.status("2", "two-a")

+	p.status("3", "three-a")

+	expected := report{

+		statusPending: &layerDetail{Count: 3},

+	}

+	if !compareReport(result, expected) {

+		t.Errorf("did not get expected report")

+	}

+}

+

+func compareReport(a, b report) bool {

+	if len(a) != len(b) {

+		return false

+	}

+	for k := range a {

+		if _, ok := b[k]; !ok {

+			return false

+		}

+		if !reflect.DeepEqual(*a[k], *b[k]) {

+			return false

+		}

+	}

+	return true

+}

+

+type progressGenerator json.Encoder

+

+func newProgressGenerator(w io.Writer) *progressGenerator {

+	return (*progressGenerator)(json.NewEncoder(w))

+}

+

+func (p *progressGenerator) status(id, status string) {

+	(*json.Encoder)(p).Encode(&progressLine{ID: id, Status: status})

+}

+func (p *progressGenerator) detail(id, status string, current, total int64) {

+	(*json.Encoder)(p).Encode(&progressLine{ID: id, Status: status, Detail: &progressDetail{Current: current, Total: total}})

+}

+func (p *progressGenerator) err(msg string) {

+	(*json.Encoder)(p).Encode(&progressLine{Error: msg})

+}

@@ -78,7 +78,7 @@ type file struct {

 }

 func (f file) Is(level int) bool {

-	return level <= f.level

+	return level <= f.level || bool(glog.V(glog.Level(level)))

 }

 func (f file) V(level int) Logger {