bugfix: remove consumer of polaris (#1962)

registry/polaris/listener.go

@@ -18,7 +18,6 @@
 package polaris
 
 import (
- "bytes"
  "net/url"
  "strconv"
 )
@@ -34,7 +33,6 @@ import (
 
 import (
  "dubbo.apache.org/dubbo-go/v3/common"
- "dubbo.apache.org/dubbo-go/v3/common/constant"
  "dubbo.apache.org/dubbo-go/v3/config_center"
  "dubbo.apache.org/dubbo-go/v3/registry"
 )
@@ -78,13 +76,6 @@ func (pl *polarisListener) Close() {
  close(pl.closeCh)
 }
 
-func getSubscribeName(url *common.URL) string {
- var buffer bytes.Buffer
- buffer.Write([]byte(common.DubboNodes[common.PROVIDER]))
- appendParam(&buffer, url, constant.InterfaceKey)
- return buffer.String()
-}
-
 func generateUrl(instance model.Instance) *common.URL {
  if instance.GetMetadata() == nil {
  logger.Errorf("polaris instance metadata is empty,instance:%+v", instance)

registry/polaris/registry.go

@@ -37,9 +37,7 @@ import (
  "dubbo.apache.org/dubbo-go/v3/common"
  "dubbo.apache.org/dubbo-go/v3/common/constant"
  "dubbo.apache.org/dubbo-go/v3/common/extension"
- "dubbo.apache.org/dubbo-go/v3/config_center"
  "dubbo.apache.org/dubbo-go/v3/registry"
- "dubbo.apache.org/dubbo-go/v3/remoting"
  "dubbo.apache.org/dubbo-go/v3/remoting/polaris"
 )
 
@@ -62,26 +60,22 @@ func newPolarisRegistry(url *common.URL) (registry.Registry, error) {
  return &polarisRegistry{}, err
  }
  pRegistry := &polarisRegistry{
- consumer: api.NewConsumerAPIByContext(sdkCtx),
  provider: api.NewProviderAPIByContext(sdkCtx),
  lock: &sync.RWMutex{},
  registryUrls: make(map[string]*PolarisHeartbeat),
  listenerLock: &sync.RWMutex{},
- watchers: make(map[string]*PolarisServiceWatcher),
  }
 
  return pRegistry, nil
 }
 
 type polarisRegistry struct {
  url *common.URL
- consumer api.ConsumerAPI
  provider api.ProviderAPI
  lock *sync.RWMutex
  registryUrls map[string]*PolarisHeartbeat
 
  listenerLock *sync.RWMutex
- watchers map[string]*PolarisServiceWatcher
 }
 
 // Register will register the service @url to its polaris registry center.
@@ -161,12 +155,6 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No
  return nil
  }
 
- watcher, err := pr.createPolarisWatcherIfAbsent(url)
-
- if err != nil {
- return err
- }
-
  for {
  listener, err := NewPolarisListener(url)
  if err != nil {
@@ -175,13 +163,6 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No
  continue
  }
 
- watcher.AddSubscriber(func(et remoting.EventType, instances []model.Instance) {
- for i := range instances {
- instance := instances[i]
- listener.events.In() <- &config_center.ConfigChangeEvent{ConfigType: et, Value: instance}
- }
- })
-
  for {
  serviceEvent, err := listener.Next()
  if err != nil {
@@ -195,33 +176,6 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No
  }
 }
 
-func (pr *polarisRegistry) createPolarisWatcherIfAbsent(url *common.URL) (*PolarisServiceWatcher, error) {
-
- pr.listenerLock.Lock()
- defer pr.listenerLock.Unlock()
-
- serviceName := getSubscribeName(url)
-
- if _, exist := pr.watchers[serviceName]; !exist {
- subscribeParam := &api.WatchServiceRequest{
- WatchServiceRequest: model.WatchServiceRequest{
- Key: model.ServiceKey{
- Namespace: url.GetParam(constant.PolarisNamespace, constant.PolarisDefaultNamespace),
- Service: serviceName,
- },
- },
- }
-
- watcher, err := newPolarisWatcher(subscribeParam, pr.consumer)
- if err != nil {
- return nil, err
- }
- pr.watchers[serviceName] = watcher
- }
-
- return pr.watchers[serviceName], nil
-}
-
 // UnSubscribe returns nil if unsubscribing registry successfully. If not returns an error.
 func (pr *polarisRegistry) UnSubscribe(url *common.URL, notifyListener registry.NotifyListener) error {
  // TODO wait polaris support it