Vanillaxi commented on code in PR #3482:
URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3583568303
##########
registry/polaris/core.go:
##########
@@ -72,60 +88,172 @@ func (watcher *PolarisServiceWatcher) lazyRun() {
})
}
+func (watcher *PolarisServiceWatcher) addSubscriberWithInitialSnapshot(
+ initialSnapshot []model.Instance,
+ subscriber item,
+) {
+ state := &subscriberState{
+ notify: subscriber,
+ initialSnapshot: copyInstances(initialSnapshot),
+ }
+
+ func() {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.subscribers = append(watcher.subscribers, state)
+ if watcher.snapshotReady {
+ watcher.reconcileSubscriberLocked(state)
+ }
+ }()
+
+ // Start only after the subscriber is registered and any available
current
+ // snapshot has been replayed.
+ watcher.lazyRun()
+}
+
+// missingInitialInstances returns initial - current by model.InstanceKey while
+// preserving the order of the initial snapshot.
+func missingInitialInstances(initial []model.Instance, current
[]model.Instance) []model.Instance {
+ currentInstances := make(map[model.InstanceKey]struct{}, len(current))
+ for _, instance := range current {
+ currentInstances[instance.GetInstanceKey()] = struct{}{}
+ }
+
+ missing := make([]model.Instance, 0, len(initial))
+ for _, instance := range initial {
+ if _, ok := currentInstances[instance.GetInstanceKey()]; !ok {
+ missing = append(missing, instance)
+ }
+ }
+ return missing
+}
+
+// handleWatchSnapshot replaces the watcher's current state and reconciles each
+// subscriber's own synchronous-load baseline exactly once.
+func (watcher *PolarisServiceWatcher) handleWatchSnapshot(current
[]model.Instance) {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.currentInstances = copyInstances(current)
+ watcher.snapshotReady = true
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.reconciled {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, watcher.currentInstances)
+ continue
+ }
+ watcher.reconcileSubscriberLocked(subscriber)
+ }
+}
+
+func (watcher *PolarisServiceWatcher) reconcileSubscriberLocked(subscriber
*subscriberState) {
+ missing := missingInitialInstances(subscriber.initialSnapshot,
watcher.currentInstances)
+ if len(missing) > 0 {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeDel, missing)
+ }
+ watcher.notifySubscriberLocked(subscriber, remoting.EventTypeAdd,
watcher.currentInstances)
+ subscriber.reconciled = true
+ subscriber.initialSnapshot = nil
+}
+
// startWatch start run work to watch target service by polaris
func (watcher *PolarisServiceWatcher) startWatch() {
for {
- resp, err :=
watcher.consumer.WatchService(watcher.subscribeParam)
- if err != nil {
+ if err := watcher.watchOnce(); err != nil {
time.Sleep(time.Duration(500 * time.Millisecond))
- continue
}
- watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: resp.GetAllInstancesResp.Instances,
- ConfigType: remoting.EventTypeAdd,
- })
-
- for event := range resp.EventChannel {
- eType := event.GetSubScribeEventType()
- if eType == internalapi.EventInstance {
- insEvent := event.(*model.InstanceEvent)
-
- if insEvent.AddEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.AddEvent.Instances,
- ConfigType:
remoting.EventTypeAdd,
- })
- }
- if insEvent.UpdateEvent != nil {
- instances := make([]model.Instance,
len(insEvent.UpdateEvent.UpdateList))
- for i := range
insEvent.UpdateEvent.UpdateList {
- instances[i] =
insEvent.UpdateEvent.UpdateList[i].After
- }
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: instances,
- ConfigType:
remoting.EventTypeUpdate,
- })
- }
- if insEvent.DeleteEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.DeleteEvent.Instances,
- ConfigType:
remoting.EventTypeDel,
- })
- }
+ }
+}
+
+func (watcher *PolarisServiceWatcher) watchOnce() error {
+ resp, err := watcher.consumer.WatchService(watcher.subscribeParam)
+ if err != nil {
+ return err
+ }
+ watcher.handleWatchSnapshot(resp.GetAllInstancesResp.Instances)
+ for event := range resp.EventChannel {
+ if event.GetSubScribeEventType() == internalapi.EventInstance {
+
watcher.handleInstanceEvent(event.(*model.InstanceEvent))
+ }
+ }
+ return nil
+}
+
+func (watcher *PolarisServiceWatcher) handleInstanceEvent(event
*model.InstanceEvent) {
+ if event == nil {
+ return
+ }
+
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ if event.AddEvent != nil {
+ instances := copyInstances(event.AddEvent.Instances)
+ for _, instance := range instances {
+ watcher.upsertCurrentInstanceLocked(instance)
+ }
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeAdd, instances)
+ }
+ if event.UpdateEvent != nil {
+ instances := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ for _, update := range event.UpdateEvent.UpdateList {
+ if update.Before.GetInstanceKey() !=
update.After.GetInstanceKey() {
+
watcher.removeCurrentInstancesLocked([]model.Instance{update.Before})
}
+ watcher.upsertCurrentInstanceLocked(update.After)
+ instances = append(instances, update.After)
}
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeUpdate, instances)
+ }
+ if event.DeleteEvent != nil {
+ instances := copyInstances(event.DeleteEvent.Instances)
+ watcher.removeCurrentInstancesLocked(instances)
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeDel, instances)
+ }
+}
+func (watcher *PolarisServiceWatcher) upsertCurrentInstanceLocked(instance
model.Instance) {
+ key := instance.GetInstanceKey()
+ for i := range watcher.currentInstances {
+ if watcher.currentInstances[i].GetInstanceKey() == key {
+ watcher.currentInstances[i] = instance
+ return
+ }
}
+ watcher.currentInstances = append(watcher.currentInstances, instance)
}
-// notifyAllSubscriber notify config_center.ConfigChangeEvent to all subscriber
-func (watcher *PolarisServiceWatcher) notifyAllSubscriber(event
*config_center.ConfigChangeEvent) {
- watcher.lock.RLock()
- defer watcher.lock.RUnlock()
+func (watcher *PolarisServiceWatcher) removeCurrentInstancesLocked(instances
[]model.Instance) {
+ keys := make(map[model.InstanceKey]struct{}, len(instances))
+ for _, instance := range instances {
+ keys[instance.GetInstanceKey()] = struct{}{}
+ }
- for i := 0; i < len(watcher.subscribers); i++ {
- subscriber := watcher.subscribers[i]
- subscriber(event.ConfigType, event.Value.([]model.Instance))
+ current := watcher.currentInstances[:0]
+ for _, instance := range watcher.currentInstances {
+ if _, remove := keys[instance.GetInstanceKey()]; !remove {
+ current = append(current, instance)
+ }
}
+ watcher.currentInstances = current
Review Comment:
Thanks,done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
