AsperforMias commented on code in PR #3482:
URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3590994810
##########
registry/polaris/core.go:
##########
@@ -72,60 +88,174 @@ 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)
Review Comment:
`handleWatchSnapshot` 也会在 `EventChannel` 关闭后的重连中执行。已 `reconciled` 的
subscriber 此处只接收 `ADD(current)`，未计算并下发 `DEL(previous-current)`。断线期间 A
下线、重连快照仅包含 B 时，directory 会保留 A 的旧 invoker。
##########
registry/polaris/registry.go:
##########
@@ -188,14 +225,91 @@ func (pr *polarisRegistry) LoadSubscribeInstances(url
*common.URL, notify regist
return perrors.New(fmt.Sprintf("could not query the instances
for serviceName=%s,namespace=%s,error=%v",
serviceName, pr.namespace, err))
}
+ initialSubscribeInstances := make([]model.Instance, 0,
len(resp.Instances))
for i := range resp.Instances {
if newUrl := generateUrl(resp.Instances[i]); newUrl != nil {
notify.Notify(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: newUrl})
+ initialSubscribeInstances =
append(initialSubscribeInstances, resp.Instances[i])
}
}
+ pr.storeInitialSubscribeInstances(key, initialSubscribeInstances)
return nil
}
+func newInitialSubscribeInstancesKey(
+ serviceName string,
+ notify registry.NotifyListener,
+) (initialSubscribeInstancesKey, error) {
+ if isNilNotifyListener(notify) {
+ return initialSubscribeInstancesKey{}, fmt.Errorf("notify
listener type %T is nil", notify)
+ }
+ if !reflect.TypeOf(notify).Comparable() {
Review Comment:
`NotifyListener` 接口未要求动态类型可比较。这里会使具名 slice/map/func
等合法实现从此前可用变为返回错误；现有集成若使用此类 listener，`LoadSubscribeInstances` 和 `Subscribe`
将无法建立订阅。
--
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]
