[GitHub] asifdxtreme closed pull request #435: SCB-890 Lost changed event when bootstrap with embedded etcd

[GitBox]

asifdxtreme closed pull request #435: SCB-890 Lost changed event when bootstrap 
with embedded etcd
URL: https://github.com/apache/incubator-servicecomb-service-center/pull/435
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/server/plugin/infra/registry/embededetcd/embededetcd.go 
b/server/plugin/infra/registry/embededetcd/embededetcd.go
index 20e7525a..4f928539 100644
--- a/server/plugin/infra/registry/embededetcd/embededetcd.go
+++ b/server/plugin/infra/registry/embededetcd/embededetcd.go
@@ -418,35 +418,9 @@ func (s *EtcdEmbed) Watch(ctx context.Context, opts 
...registry.PluginOpOption)
                                        return
                                }
 
-                               l := len(resp.Events)
-                               kvs := make([]*mvccpb.KeyValue, l)
-                               pIdx, prevAction := 0, mvccpb.PUT
-                               pResp := &registry.PluginResponse{Action: 
registry.Put, Succeeded: true}
-
-                               for _, evt := range resp.Events {
-                                       if prevAction != evt.Type {
-                                               prevAction = evt.Type
-
-                                               if pIdx > 0 {
-                                                       err = 
setResponseAndCallback(pResp, kvs[:pIdx], op.WatchCallback)
-                                                       if err != nil {
-                                                               return
-                                                       }
-                                                       pIdx = 0
-                                               }
-                                       }
-
-                                       pResp.Revision = evt.Kv.ModRevision
-                                       pResp.Action = 
setKvsAndConvertAction(kvs, pIdx, &evt)
-
-                                       pIdx++
-                               }
-
-                               if pIdx > 0 {
-                                       err = setResponseAndCallback(pResp, 
kvs[:pIdx], op.WatchCallback)
-                                       if err != nil {
-                                               return
-                                       }
+                               err = dispatch(resp.Events, op.WatchCallback)
+                               if err != nil {
+                                       return
                                }
                        }
                }
@@ -478,7 +452,39 @@ func (s *EtcdEmbed) readyNotify() {
        }
 }
 
-func setKvsAndConvertAction(kvs []*mvccpb.KeyValue, pIdx int, evt 
*mvccpb.Event) registry.ActionType {
+func dispatch(evts []mvccpb.Event, cb registry.WatchCallback) error {
+       l := len(evts)
+       kvs := make([]*mvccpb.KeyValue, l)
+       sIdx, eIdx, rev := 0, 0, int64(0)
+       action, prevEvtType := registry.Put, mvccpb.PUT
+
+       for _, evt := range evts {
+               if prevEvtType != evt.Type {
+                       if eIdx > 0 {
+                               err := callback(action, rev, kvs[sIdx:eIdx], cb)
+                               if err != nil {
+                                       return err
+                               }
+                               sIdx = eIdx
+                       }
+                       prevEvtType = evt.Type
+               }
+
+               if rev < evt.Kv.ModRevision {
+                       rev = evt.Kv.ModRevision
+               }
+               action = setKvsAndConvertAction(kvs, eIdx, evt)
+
+               eIdx++
+       }
+
+       if eIdx > 0 {
+               return callback(action, rev, kvs[sIdx:eIdx], cb)
+       }
+       return nil
+}
+
+func setKvsAndConvertAction(kvs []*mvccpb.KeyValue, pIdx int, evt 
mvccpb.Event) registry.ActionType {
        switch evt.Type {
        case mvccpb.DELETE:
                kv := evt.PrevKv
@@ -493,15 +499,14 @@ func setKvsAndConvertAction(kvs []*mvccpb.KeyValue, pIdx 
int, evt *mvccpb.Event)
        }
 }
 
-func setResponseAndCallback(pResp *registry.PluginResponse, kvs 
[]*mvccpb.KeyValue, cb registry.WatchCallback) error {
-       pResp.Count = int64(len(kvs))
-       pResp.Kvs = kvs
-
-       err := cb("key information changed", pResp)
-       if err != nil {
-               return err
-       }
-       return nil
+func callback(action registry.ActionType, rev int64, kvs []*mvccpb.KeyValue, 
cb registry.WatchCallback) error {
+       return cb("key information changed", &registry.PluginResponse{
+               Action:    action,
+               Kvs:       kvs,
+               Count:     int64(len(kvs)),
+               Revision:  rev,
+               Succeeded: true,
+       })
 }
 
 func getEmbedInstance() mgr.PluginInstance {


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services