vcsim: add support for PropertyCollector incremental updates

Prior to this change, clients could only call WaitForUpdatesEx once,
which would return the current state.  The next invocation with Version
given would fail as unsupported.  Clients can now use WaitForUpdatesEx
to wait for multiple updates.

The basic idea is that methods should now update object fields using
the Registry.UpdateObject method instead of setting fields directly.
UpdateObject uses the same PropertyChange structure that is sent as
part of the WaitForUpdatesEx response.  UpdateObject will also apply
the PropertyChange(s) to the local object, removing the need to set
fields directly.

- govc commands now use Client.Flag.WithCancel to catch SIGINT,
  allowing proper cleanup of collectors

- Add vcsim ListView support to ViewManager

Fixes vmware#922

govc/events/command.go

@@ -164,24 +164,21 @@ func (cmd *events) Run(ctx context.Context, f *flag.FlagSet) error {
 		return err
 	}
 
-	if len(objs) > 0 {
-		// need an event manager
-		m := event.NewManager(c)
-
-		// get the event stream
-		err = m.Events(ctx, objs, cmd.Max, cmd.Tail, cmd.Force, func(obj types.ManagedObjectReference, ee []types.BaseEvent) error {
-			var o *types.ManagedObjectReference
-			if len(objs) > 1 {
-				o = &obj
-			}
+	if len(objs) == 0 {
+		return nil
+	}
 
-			return cmd.printEvents(ctx, o, ee, m)
-		}, cmd.Kind...)
+	m := event.NewManager(c)
 
-		if err != nil {
-			return err
-		}
-	}
+	return cmd.WithCancel(ctx, func(wctx context.Context) error {
+		return m.Events(wctx, objs, cmd.Max, cmd.Tail, cmd.Force,
+			func(obj types.ManagedObjectReference, ee []types.BaseEvent) error {
+				var o *types.ManagedObjectReference
+				if len(objs) > 1 {
+					o = &obj
+				}
 
-	return nil
+				return cmd.printEvents(ctx, o, ee, m)
+			}, cmd.Kind...)
+	})
 }

govc/flags/client.go

@@ -28,8 +28,10 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"os/signal"
 	"path/filepath"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/vmware/govmomi/session"
@@ -625,3 +627,29 @@ func (flag *ClientFlag) Environ(extra bool) []string {
 
 	return env
 }
+
+// WithCancel calls the given function, returning when complete or canceled via SIGINT.
+func (flag *ClientFlag) WithCancel(ctx context.Context, f func(context.Context) error) error {
+	wctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	done := make(chan bool)
+	var werr error
+
+	go func() {
+		defer close(done)
+		werr = f(wctx)
+	}()
+
+	sig := make(chan os.Signal)
+	signal.Notify(sig, syscall.SIGINT)
+
+	select {
+	case <-sig:
+		cancel()
+		<-done // Wait for f() to complete
+	case <-done:
+	}
+
+	return werr
+}

govc/object/collect.go

