func (dg *dockerGoClient) ContainerEvents(ctx context.Context) (<-chan DockerContainerChangeEvent, error) {
	client, err := dg.sdkDockerClient()
	if err != nil {
		return nil, err
	}

	events := make(chan *events.Message)
	buffer := NewInfiniteBuffer()

	derivedCtx, cancel := context.WithCancel(ctx)
	dockerEvents, eventErr := client.Events(derivedCtx, types.EventsOptions{})

	// Cache the event from docker client. Channel closes when an error is passed to eventErr.
	go buffer.StartListening(derivedCtx, dockerEvents)
	// Receive errors from channels. If error thrown is not EOF, log and reopen channel.
	// TODO: move the error check into StartListening() to keep event streaming and error handling in one place.
	go func() {
		for {
			select {
			case err := <-eventErr:
				if err == io.EOF || err == io.ErrUnexpectedEOF {
					seelog.Info("DockerGoClient: All events have been received")
					cancel()
					return
				} else {
					// If an error is returned, we need to reopen channel to continue listening
					seelog.Errorf("DockerGoClient: error while listening to Docker Events : %v", err)
					nextCtx, nextCancel := context.WithCancel(ctx)
					dockerEvents, eventErr = client.Events(nextCtx, types.EventsOptions{})
					// Cache the event from docker client.
					go buffer.StartListening(nextCtx, dockerEvents)
					// Close previous stream after starting to listen on new one
					cancel()
					// Reassign cancel variable next Cancel function to setup next iteration of loop.
					cancel = nextCancel
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	// Read the buffered events and send to task engine
	go buffer.Consume(events)
	changedContainers := make(chan DockerContainerChangeEvent)
	go dg.handleContainerEvents(ctx, events, changedContainers)

	return changedContainers, nil
}
func (dg *dockerGoClient) ContainerEvents(ctx context.Context) (<-chan DockerContainerChangeEvent, error) {
	client, err := dg.sdkDockerClient()
	if err != nil {
		return nil, err
	}

	events := make(chan *events.Message)
	buffer := NewInfiniteBuffer()

	derivedCtx, cancel := context.WithCancel(ctx)
	dockerEvents, eventErr := client.Events(derivedCtx, types.EventsOptions{})

	
	go buffer.StartListening(derivedCtx, dockerEvents)
	
	
	go func() {
		for {
			select {
			case err := <-eventErr:
				if err == io.EOF || err == io.ErrUnexpectedEOF {
					seelog.Info("DockerGoClient: All events have been received")
					cancel()
					return
				} else {
					
					seelog.Errorf("DockerGoClient: error while listening to Docker Events : %v", err)
					nextCtx, nextCancel := context.WithCancel(ctx)
					dockerEvents, eventErr = client.Events(nextCtx, types.EventsOptions{})
					
					go buffer.StartListening(nextCtx, dockerEvents)
					
					cancel()
					
					cancel = nextCancel
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	
	go buffer.Consume(events)
	changedContainers := make(chan DockerContainerChangeEvent)
	go dg.handleContainerEvents(ctx, events, changedContainers)

	return changedContainers, nil
}
func ( dg * dockerGoClient ) ContainerEvents ( ctx context . Context ) ( <- chan DockerContainerChangeEvent , error ) { client , err := dg . sdkDockerClient ( ) if err != nil { return nil , err } events := make ( chan * events . Message ) buffer := NewInfiniteBuffer ( ) derivedCtx , cancel := context . WithCancel ( ctx ) dockerEvents , eventErr := client . Events ( derivedCtx , types . EventsOptions { } ) // Cache the event from docker client. Channel closes when an error is passed to eventErr. go buffer . StartListening ( derivedCtx , dockerEvents ) // Receive errors from channels. If error thrown is not EOF, log and reopen channel. // TODO: move the error check into StartListening() to keep event streaming and error handling in one place. go func ( ) { for { select { case err := <- eventErr : if err == io . EOF || err == io . ErrUnexpectedEOF { seelog . Info ( " " ) cancel ( ) return } else { // If an error is returned, we need to reopen channel to continue listening seelog . Errorf ( " " , err ) nextCtx , nextCancel := context . WithCancel ( ctx ) dockerEvents , eventErr = client . Events ( nextCtx , types . EventsOptions { } ) // Cache the event from docker client. go buffer . StartListening ( nextCtx , dockerEvents ) // Close previous stream after starting to listen on new one cancel ( ) // Reassign cancel variable next Cancel function to setup next iteration of loop. cancel = nextCancel } case <- ctx . Done ( ) : return } } } ( ) // Read the buffered events and send to task engine go buffer . Consume ( events ) changedContainers := make ( chan DockerContainerChangeEvent ) go dg . handleContainerEvents ( ctx , events , changedContainers ) return changedContainers , nil }
// Listen to the docker event stream for container changes and pass them up
Listen to the docker event stream for container changes and pass them up