From 62cc50a46507b1f62dac8d2a623303dbd9ae37e2 Mon Sep 17 00:00:00 2001
From: "vito.he" <hxmhlt@163.com>
Date: Tue, 14 May 2019 11:44:16 +0800
Subject: [PATCH] Mod:cluster some rename & fix a bug in registryDirectory
---
cluster/cluster_impl/failover_cluster.go | 6 +-
cluster/cluster_impl/mock_cluster.go | 6 +-
.../cluster_impl/registry_aware_cluster.go | 6 +-
cluster/directory/service_array.go | 78 -------------------
cluster/directory/static_directory.go | 12 +--
cluster/loadbalance/random.go | 6 +-
registry/directory/directory.go | 43 +++++-----
registry/directory/directory_test.go | 2 +-
8 files changed, 38 insertions(+), 121 deletions(-)
delete mode 100644 cluster/directory/service_array.go
diff --git a/cluster/cluster_impl/failover_cluster.go b/cluster/cluster_impl/failover_cluster.go
index 08ffff193..f85680465 100644
--- a/cluster/cluster_impl/failover_cluster.go
+++ b/cluster/cluster_impl/failover_cluster.go
@@ -6,7 +6,7 @@ import (
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
-type FailoverCluster struct {
+type failoverCluster struct {
}
const name = "failover"
@@ -16,9 +16,9 @@ func init() {
}
func NewFailoverCluster() cluster.Cluster {
- return &FailoverCluster{}
+ return &failoverCluster{}
}
-func (cluster *FailoverCluster) Join(directory cluster.Directory) protocol.Invoker {
+func (cluster *failoverCluster) Join(directory cluster.Directory) protocol.Invoker {
return newFailoverClusterInvoker(directory)
}
diff --git a/cluster/cluster_impl/mock_cluster.go b/cluster/cluster_impl/mock_cluster.go
index e8a3eea7b..d046bd321 100644
--- a/cluster/cluster_impl/mock_cluster.go
+++ b/cluster/cluster_impl/mock_cluster.go
@@ -5,13 +5,13 @@ import (
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
-type MockCluster struct {
+type mockCluster struct {
}
func NewMockCluster() cluster.Cluster {
- return &MockCluster{}
+ return &mockCluster{}
}
-func (cluster *MockCluster) Join(directory cluster.Directory) protocol.Invoker {
+func (cluster *mockCluster) Join(directory cluster.Directory) protocol.Invoker {
return protocol.NewBaseInvoker(directory.GetUrl())
}
diff --git a/cluster/cluster_impl/registry_aware_cluster.go b/cluster/cluster_impl/registry_aware_cluster.go
index 1f2e030d5..f722c3e9b 100644
--- a/cluster/cluster_impl/registry_aware_cluster.go
+++ b/cluster/cluster_impl/registry_aware_cluster.go
@@ -6,7 +6,7 @@ import (
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
-type RegistryAwareCluster struct {
+type registryAwareCluster struct {
}
func init() {
@@ -14,9 +14,9 @@ func init() {
}
func NewRegistryAwareCluster() cluster.Cluster {
- return &RegistryAwareCluster{}
+ return ®istryAwareCluster{}
}
-func (cluster *RegistryAwareCluster) Join(directory cluster.Directory) protocol.Invoker {
+func (cluster *registryAwareCluster) Join(directory cluster.Directory) protocol.Invoker {
return newRegistryAwareClusterInvoker(directory)
}
diff --git a/cluster/directory/service_array.go b/cluster/directory/service_array.go
deleted file mode 100644
index 4d4a92254..000000000
--- a/cluster/directory/service_array.go
+++ /dev/null
@@ -1,78 +0,0 @@
-package directory
-
-import (
- "context"
- "fmt"
- "strings"
- "time"
-)
-
-import (
- jerrors "github.com/juju/errors"
-)
-import (
- "github.com/dubbo/go-for-apache-dubbo/common"
-)
-
-//////////////////////////////////////////
-// registry array
-// should be returned by registry ,will be used by client & waiting to selector
-//////////////////////////////////////////
-
-var (
- ErrServiceArrayEmpty = jerrors.New("registryArray empty")
- ErrServiceArrayTimeout = jerrors.New("registryArray timeout")
-)
-
-type ServiceArray struct {
- context context.Context
- arr []common.URL
- birth time.Time
- idx int64
-}
-
-func NewServiceArray(ctx context.Context, arr []common.URL) *ServiceArray {
- return &ServiceArray{
- context: ctx,
- arr: arr,
- birth: time.Now(),
- }
-}
-
-func (s *ServiceArray) GetIdx() *int64 {
- return &s.idx
-}
-
-func (s *ServiceArray) GetSize() int64 {
- return int64(len(s.arr))
-}
-
-func (s *ServiceArray) GetService(i int64) common.URL {
- return s.arr[i]
-}
-
-func (s *ServiceArray) String() string {
- var builder strings.Builder
- builder.WriteString(fmt.Sprintf("birth:%s, idx:%d, arr len:%d, arr:{", s.birth, s.idx, len(s.arr)))
- for i := range s.arr {
- builder.WriteString(fmt.Sprintf("%d:%s, ", i, s.arr[i]))
- }
- builder.WriteString("}")
-
- return builder.String()
-}
-
-func (s *ServiceArray) Add(url common.URL, ttl time.Duration) {
- s.arr = append(s.arr, url)
- s.birth = time.Now().Add(ttl)
-}
-
-func (s *ServiceArray) Del(url common.URL, ttl time.Duration) {
- for i, svc := range s.arr {
- if svc.PrimitiveURL == url.PrimitiveURL {
- s.arr = append(s.arr[:i], s.arr[i+1:]...)
- s.birth = time.Now().Add(ttl)
- break
- }
- }
-}
diff --git a/cluster/directory/static_directory.go b/cluster/directory/static_directory.go
index bab58e3c9..cedc019ac 100644
--- a/cluster/directory/static_directory.go
+++ b/cluster/directory/static_directory.go
@@ -5,20 +5,20 @@ import (
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
-type StaticDirectory struct {
+type staticDirectory struct {
BaseDirectory
invokers []protocol.Invoker
}
-func NewStaticDirectory(invokers []protocol.Invoker) *StaticDirectory {
- return &StaticDirectory{
+func NewStaticDirectory(invokers []protocol.Invoker) *staticDirectory {
+ return &staticDirectory{
BaseDirectory: NewBaseDirectory(&common.URL{}),
invokers: invokers,
}
}
//for-loop invokers ,if all invokers is available ,then it means directory is available
-func (dir *StaticDirectory) IsAvailable() bool {
+func (dir *staticDirectory) IsAvailable() bool {
for _, invoker := range dir.invokers {
if !invoker.IsAvailable() {
return false
@@ -27,12 +27,12 @@ func (dir *StaticDirectory) IsAvailable() bool {
return true
}
-func (dir *StaticDirectory) List(invocation protocol.Invocation) []protocol.Invoker {
+func (dir *staticDirectory) List(invocation protocol.Invocation) []protocol.Invoker {
//TODO:Here should add router
return dir.invokers
}
-func (dir *StaticDirectory) Destroy() {
+func (dir *staticDirectory) Destroy() {
dir.BaseDirectory.Destroy(func() {
for _, ivk := range dir.invokers {
ivk.Destroy()
diff --git a/cluster/loadbalance/random.go b/cluster/loadbalance/random.go
index 85780e345..2772f17e3 100644
--- a/cluster/loadbalance/random.go
+++ b/cluster/loadbalance/random.go
@@ -17,14 +17,14 @@ func init() {
extension.SetLoadbalance(name, NewRandomLoadBalance)
}
-type RandomLoadBalance struct {
+type randomLoadBalance struct {
}
func NewRandomLoadBalance() cluster.LoadBalance {
- return &RandomLoadBalance{}
+ return &randomLoadBalance{}
}
-func (lb *RandomLoadBalance) Select(invokers []protocol.Invoker, url common.URL, invocation protocol.Invocation) protocol.Invoker {
+func (lb *randomLoadBalance) Select(invokers []protocol.Invoker, url common.URL, invocation protocol.Invocation) protocol.Invoker {
var length int
if length = len(invokers); length == 1 {
return invokers[0]
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index 504677dca..fd5a94fd9 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -27,7 +27,7 @@ type Options struct {
}
type Option func(*Options)
-type RegistryDirectory struct {
+type registryDirectory struct {
directory.BaseDirectory
cacheInvokers []protocol.Invoker
listenerLock sync.Mutex
@@ -38,7 +38,7 @@ type RegistryDirectory struct {
Options
}
-func NewRegistryDirectory(url *common.URL, registry registry.Registry, opts ...Option) (*RegistryDirectory, error) {
+func NewRegistryDirectory(url *common.URL, registry registry.Registry, opts ...Option) (*registryDirectory, error) {
options := Options{
//default 300s
serviceTTL: time.Duration(300e9),
@@ -49,7 +49,7 @@ func NewRegistryDirectory(url *common.URL, registry registry.Registry, opts ...O
if url.SubURL == nil {
return nil, jerrors.Errorf("url is invalid, suburl can not be nil")
}
- return &RegistryDirectory{
+ return ®istryDirectory{
BaseDirectory: directory.NewBaseDirectory(url),
cacheInvokers: []protocol.Invoker{},
cacheInvokersMap: &sync.Map{},
@@ -60,7 +60,7 @@ func NewRegistryDirectory(url *common.URL, registry registry.Registry, opts ...O
}
//subscibe from registry
-func (dir *RegistryDirectory) Subscribe(url common.URL) {
+func (dir *registryDirectory) Subscribe(url common.URL) {
for {
if !dir.registry.IsAvailable() {
log.Warn("event listener game over.")
@@ -95,7 +95,7 @@ func (dir *RegistryDirectory) Subscribe(url common.URL) {
}
//subscribe service from registry , and update the cacheServices
-func (dir *RegistryDirectory) update(res *registry.ServiceEvent) {
+func (dir *registryDirectory) update(res *registry.ServiceEvent) {
if res == nil {
return
}
@@ -107,35 +107,34 @@ func (dir *RegistryDirectory) update(res *registry.ServiceEvent) {
dir.refreshInvokers(res)
}
-func (dir *RegistryDirectory) refreshInvokers(res *registry.ServiceEvent) {
- var newCacheInvokersMap sync.Map
+func (dir *registryDirectory) refreshInvokers(res *registry.ServiceEvent) {
switch res.Action {
case registry.ServiceAdd:
//dir.cacheService.Add(res.Path, dir.serviceTTL)
- newCacheInvokersMap = *dir.cacheInvoker(res.Service)
+ dir.cacheInvoker(res.Service)
case registry.ServiceDel:
//dir.cacheService.Del(res.Path, dir.serviceTTL)
- newCacheInvokersMap = *dir.uncacheInvoker(res.Service)
+ dir.uncacheInvoker(res.Service)
log.Info("selector delete service url{%s}", res.Service)
default:
return
}
- newInvokers := dir.toGroupInvokers(&newCacheInvokersMap)
+ newInvokers := dir.toGroupInvokers()
dir.listenerLock.Lock()
defer dir.listenerLock.Unlock()
dir.cacheInvokers = newInvokers
}
-func (dir *RegistryDirectory) toGroupInvokers(newInvokersMap *sync.Map) []protocol.Invoker {
+func (dir *registryDirectory) toGroupInvokers() []protocol.Invoker {
newInvokersList := []protocol.Invoker{}
groupInvokersMap := make(map[string][]protocol.Invoker)
groupInvokersList := []protocol.Invoker{}
- newInvokersMap.Range(func(key, value interface{}) bool {
+ dir.cacheInvokersMap.Range(func(key, value interface{}) bool {
newInvokersList = append(newInvokersList, value.(protocol.Invoker))
return true
})
@@ -163,42 +162,38 @@ func (dir *RegistryDirectory) toGroupInvokers(newInvokersMap *sync.Map) []protoc
return groupInvokersList
}
-func (dir *RegistryDirectory) uncacheInvoker(url common.URL) *sync.Map {
+func (dir *registryDirectory) uncacheInvoker(url common.URL) {
log.Debug("service will be deleted in cache invokers: invokers key is %s!", url.Key())
- newCacheInvokers := dir.cacheInvokersMap
- newCacheInvokers.Delete(url.Key())
- return newCacheInvokers
+ dir.cacheInvokersMap.Delete(url.Key())
}
-func (dir *RegistryDirectory) cacheInvoker(url common.URL) *sync.Map {
+func (dir *registryDirectory) cacheInvoker(url common.URL) {
referenceUrl := dir.GetUrl().SubURL
- newCacheInvokers := dir.cacheInvokersMap
//check the url's protocol is equal to the protocol which is configured in reference config or referenceUrl is not care about protocol
if url.Protocol == referenceUrl.Protocol || referenceUrl.Protocol == "" {
url = mergeUrl(url, referenceUrl)
- if _, ok := newCacheInvokers.Load(url.Key()); !ok {
+ if _, ok := dir.cacheInvokersMap.Load(url.Key()); !ok {
log.Debug("service will be added in cache invokers: invokers key is %s!", url.Key())
newInvoker := extension.GetProtocolExtension(protocolwrapper.FILTER).Refer(url)
if newInvoker != nil {
- newCacheInvokers.Store(url.Key(), newInvoker)
+ dir.cacheInvokersMap.Store(url.Key(), newInvoker)
}
}
}
- return newCacheInvokers
}
//select the protocol invokers from the directory
-func (dir *RegistryDirectory) List(invocation protocol.Invocation) []protocol.Invoker {
+func (dir *registryDirectory) List(invocation protocol.Invocation) []protocol.Invoker {
//TODO:router
return dir.cacheInvokers
}
-func (dir *RegistryDirectory) IsAvailable() bool {
+func (dir *registryDirectory) IsAvailable() bool {
return dir.BaseDirectory.IsAvailable()
}
-func (dir *RegistryDirectory) Destroy() {
+func (dir *registryDirectory) Destroy() {
//TODO:unregister & unsubscribe
dir.BaseDirectory.Destroy(func() {
for _, ivk := range dir.cacheInvokers {
diff --git a/registry/directory/directory_test.go b/registry/directory/directory_test.go
index 10366ffd2..9d9e2b94c 100644
--- a/registry/directory/directory_test.go
+++ b/registry/directory/directory_test.go
@@ -98,7 +98,7 @@ func Test_List(t *testing.T) {
}
-func normalRegistryDir() (*RegistryDirectory, *registry.MockRegistry) {
+func normalRegistryDir() (*registryDirectory, *registry.MockRegistry) {
extension.SetProtocol(protocolwrapper.FILTER, protocolwrapper.NewMockProtocolFilter)
url, _ := common.NewURL(context.TODO(), "mock://127.0.0.1:1111")
--
GitLab