@@ -26,6 +26,7 @@ import (
 	"reflect"
 	"strings"
 	"text/tabwriter"
+	"time"
 
 	"github.com/vmware/govmomi/govc/cli"
 	"github.com/vmware/govmomi/govc/flags"
@@ -47,6 +48,7 @@ type collect struct {
 	dump   bool
 	n      int
 	kind   kinds
+	wait   time.Duration
 
 	filter property.Filter
 	obj    string
@@ -65,6 +67,7 @@ func (cmd *collect) Register(ctx context.Context, f *flag.FlagSet) {
 	f.StringVar(&cmd.raw, "R", "", "Raw XML encoded CreateFilter request")
 	f.IntVar(&cmd.n, "n", 0, "Wait for N property updates")
 	f.Var(&cmd.kind, "type", "Resource type.  If specified, MOID is used for a container view root")
+	f.DurationVar(&cmd.wait, "wait", 0, "Max wait time for updates")
 }
 
 func (cmd *collect) Usage() string {
@@ -388,41 +391,48 @@ func (cmd *collect) Run(ctx context.Context, f *flag.FlagSet) error {
 	entered := false
 	hasFilter := len(cmd.filter) != 0
 
-	return property.WaitForUpdates(ctx, p, filter, func(updates []types.ObjectUpdate) bool {
-		matches := 0
+	if cmd.wait != 0 {
+		filter.Options = &types.WaitOptions{
+			MaxWaitSeconds: types.NewInt32(int32(cmd.wait.Seconds())),
+		}
+	}
 
-		for _, update := range updates {
-			if entered && update.Kind == types.ObjectUpdateKindEnter {
-				// on the first update we only get kind "enter"
-				// if a new object is added, the next update with have both "enter" and "modify".
-				continue
-			}
+	return cmd.WithCancel(ctx, func(wctx context.Context) error {
+		return property.WaitForUpdates(wctx, p, filter, func(updates []types.ObjectUpdate) bool {
+			matches := 0
+			for _, update := range updates {
+				if entered && update.Kind == types.ObjectUpdateKindEnter {
+					// on the first update we only get kind "enter"
+					// if a new object is added, the next update with have both "enter" and "modify".
+					continue
+				}
 
-			c := &change{cmd, update}
+				c := &change{cmd, update}
 
-			if hasFilter {
-				if cmd.match(update) {
-					matches++
-				} else {
-					continue
+				if hasFilter {
+					if cmd.match(update) {
+						matches++
+					} else {
+						continue
+					}
 				}
+
+				_ = cmd.WriteResult(c)
 			}
 
-			_ = cmd.WriteResult(c)
-		}
+			entered = true
 
-		entered = true
+			if hasFilter {
+				if matches > 0 {
+					return true
+				}
 
-		if hasFilter {
-			if matches > 0 {
-				return true
+				return false
 			}
 
-			return false
-		}
-
-		cmd.n--
+			cmd.n--
 
-		return cmd.n == -1
+			return cmd.n == -1 && cmd.wait == 0
+		})
 	})
 }

property/collector.go

@@ -111,6 +111,12 @@ func (p *Collector) WaitForUpdates(ctx context.Context, v string) (*types.Update
 	return res.Returnval, nil
 }
 
+func (p *Collector) CancelWaitForUpdates(ctx context.Context) error {
+	req := &types.CancelWaitForUpdates{This: p.Reference()}
+	_, err := methods.CancelWaitForUpdates(ctx, p.roundTripper, req)
+	return err
+}
+
 func (p *Collector) RetrieveProperties(ctx context.Context, req types.RetrieveProperties) (*types.RetrievePropertiesResponse, error) {
 	req.This = p.Reference()
 	return methods.RetrieveProperties(ctx, p.roundTripper, &req)

property/wait.go

@@ -19,12 +19,14 @@ package property
 import (
 	"context"
 
+	"github.com/vmware/govmomi/vim25/methods"
 	"github.com/vmware/govmomi/vim25/types"
 )
 
 // WaitFilter provides helpers to construct a types.CreateFilter for use with property.Wait
 type WaitFilter struct {
 	types.CreateFilter
+	Options *types.WaitOptions
 }
 
 // Add a new ObjectSpec and PropertySpec to the WaitFilter
@@ -75,6 +77,7 @@ func Wait(ctx context.Context, c *Collector, obj types.ManagedObjectReference, p
 // creates a new property collector and calls CreateFilter. A new property
 // collector is required because filters can only be added, not removed.
 //
+// If the Context is canceled, a call to CancelWaitForUpdates() is made and its error value is returned.
 // The newly created collector is destroyed before this function returns (both
 // in case of success or error).
 //
@@ -85,28 +88,41 @@ func WaitForUpdates(ctx context.Context, c *Collector, filter *WaitFilter, f fun
 	}
 
 	// Attempt to destroy the collector using the background context, as the
-	// specified context may have timed out or have been cancelled.
+	// specified context may have timed out or have been canceled.
 	defer p.Destroy(context.Background())
 
 	err = p.CreateFilter(ctx, filter.CreateFilter)
 	if err != nil {
 		return err
 	}
 
-	for version := ""; ; {
-		res, err := p.WaitForUpdates(ctx, version)
+	req := types.WaitForUpdatesEx{
+		This:    p.Reference(),
+		Options: filter.Options,
+	}
+
+	for {
+		res, err := methods.WaitForUpdatesEx(ctx, p.roundTripper, &req)
 		if err != nil {
+			if ctx.Err() == context.Canceled {
+				werr := p.CancelWaitForUpdates(context.Background())
+				return werr
+			}
 			return err
 		}
 
-		// Retry if the result came back empty
-		if res == nil {
+		set := res.Returnval
+		if set == nil {
+			if req.Options != nil && req.Options.MaxWaitSeconds != nil {
+				return nil // WaitOptions.MaxWaitSeconds exceeded
+			}
+			// Retry if the result came back empty
 			continue
 		}
 
-		version = res.Version
+		req.Version = set.Version
 
-		for _, fs := range res.FilterSet {
+		for _, fs := range set.FilterSet {
 			if f(fs.ObjectSet) {
 				return nil
 			}

scripts/debug-xmlformat.sh

@@ -13,6 +13,8 @@ for file in *.req.xml; do
     header Request "$file" "${base}.req.headers"
     xmlformat < "$file"
     file="${base}.res.xml"
-    header Response "$file" "${base}.res.headers"
-    xmlformat < "$file"
+    if [ -e "$file" ] ; then
+      header Response "$file" "${base}.res.headers"
+      xmlformat < "$file"
+    fi
 done

simulator/event_manager.go

@@ -157,14 +157,15 @@ func (m *EventManager) PostEvent(ctx *Context, req *types.PostEvent) soap.HasFau
 	event.CreatedTime = time.Now()
 	event.UserName = ctx.Session.UserName
 
-	m.page = m.page.Next()
+	m.page = m.page.Prev()
 	m.page.Value = req.EventToPost
 	m.formatMessage(req.EventToPost)
 
 	for _, c := range m.collectors {
 		if c.eventMatches(req.EventToPost) {
-			c.page = c.page.Next()
-			c.page.Value = event
+			c.page = c.page.Prev()
+			c.page.Value = req.EventToPost
+			Map.Update(c, []types.PropertyChange{{Name: "latestPage", Val: c.GetLatestPage()}})
 		}
 	}