syncbase: Send initial root update in Watch.
An empty update for the root entity is sent as the first update on the
watch stream.
* For watches from an empty resume marker, this update is propagated to
the client as part of the initial state batch. It is necessary because
the initial state batch may not contain other updates (if the watch
filter matches no collections) and the client needs to be notified of
this. It is intended to be filtered out by the high-level API when
the initial state batch is being separated out.
* For watches from 'now' or a specific resume marker, the initial update
is sent with initial_state_skipped for compliance with
v.io/v23/services/watch. It is filtered out by the client library
instead of propagating to the user.
Initial update was hooked up end-to-end in Go. Only row updates are
propagated in Java and Cgo for now.
Errors in watch code have also been scrubbed to prevent leaking info
to the client before the authorization step.
MultiPart: 4/4
Change-Id: Ice6b87cae732cc165a9b4b64a83ee19df6ed436a
diff --git a/go/src/v.io/x/sensorlog/internal/client/list.go b/go/src/v.io/x/sensorlog/internal/client/list.go
index 04f1ea8..d21cb8b 100644
--- a/go/src/v.io/x/sensorlog/internal/client/list.go
+++ b/go/src/v.io/x/sensorlog/internal/client/list.go
@@ -74,6 +74,9 @@
trans := make([]*dataPoint, 0, 16)
for ws.Advance() {
c := ws.Change()
+ if c.EntityType != syncbase.EntityRow {
+ continue
+ }
var elem dataPoint
if err := elem.Key.Parse(c.Row); err != nil {
return fmt.Errorf("invalid DataPoint key for prefix %s: %v", dataPrefix, err)
diff --git a/go/src/v.io/x/sensorlog/internal/measure/watcher.go b/go/src/v.io/x/sensorlog/internal/measure/watcher.go
index e9776c0..bf59a10 100644
--- a/go/src/v.io/x/sensorlog/internal/measure/watcher.go
+++ b/go/src/v.io/x/sensorlog/internal/measure/watcher.go
@@ -40,6 +40,9 @@
for ws.Advance() {
c := ws.Change()
+ if c.EntityType != syncbase.EntityRow {
+ continue
+ }
key := &sbmodel.KStreamDef{}
if err := key.Parse(c.Row); err != nil {
return fmt.Errorf("invalid StreamDef key for prefix %s: %v", watchPrefix, err)
