client/events.go
7c36a1af
 package client
 
 import (
d6bd79c1
 	"encoding/json"
7c36a1af
 	"net/url"
 	"time"
 
 	"golang.org/x/net/context"
 
 	"github.com/docker/docker/api/types"
d6bd79c1
 	"github.com/docker/docker/api/types/events"
7c36a1af
 	"github.com/docker/docker/api/types/filters"
 	timetypes "github.com/docker/docker/api/types/time"
 )
 
d6bd79c1
 // Events returns a stream of events in the daemon. It's up to the caller to close the stream
 // by cancelling the context. Once the stream has been completely read an io.EOF error will
 // be sent over the error channel. If an error is sent all processing will be stopped. It's up
 // to the caller to reopen the stream in the event of an error by reinvoking this method.
 func (cli *Client) Events(ctx context.Context, options types.EventsOptions) (<-chan events.Message, <-chan error) {
 
 	messages := make(chan events.Message)
 	errs := make(chan error, 1)
 
47585996
 	started := make(chan struct{})
d6bd79c1
 	go func() {
 		defer close(errs)
 
 		query, err := buildEventsQueryParams(cli.version, options)
 		if err != nil {
47585996
 			close(started)
d6bd79c1
 			errs <- err
 			return
 		}
 
 		resp, err := cli.get(ctx, "/events", query, nil)
 		if err != nil {
47585996
 			close(started)
d6bd79c1
 			errs <- err
 			return
 		}
 		defer resp.body.Close()
 
 		decoder := json.NewDecoder(resp.body)
 
47585996
 		close(started)
d6bd79c1
 		for {
 			select {
 			case <-ctx.Done():
 				errs <- ctx.Err()
 				return
 			default:
 				var event events.Message
 				if err := decoder.Decode(&event); err != nil {
 					errs <- err
 					return
 				}
 
 				select {
 				case messages <- event:
 				case <-ctx.Done():
 					errs <- ctx.Err()
 					return
 				}
 			}
 		}
 	}()
47585996
 	<-started
d6bd79c1
 
 	return messages, errs
 }
 
 func buildEventsQueryParams(cliVersion string, options types.EventsOptions) (url.Values, error) {
7c36a1af
 	query := url.Values{}
 	ref := time.Now()
 
 	if options.Since != "" {
 		ts, err := timetypes.GetTimestamp(options.Since, ref)
 		if err != nil {
 			return nil, err
 		}
 		query.Set("since", ts)
 	}
d6bd79c1
 
7c36a1af
 	if options.Until != "" {
 		ts, err := timetypes.GetTimestamp(options.Until, ref)
 		if err != nil {
 			return nil, err
 		}
 		query.Set("until", ts)
 	}
d6bd79c1
 
7c36a1af
 	if options.Filters.Len() > 0 {
d6bd79c1
 		filterJSON, err := filters.ToParamWithVersion(cliVersion, options.Filters)
7c36a1af
 		if err != nil {
 			return nil, err
 		}
 		query.Set("filters", filterJSON)
 	}
 
d6bd79c1
 	return query, nil
7c36a1af
 }