init fields

Author: karimra

client/cmd/datastore_watchdeviations.go

@@ -41,15 +41,31 @@ var datastoreWatchDeviationCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
+		count := 0
 		for {
 			rsp, err := stream.Recv()
 			if err != nil {
 				return err
 			}
 			switch format {
 			case "":
-				fmt.Printf("%s: %s: %s: %s: %s : %s -> %s\n", rsp.GetName(), rsp.GetIntent(), rsp.GetEvent(), rsp.GetReason(), utils.ToXPath(rsp.GetPath(), false), rsp.GetExpectedValue(), rsp.GetCurrentValue())
-				rsp.GetName()
+				switch rsp.Event {
+				case sdcpb.DeviationEvent_START:
+					fmt.Printf("%s: %s: %s\n",
+						rsp.GetName(), rsp.GetIntent(), rsp.GetEvent())
+				case sdcpb.DeviationEvent_END:
+					fmt.Printf("%s: %s: %s\n",
+						rsp.GetName(), rsp.GetIntent(), rsp.GetEvent())
+
+					fmt.Printf("Received %d deviations\n", count)
+					count = 0
+				default:
+					count++
+					fmt.Printf("%s: %s: %s: %s: %s : %s -> %s\n",
+						rsp.GetName(), rsp.GetIntent(), rsp.GetEvent(),
+						rsp.GetReason(), utils.ToXPath(rsp.GetPath(), false),
+						rsp.GetExpectedValue(), rsp.GetCurrentValue())
+				}
 			case "json":
 				b, err := json.MarshalIndent(rsp, "", "  ")
 				if err != nil {
@@ -58,8 +74,6 @@ var datastoreWatchDeviationCmd = &cobra.Command{
 				fmt.Println(string(b))
 			}
 		}
-
-		return nil
 	},
 }
 

pkg/datastore/datastore_rpc.go

@@ -85,12 +85,14 @@ type Datastore struct {
 // func New(c *config.DatastoreConfig, schemaServer *config.RemoteSchemaServer) *Datastore {
 func New(ctx context.Context, c *config.DatastoreConfig, scc schema.Client, cc cache.Client, opts ...grpc.DialOption) *Datastore {
 	ds := &Datastore{
-		config:           c,
-		schemaClient:     scc,
-		cacheClient:      cc,
-		intentMutex:      new(sync.Mutex),
-		m:                new(sync.RWMutex),
-		deviationClients: make(map[string]sdcpb.DataServer_WatchDeviationsServer),
+		config:                   c,
+		schemaClient:             scc,
+		cacheClient:              cc,
+		intentMutex:              new(sync.Mutex),
+		m:                        new(sync.RWMutex),
+		deviationClients:         make(map[string]sdcpb.DataServer_WatchDeviationsServer),
+		md:                       new(sync.RWMutex),
+		currentIntentsDeviations: make(map[string][]*sdcpb.WatchDeviationResponse),
 	}
 	if c.Sync != nil {
 		ds.synCh = make(chan *target.SyncUpdate, c.Sync.Buffer)
@@ -1102,6 +1104,9 @@ func (d *Datastore) runDeviationUpdate(ctx context.Context, dm map[string]sdcpb.
 				}
 			}
 			xp := utils.ToXPath(sp, false)
+			if _, ok := newDeviations[xp]; !ok {
+				newDeviations[xp] = make([]*sdcpb.WatchDeviationResponse, 0, 1)
+			}
 			newDeviations[xp] = append(newDeviations[xp], rsp)
 		}
 		// remaining intents
@@ -1148,6 +1153,9 @@ func (d *Datastore) runDeviationUpdate(ctx context.Context, dm map[string]sdcpb.
 					}
 				}
 				xp := utils.ToXPath(sp, false)
+				if _, ok := newDeviations[xp]; !ok {
+					newDeviations[xp] = make([]*sdcpb.WatchDeviationResponse, 0, 1)
+				}
 				newDeviations[xp] = append(newDeviations[xp], rsp)
 			}
 		}