Fixes hacks from progressreader refactor

related to moby#10959

Signed-off-by: bobby abbott <[email protected]>

api/client/build.go

@@ -22,9 +22,11 @@ import (
 	"github.com/docker/docker/graph"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/fileutils"
+	"github.com/docker/docker/pkg/jsonmessage"
 	flag "github.com/docker/docker/pkg/mflag"
 	"github.com/docker/docker/pkg/parsers"
 	"github.com/docker/docker/pkg/progressreader"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/symlink"
 	"github.com/docker/docker/pkg/units"
 	"github.com/docker/docker/pkg/urlutil"
@@ -198,7 +200,7 @@ func (cli *DockerCli) CmdBuild(args ...string) error {
 	// Setup an upload progress bar
 	// FIXME: ProgressReader shouldn't be this annoying to use
 	if context != nil {
-		sf := utils.NewStreamFormatter(false)
+		sf := streamformatter.NewStreamFormatter(false)
 		body = progressreader.New(progressreader.Config{
 			In:        context,
 			Out:       cli.out,
@@ -291,7 +293,7 @@ func (cli *DockerCli) CmdBuild(args ...string) error {
 		headers.Set("Content-Type", "application/tar")
 	}
 	err = cli.stream("POST", fmt.Sprintf("/build?%s", v.Encode()), body, cli.out, headers)
-	if jerr, ok := err.(*utils.JSONError); ok {
+	if jerr, ok := err.(*jsonmessage.JSONError); ok {
 		// If no error code is set, default to 1
 		if jerr.Code == 0 {
 			jerr.Code = 1

api/client/utils.go

@@ -19,11 +19,11 @@ import (
 	"github.com/docker/docker/api"
 	"github.com/docker/docker/autogen/dockerversion"
 	"github.com/docker/docker/engine"
+	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/pkg/signal"
 	"github.com/docker/docker/pkg/stdcopy"
 	"github.com/docker/docker/pkg/term"
 	"github.com/docker/docker/registry"
-	"github.com/docker/docker/utils"
 )
 
 var (
@@ -164,7 +164,7 @@ func (cli *DockerCli) streamHelper(method, path string, setRawTerminal bool, in
 	}
 
 	if api.MatchesContentType(resp.Header.Get("Content-Type"), "application/json") {
-		return utils.DisplayJSONMessagesStream(resp.Body, stdout, cli.outFd, cli.isTerminalOut)
+		return jsonmessage.DisplayJSONMessagesStream(resp.Body, stdout, cli.outFd, cli.isTerminalOut)
 	}
 	if stdout != nil || stderr != nil {
 		// When TTY is ON, use regular copy

api/server/server.go

@@ -32,6 +32,7 @@ import (
 	"github.com/docker/docker/pkg/listenbuffer"
 	"github.com/docker/docker/pkg/parsers"
 	"github.com/docker/docker/pkg/stdcopy"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/version"
 	"github.com/docker/docker/registry"
 	"github.com/docker/docker/utils"
@@ -595,7 +596,7 @@ func postImagesCreate(eng *engine.Engine, version version.Version, w http.Respon
 		if !job.Stdout.Used() {
 			return err
 		}
-		sf := utils.NewStreamFormatter(version.GreaterThan("1.0"))
+		sf := streamformatter.NewStreamFormatter(version.GreaterThan("1.0"))
 		w.Write(sf.FormatError(err))
 	}
 
@@ -680,7 +681,7 @@ func postImagesPush(eng *engine.Engine, version version.Version, w http.Response
 		if !job.Stdout.Used() {
 			return err
 		}
-		sf := utils.NewStreamFormatter(version.GreaterThan("1.0"))
+		sf := streamformatter.NewStreamFormatter(version.GreaterThan("1.0"))
 		w.Write(sf.FormatError(err))
 	}
 	return nil
@@ -1107,7 +1108,7 @@ func postBuild(eng *engine.Engine, version version.Version, w http.ResponseWrite
 		if !job.Stdout.Used() {
 			return err
 		}
-		sf := utils.NewStreamFormatter(version.GreaterThanOrEqualTo("1.8"))
+		sf := streamformatter.NewStreamFormatter(version.GreaterThanOrEqualTo("1.8"))
 		w.Write(sf.FormatError(err))
 	}
 	return nil

builder/evaluator.go

@@ -33,6 +33,7 @@ import (
 	"github.com/docker/docker/daemon"
 	"github.com/docker/docker/engine"
 	"github.com/docker/docker/pkg/fileutils"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/docker/docker/pkg/symlink"
 	"github.com/docker/docker/pkg/tarsum"
@@ -105,7 +106,7 @@ type Builder struct {
 
 	// Deprecated, original writer used for ImagePull. To be removed.
 	OutOld          io.Writer
-	StreamFormatter *utils.StreamFormatter
+	StreamFormatter *streamformatter.StreamFormatter
 
 	Config *runconfig.Config // runconfig for cmd, run, entrypoint etc.
 

builder/internals.go

@@ -26,6 +26,7 @@ import (
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/chrootarchive"
 	"github.com/docker/docker/pkg/ioutils"
+	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/pkg/parsers"
 	"github.com/docker/docker/pkg/progressreader"
 	"github.com/docker/docker/pkg/stringid"
@@ -601,7 +602,7 @@ func (b *Builder) run(c *daemon.Container) error {
 
 	// Wait for it to finish
 	if ret, _ := c.WaitStop(-1 * time.Second); ret != 0 {
-		err := &utils.JSONError{
+		err := &jsonmessage.JSONError{
 			Message: fmt.Sprintf("The command %v returned a non-zero code: %d", b.Config.Cmd, ret),
 			Code:    ret,
 		}

builder/job.go

@@ -17,6 +17,7 @@ import (
 	"github.com/docker/docker/graph"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/parsers"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/urlutil"
 	"github.com/docker/docker/registry"
 	"github.com/docker/docker/runconfig"
@@ -127,16 +128,16 @@ func (b *BuilderJob) CmdBuild(job *engine.Job) error {
 	}
 	defer context.Close()
 
-	sf := utils.NewStreamFormatter(job.GetenvBool("json"))
+	sf := streamformatter.NewStreamFormatter(job.GetenvBool("json"))
 
 	builder := &Builder{
 		Daemon: b.Daemon,
 		Engine: b.Engine,
-		OutStream: &utils.StdoutFormater{
+		OutStream: &streamformatter.StdoutFormater{
 			Writer:          job.Stdout,
 			StreamFormatter: sf,
 		},
-		ErrStream: &utils.StderrFormater{
+		ErrStream: &streamformatter.StderrFormater{
 			Writer:          job.Stdout,
 			StreamFormatter: sf,
 		},

events/events.go

@@ -10,23 +10,23 @@ import (
 	"time"
 
 	"github.com/docker/docker/engine"
+	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/pkg/parsers/filters"
-	"github.com/docker/docker/utils"
 )
 
 const eventsLimit = 64
 
-type listener chan<- *utils.JSONMessage
+type listener chan<- *jsonmessage.JSONMessage
 
 type Events struct {
 	mu          sync.RWMutex
-	events      []*utils.JSONMessage
+	events      []*jsonmessage.JSONMessage
 	subscribers []listener
 }
 
 func New() *Events {
 	return &Events{
-		events: make([]*utils.JSONMessage, 0, eventsLimit),
+		events: make([]*jsonmessage.JSONMessage, 0, eventsLimit),
 	}
 }
 
@@ -63,7 +63,7 @@ func (e *Events) Get(job *engine.Job) error {
 		timeout.Stop()
 	}
 
-	listener := make(chan *utils.JSONMessage)
+	listener := make(chan *jsonmessage.JSONMessage)
 	e.subscribe(listener)
 	defer e.unsubscribe(listener)
 
@@ -107,7 +107,7 @@ func (e *Events) SubscribersCount(job *engine.Job) error {
 	return nil
 }
 
-func writeEvent(job *engine.Job, event *utils.JSONMessage, eventFilters filters.Args) error {
+func writeEvent(job *engine.Job, event *jsonmessage.JSONMessage, eventFilters filters.Args) error {
 	isFiltered := func(field string, filter []string) bool {
 		if len(filter) == 0 {
 			return false
@@ -170,7 +170,7 @@ func (e *Events) subscribersCount() int {
 func (e *Events) log(action, id, from string) {
 	e.mu.Lock()
 	now := time.Now().UTC().Unix()
-	jm := &utils.JSONMessage{Status: action, ID: id, From: from, Time: now}
+	jm := &jsonmessage.JSONMessage{Status: action, ID: id, From: from, Time: now}
 	if len(e.events) == cap(e.events) {
 		// discard oldest event
 		copy(e.events, e.events[1:])

events/events_test.go

@@ -9,13 +9,13 @@ import (
 	"time"
 
 	"github.com/docker/docker/engine"
-	"github.com/docker/docker/utils"
+	"github.com/docker/docker/pkg/jsonmessage"
 )
 
 func TestEventsPublish(t *testing.T) {
 	e := New()
-	l1 := make(chan *utils.JSONMessage)
-	l2 := make(chan *utils.JSONMessage)
+	l1 := make(chan *jsonmessage.JSONMessage)
+	l2 := make(chan *jsonmessage.JSONMessage)
 	e.subscribe(l1)
 	e.subscribe(l2)
 	count := e.subscribersCount()
@@ -61,7 +61,7 @@ func TestEventsPublish(t *testing.T) {
 
 func TestEventsPublishTimeout(t *testing.T) {
 	e := New()
-	l := make(chan *utils.JSONMessage)
+	l := make(chan *jsonmessage.JSONMessage)
 	e.subscribe(l)
 
 	c := make(chan struct{})
@@ -108,9 +108,9 @@ func TestLogEvents(t *testing.T) {
 	}
 	buf = bytes.NewBuffer(buf.Bytes())
 	dec := json.NewDecoder(buf)
-	var msgs []utils.JSONMessage
+	var msgs []jsonmessage.JSONMessage
 	for {
-		var jm utils.JSONMessage
+		var jm jsonmessage.JSONMessage
 		if err := dec.Decode(&jm); err != nil {
 			if err == io.EOF {
 				break
@@ -138,8 +138,8 @@ func TestEventsCountJob(t *testing.T) {
 	if err := e.Install(eng); err != nil {
 		t.Fatal(err)
 	}
-	l1 := make(chan *utils.JSONMessage)
-	l2 := make(chan *utils.JSONMessage)
+	l1 := make(chan *jsonmessage.JSONMessage)
+	l2 := make(chan *jsonmessage.JSONMessage)
 	e.subscribe(l1)
 	e.subscribe(l2)
 	job := eng.Job("subscribers_count")

graph/graph.go

@@ -18,6 +18,7 @@ import (
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/progressreader"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/docker/docker/pkg/truncindex"
 	"github.com/docker/docker/runconfig"
@@ -198,7 +199,7 @@ func (graph *Graph) Register(img *image.Image, layerData archive.ArchiveReader)
 //   The archive is stored on disk and will be automatically deleted as soon as has been read.
 //   If output is not nil, a human-readable progress bar will be written to it.
 //   FIXME: does this belong in Graph? How about MktempFile, let the caller use it for archives?
-func (graph *Graph) TempLayerArchive(id string, sf *utils.StreamFormatter, output io.Writer) (*archive.TempArchive, error) {
+func (graph *Graph) TempLayerArchive(id string, sf *streamformatter.StreamFormatter, output io.Writer) (*archive.TempArchive, error) {
 	image, err := graph.Get(id)
 	if err != nil {
 		return nil, err

graph/import.go

@@ -11,6 +11,7 @@ import (
 	"github.com/docker/docker/engine"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/progressreader"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/runconfig"
 	"github.com/docker/docker/utils"
 )
@@ -23,7 +24,7 @@ func (s *TagStore) CmdImport(job *engine.Job) error {
 		src          = job.Args[0]
 		repo         = job.Args[1]
 		tag          string
-		sf           = utils.NewStreamFormatter(job.GetenvBool("json"))
+		sf           = streamformatter.NewStreamFormatter(job.GetenvBool("json"))
 		archive      archive.ArchiveReader
 		resp         *http.Response
 		stdoutBuffer = bytes.NewBuffer(nil)

graph/pull.go

@@ -15,6 +15,7 @@ import (
 	"github.com/docker/docker/engine"
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/pkg/progressreader"
+	"github.com/docker/docker/pkg/streamformatter"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/docker/docker/registry"
 	"github.com/docker/docker/utils"
@@ -28,7 +29,7 @@ func (s *TagStore) CmdPull(job *engine.Job) error {
 	var (
 		localName   = job.Args[0]
 		tag         string
-		sf          = utils.NewStreamFormatter(job.GetenvBool("json"))
+		sf          = streamformatter.NewStreamFormatter(job.GetenvBool("json"))
 		authConfig  = &registry.AuthConfig{}
 		metaHeaders map[string][]string
 	)
@@ -107,7 +108,7 @@ func (s *TagStore) CmdPull(job *engine.Job) error {
 	return nil
 }
 
-func (s *TagStore) pullRepository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, askedTag string, sf *utils.StreamFormatter, parallel bool) error {
+func (s *TagStore) pullRepository(r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, askedTag string, sf *streamformatter.StreamFormatter, parallel bool) error {
 	out.Write(sf.FormatStatus("", "Pulling repository %s", repoInfo.CanonicalName))
 
 	repoData, err := r.GetRepositoryData(repoInfo.RemoteName)
@@ -265,7 +266,7 @@ func (s *TagStore) pullRepository(r *registry.Session, out io.Writer, repoInfo *
 	return nil
 }
 
-func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint string, token []string, sf *utils.StreamFormatter) (bool, error) {
+func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint string, token []string, sf *streamformatter.StreamFormatter) (bool, error) {
 	history, err := r.GetRemoteHistory(imgID, endpoint, token)
 	if err != nil {
 		return false, err
@@ -363,7 +364,7 @@ func (s *TagStore) pullImage(r *registry.Session, out io.Writer, imgID, endpoint
 	return layers_downloaded, nil
 }
 
-func WriteStatus(requestedTag string, out io.Writer, sf *utils.StreamFormatter, layers_downloaded bool) {
+func WriteStatus(requestedTag string, out io.Writer, sf *streamformatter.StreamFormatter, layers_downloaded bool) {
 	if layers_downloaded {
 		out.Write(sf.FormatStatus("", "Status: Downloaded newer image for %s", requestedTag))
 	} else {
@@ -382,7 +383,7 @@ type downloadInfo struct {
 	err        chan error
 }
 
-func (s *TagStore) pullV2Repository(eng *engine.Engine, r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *utils.StreamFormatter, parallel bool) error {
+func (s *TagStore) pullV2Repository(eng *engine.Engine, r *registry.Session, out io.Writer, repoInfo *registry.RepositoryInfo, tag string, sf *streamformatter.StreamFormatter, parallel bool) error {
 	endpoint, err := r.V2RegistryEndpoint(repoInfo.Index)
 	if err != nil {
 		if repoInfo.Index.Official {
@@ -428,7 +429,7 @@ func (s *TagStore) pullV2Repository(eng *engine.Engine, r *registry.Session, out
 	return nil
 }
 
-func (s *TagStore) pullV2Tag(eng *engine.Engine, r *registry.Session, out io.Writer, endpoint *registry.Endpoint, repoInfo *registry.RepositoryInfo, tag string, sf *utils.StreamFormatter, parallel bool, auth *registry.RequestAuthorization) (bool, error) {
+func (s *TagStore) pullV2Tag(eng *engine.Engine, r *registry.Session, out io.Writer, endpoint *registry.Endpoint, repoInfo *registry.RepositoryInfo, tag string, sf *streamformatter.StreamFormatter, parallel bool, auth *registry.RequestAuthorization) (bool, error) {
 	log.Debugf("Pulling tag from V2 registry: %q", tag)
 
 	manifestBytes, manifestDigest, err := r.GetV2ImageManifest(endpoint, repoInfo.RemoteName, tag, auth)