Files
docker-cli/components/cli/vendor/github.com/docker/docker/client/events.go
Sebastiaan van Stijn a964103223 Bump docker/docker to 079ed017b61eb819b8184b90013ce89465d3aaba
- Add API support for SCTP port mapping
- Add canonical import path
- Add `REMOVE` and `ORPHANED` to TaskState
- Fix TLS from environment variables in client
- Introduce NewClientWithOpts func to build custom client easily
- Wrap response errors for container copy methodsto fix error detection using
  `IsErrNotFound` and `IsErrNotImplemented` for `ContainerStatPath`,
  `CopyFromContainer`, and `CopyToContainer` methods.
- Produce errors when empty ids are passed into inspect calls

Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Upstream-commit: aaa7a7cb95
Component: cli
2018-02-20 17:17:49 +01:00

103 lines
2.2 KiB
Go

package client // import "github.com/docker/docker/client"
import (
"encoding/json"
"net/url"
"time"
"golang.org/x/net/context"
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/events"
"github.com/docker/docker/api/types/filters"
timetypes "github.com/docker/docker/api/types/time"
)
// 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)
started := make(chan struct{})
go func() {
defer close(errs)
query, err := buildEventsQueryParams(cli.version, options)
if err != nil {
close(started)
errs <- err
return
}
resp, err := cli.get(ctx, "/events", query, nil)
if err != nil {
close(started)
errs <- err
return
}
defer resp.body.Close()
decoder := json.NewDecoder(resp.body)
close(started)
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
}
}
}
}()
<-started
return messages, errs
}
func buildEventsQueryParams(cliVersion string, options types.EventsOptions) (url.Values, error) {
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)
}
if options.Until != "" {
ts, err := timetypes.GetTimestamp(options.Until, ref)
if err != nil {
return nil, err
}
query.Set("until", ts)
}
if options.Filters.Len() > 0 {
filterJSON, err := filters.ToParamWithVersion(cliVersion, options.Filters)
if err != nil {
return nil, err
}
query.Set("filters", filterJSON)
}
return query, nil
}