Merge pull request #18266 from calavera/events_pub_sub

Event PubSub topics + linear filtering.
Upstream-commit: 33ab2bb52c130380e038013d68fdd8ad3c663360
Component: engine

components/engine/api/client/events.go

@@ -26,7 +26,7 @@ func (cli *DockerCli) CmdEvents(args ...string) error {
 
 	var (
 		v               = url.Values{}
-		eventFilterArgs = filters.Args{}
+		eventFilterArgs = filters.NewArgs()
 	)
 
 	// Consolidate all filter flags, and sanity check them early.
@@ -53,7 +53,7 @@ func (cli *DockerCli) CmdEvents(args ...string) error {
 		}
 		v.Set("until", ts)
 	}
-	if len(eventFilterArgs) > 0 {
+	if eventFilterArgs.Len() > 0 {
 		filterJSON, err := filters.ToParam(eventFilterArgs)
 		if err != nil {
 			return err

components/engine/api/client/images.go

@@ -36,7 +36,7 @@ func (cli *DockerCli) CmdImages(args ...string) error {
 
 	// Consolidate all filter flags, and sanity check them early.
 	// They'll get process in the daemon/server.
-	imageFilterArgs := filters.Args{}
+	imageFilterArgs := filters.NewArgs()
 	for _, f := range flFilter.GetAll() {
 		var err error
 		imageFilterArgs, err = filters.ParseFlag(f, imageFilterArgs)
@@ -47,7 +47,7 @@ func (cli *DockerCli) CmdImages(args ...string) error {
 
 	matchName := cmd.Arg(0)
 	v := url.Values{}
-	if len(imageFilterArgs) > 0 {
+	if imageFilterArgs.Len() > 0 {
 		filterJSON, err := filters.ToParam(imageFilterArgs)
 		if err != nil {
 			return err

components/engine/api/client/ps.go

@@ -20,7 +20,7 @@ func (cli *DockerCli) CmdPs(args ...string) error {
 	var (
 		err error
 
-		psFilterArgs = filters.Args{}
+		psFilterArgs = filters.NewArgs()
 		v            = url.Values{}
 
 		cmd      = Cli.Subcmd("ps", nil, Cli.DockerCommands["ps"].Description, true)
@@ -72,7 +72,7 @@ func (cli *DockerCli) CmdPs(args ...string) error {
 		}
 	}
 
-	if len(psFilterArgs) > 0 {
+	if psFilterArgs.Len() > 0 {
 		filterJSON, err := filters.ToParam(psFilterArgs)
 		if err != nil {
 			return err

components/engine/api/client/volume.go

@@ -54,7 +54,7 @@ func (cli *DockerCli) CmdVolumeLs(args ...string) error {
 	cmd.Require(flag.Exact, 0)
 	cmd.ParseFlags(args, true)
 
-	volFilterArgs := filters.Args{}
+	volFilterArgs := filters.NewArgs()
 	for _, f := range flFilter.GetAll() {
 		var err error
 		volFilterArgs, err = filters.ParseFlag(f, volFilterArgs)
@@ -64,7 +64,7 @@ func (cli *DockerCli) CmdVolumeLs(args ...string) error {
 	}
 
 	v := url.Values{}
-	if len(volFilterArgs) > 0 {
+	if volFilterArgs.Len() > 0 {
 		filterJSON, err := filters.ToParam(volFilterArgs)
 		if err != nil {
 			return err

components/engine/api/server/router/local/info.go

@@ -92,27 +92,11 @@ func (s *router) getEvents(ctx context.Context, w http.ResponseWriter, r *http.R
 
 	enc := json.NewEncoder(output)
 
-	current, l, cancel := s.daemon.SubscribeToEvents()
-	defer cancel()
+	buffered, l := s.daemon.SubscribeToEvents(since, sinceNano, ef)
+	defer s.daemon.UnsubscribeFromEvents(l)
 
-	eventFilter := s.daemon.GetEventFilter(ef)
-	handleEvent := func(ev *jsonmessage.JSONMessage) error {
-		if eventFilter.Include(ev) {
-			if err := enc.Encode(ev); err != nil {
-				return err
-			}
-		}
-		return nil
-	}
-
-	if since == -1 {
-		current = nil
-	}
-	for _, ev := range current {
-		if ev.Time < since || ((ev.Time == since) && (ev.TimeNano < sinceNano)) {
-			continue
-		}
-		if err := handleEvent(ev); err != nil {
+	for _, ev := range buffered {
+		if err := enc.Encode(ev); err != nil {
 			return err
 		}
 	}
@@ -129,7 +113,7 @@ func (s *router) getEvents(ctx context.Context, w http.ResponseWriter, r *http.R
 			if !ok {
 				continue
 			}
-			if err := handleEvent(jev); err != nil {
+			if err := enc.Encode(jev); err != nil {
 				return err
 			}
 		case <-timer.C:

components/engine/api/server/router/network/network_routes.go

@@ -29,27 +29,23 @@ func (n *networkRouter) getNetworksList(ctx context.Context, w http.ResponseWrit
 	}
 
 	list := []*types.NetworkResource{}
-	var nameFilter, idFilter bool
-	var names, ids []string
-	if names, nameFilter = netFilters["name"]; nameFilter {
-		for _, name := range names {
-			if nw, err := n.backend.GetNetwork(name, daemon.NetworkByName); err == nil {
-				list = append(list, buildNetworkResource(nw))
-			} else {
-				logrus.Errorf("failed to get network for filter=%s : %v", name, err)
-			}
+	netFilters.WalkValues("name", func(name string) error {
+		if nw, err := n.backend.GetNetwork(name, daemon.NetworkByName); err == nil {
+			list = append(list, buildNetworkResource(nw))
+		} else {
+			logrus.Errorf("failed to get network for filter=%s : %v", name, err)
 		}
-	}
+		return nil
+	})
 
-	if ids, idFilter = netFilters["id"]; idFilter {
-		for _, id := range ids {
-			for _, nw := range n.backend.GetNetworksByID(id) {
-				list = append(list, buildNetworkResource(nw))
-			}
+	netFilters.WalkValues("id", func(id string) error {
+		for _, nw := range n.backend.GetNetworksByID(id) {
+			list = append(list, buildNetworkResource(nw))
 		}
-	}
+		return nil
+	})
 
-	if !nameFilter && !idFilter {
+	if !netFilters.Include("name") && !netFilters.Include("id") {
 		nwList := n.backend.GetNetworksByID("")
 		for _, nw := range nwList {
 			list = append(list, buildNetworkResource(nw))