Refactor to new events api

Signed-off-by: Josh Horwitz <horwitzja@gmail.com>

command/container/run.go

@@ -211,10 +211,7 @@ func runRun(dockerCli *command.DockerCli, flags *pflag.FlagSet, opts *runOptions
 		})
 	}
 
-	statusChan, err := waitExitOrRemoved(dockerCli, context.Background(), createResponse.ID, hostConfig.AutoRemove)
-	if err != nil {
-		return fmt.Errorf("Error waiting container's exit code: %v", err)
-	}
+	statusChan := waitExitOrRemoved(dockerCli, ctx, createResponse.ID, hostConfig.AutoRemove)
 
 	//start the container
 	if err := client.ContainerStart(ctx, createResponse.ID, types.ContainerStartOptions{}); err != nil {

command/container/start.go

@@ -108,7 +108,7 @@ func runStart(dockerCli *command.DockerCli, opts *startOptions) error {
 
 		// 3. We should open a channel for receiving status code of the container
 		// no matter it's detached, removed on daemon side(--rm) or exit normally.
-		statusChan, statusErr := waitExitOrRemoved(dockerCli, context.Background(), c.ID, c.HostConfig.AutoRemove)
+		statusChan := waitExitOrRemoved(dockerCli, ctx, c.ID, c.HostConfig.AutoRemove)
 		startOptions := types.ContainerStartOptions{
 			CheckpointID: opts.checkpoint,
 		}
@@ -117,7 +117,7 @@ func runStart(dockerCli *command.DockerCli, opts *startOptions) error {
 		if err := dockerCli.Client().ContainerStart(ctx, c.ID, startOptions); err != nil {
 			cancelFun()
 			<-cErr
-			if c.HostConfig.AutoRemove && statusErr == nil {
+			if c.HostConfig.AutoRemove {
 				// wait container to be removed
 				<-statusChan
 			}
@@ -134,10 +134,6 @@ func runStart(dockerCli *command.DockerCli, opts *startOptions) error {
 			return attchErr
 		}
 
-		if statusErr != nil {
-			return fmt.Errorf("can't get container's exit code: %v", statusErr)
-		}
-
 		if status := <-statusChan; status != 0 {
 			return cli.StatusError{StatusCode: status}
 		}

command/container/stats.go

@@ -63,24 +63,22 @@ func runStats(dockerCli *command.DockerCli, opts *statsOptions) error {
 		options := types.EventsOptions{
 			Filters: f,
 		}
-		resBody, err := dockerCli.Client().Events(ctx, options)
-		// Whether we successfully subscribed to events or not, we can now
+
+		eventq, errq := dockerCli.Client().Events(ctx, options)
+
+		// Whether we successfully subscribed to eventq or not, we can now
 		// unblock the main goroutine.
 		close(started)
-		if err != nil {
-			closeChan <- err
-			return
-		}
-		defer resBody.Close()
 
-		system.DecodeEvents(resBody, func(event events.Message, err error) error {
-			if err != nil {
+		for {
+			select {
+			case event := <-eventq:
+				c <- event
+			case err := <-errq:
 				closeChan <- err
-				return nil
+				return
 			}
-			c <- event
-			return nil
-		})
+		}
 	}
 
 	// waitFirst is a WaitGroup to wait first stat data's reach for each container

command/container/utils.go

@@ -1,7 +1,6 @@
 package container
 
 import (
-	"fmt"
 	"strconv"
 
 	"golang.org/x/net/context"
@@ -11,11 +10,10 @@ import (
 	"github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/cli/command"
-	"github.com/docker/docker/cli/command/system"
 	clientapi "github.com/docker/docker/client"
 )
 
-func waitExitOrRemoved(dockerCli *command.DockerCli, ctx context.Context, containerID string, waitRemove bool) (chan int, error) {
+func waitExitOrRemoved(dockerCli *command.DockerCli, ctx context.Context, containerID string, waitRemove bool) chan int {
 	if len(containerID) == 0 {
 		// containerID can never be empty
 		panic("Internal Error: waitExitOrRemoved needs a containerID as parameter")
@@ -24,11 +22,7 @@ func waitExitOrRemoved(dockerCli *command.DockerCli, ctx context.Context, contai
 	statusChan := make(chan int)
 	exitCode := 125
 
-	eventProcessor := func(e events.Message, err error) error {
-		if err != nil {
-			statusChan <- exitCode
-			return fmt.Errorf("failed to decode event: %v", err)
-		}
+	eventProcessor := func(e events.Message) bool {
 
 		stopProcessing := false
 		switch e.Status {
@@ -53,11 +47,10 @@ func waitExitOrRemoved(dockerCli *command.DockerCli, ctx context.Context, contai
 
 		if stopProcessing {
 			statusChan <- exitCode
-			// stop the loop processing
-			return fmt.Errorf("done")
+			return true
 		}
 
-		return nil
+		return false
 	}
 
 	// Get events via Events API
@@ -67,14 +60,29 @@ func waitExitOrRemoved(dockerCli *command.DockerCli, ctx context.Context, contai
 	options := types.EventsOptions{
 		Filters: f,
 	}
-	resBody, err := dockerCli.Client().Events(ctx, options)
-	if err != nil {
-		return nil, fmt.Errorf("can't get events from daemon: %v", err)
-	}
 
-	go system.DecodeEvents(resBody, eventProcessor)
+	eventCtx, cancel := context.WithCancel(ctx)
+	eventq, errq := dockerCli.Client().Events(eventCtx, options)
 
-	return statusChan, nil
+	go func() {
+		defer cancel()
+
+		for {
+			select {
+			case evt := <-eventq:
+				if eventProcessor(evt) {
+					return
+				}
+
+			case err := <-errq:
+				logrus.Errorf("error getting events from daemon: %v", err)
+				statusChan <- exitCode
+				return
+			}
+		}
+	}()
+
+	return statusChan
 }
 
 // getExitCode performs an inspect on the container. It returns