diff --git a/common/constant/key.go b/common/constant/key.go
index 1479af2305b1ce4280c2aa7f4016e314ac358513..da21a3a9e1254d5a22d670a11c5c01022892e096 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -26,6 +26,7 @@ const (
VERSION_KEY = "version"
INTERFACE_KEY = "interface"
PATH_KEY = "path"
+ PROTOCOL_KEY = "protocol"
SERVICE_KEY = "service"
METHODS_KEY = "methods"
TIMEOUT_KEY = "timeout"
@@ -40,6 +41,7 @@ const (
TOKEN_KEY = "token"
LOCAL_ADDR = "local-addr"
REMOTE_ADDR = "remote-addr"
+ PATH_SEPARATOR = "/"
DUBBO_KEY = "dubbo"
RELEASE_KEY = "release"
ANYHOST_KEY = "anyhost"
diff --git a/common/extension/event_dispatcher.go b/common/extension/event_dispatcher.go
new file mode 100644
index 0000000000000000000000000000000000000000..2d33528259e85f50a0784aceba37ecd87be61620
--- /dev/null
+++ b/common/extension/event_dispatcher.go
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package extension
+
+import (
+ "github.com/apache/dubbo-go/common/logger"
+ "github.com/apache/dubbo-go/common/observer"
+)
+
+var (
+ globalEventDispatcher observer.EventDispatcher
+ initEventListeners []observer.EventListener
+)
+
+var (
+ dispatchers = make(map[string]func() observer.EventDispatcher, 8)
+)
+
+// SetEventDispatcher by name
+func SetEventDispatcher(name string, v func() observer.EventDispatcher) {
+ dispatchers[name] = v
+}
+
+// SetAndInitGlobalDispatcher
+func SetAndInitGlobalDispatcher(name string) {
+ if len(name) == 0 {
+ name = "direct"
+ }
+ if globalEventDispatcher != nil {
+ logger.Warnf("EventDispatcher already init. It will be replaced")
+ }
+ if dp, ok := dispatchers[name]; !ok || dp == nil {
+ panic("EventDispatcher for " + name + " is not existing, make sure you have import the package.")
+ }
+ globalEventDispatcher = dispatchers[name]()
+ globalEventDispatcher.AddEventListeners(initEventListeners)
+}
+
+// GetGlobalDispatcher
+func GetGlobalDispatcher() observer.EventDispatcher {
+ return globalEventDispatcher
+}
+
+// AddEventListener it will be added in global event dispatcher
+func AddEventListener(listener observer.EventListener) {
+ initEventListeners = append(initEventListeners, listener)
+}
diff --git a/common/extension/metadata_report_factory.go b/common/extension/metadata_report_factory.go
index 0ae0793bb4459767cb42fb1860fc484388aae1a3..89dab0409987d8959b5761771c8893abaf88673b 100644
--- a/common/extension/metadata_report_factory.go
+++ b/common/extension/metadata_report_factory.go
@@ -18,20 +18,20 @@
package extension
import (
- "github.com/apache/dubbo-go/metadata"
+ "github.com/apache/dubbo-go/metadata/report/factory"
)
var (
- metaDataReportFactories = make(map[string]func() metadata.MetadataReportFactory, 8)
+ metaDataReportFactories = make(map[string]func() factory.MetadataReportFactory, 8)
)
// SetMetadataReportFactory ...
-func SetMetadataReportFactory(name string, v func() metadata.MetadataReportFactory) {
+func SetMetadataReportFactory(name string, v func() factory.MetadataReportFactory) {
metaDataReportFactories[name] = v
}
// GetMetadataReportFactory ...
-func GetMetadataReportFactory(name string) metadata.MetadataReportFactory {
+func GetMetadataReportFactory(name string) factory.MetadataReportFactory {
if metaDataReportFactories[name] == nil {
panic("metadata report for " + name + " is not existing, make sure you have import the package.")
}
diff --git a/common/observer/dispatcher/direct_event_dispatcher.go b/common/observer/dispatcher/direct_event_dispatcher.go
new file mode 100644
index 0000000000000000000000000000000000000000..2b7567b47ed40caa8867901ff0a05e0a2497cd31
--- /dev/null
+++ b/common/observer/dispatcher/direct_event_dispatcher.go
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dispatcher
+
+import (
+ "reflect"
+)
+
+import (
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/common/logger"
+ "github.com/apache/dubbo-go/common/observer"
+)
+
+func init() {
+ extension.SetEventDispatcher("direct", NewDirectEventDispatcher)
+}
+
+// DirectEventDispatcher is align with DirectEventDispatcher interface in Java.
+// it's the top abstraction
+// Align with 2.7.5
+// Dispatcher event to listener direct
+type DirectEventDispatcher struct {
+ observer.BaseListenable
+}
+
+// NewDirectEventDispatcher ac constructor of DirectEventDispatcher
+func NewDirectEventDispatcher() observer.EventDispatcher {
+ return &DirectEventDispatcher{}
+}
+
+// Dispatch event directly
+func (ded *DirectEventDispatcher) Dispatch(event observer.Event) {
+ if event == nil {
+ logger.Warnf("[DirectEventDispatcher] dispatch event nil")
+ return
+ }
+ eventType := reflect.TypeOf(event).Elem()
+ value, loaded := ded.ListenersCache.Load(eventType)
+ if !loaded {
+ return
+ }
+ listenersSlice := value.([]observer.EventListener)
+ for _, listener := range listenersSlice {
+ if err := listener.OnEvent(event); err != nil {
+ logger.Warnf("[DirectEventDispatcher] dispatch event error:%v", err)
+ }
+ }
+}
diff --git a/common/observer/dispatcher/direct_event_dispatcher_test.go b/common/observer/dispatcher/direct_event_dispatcher_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..355c930a9e86dbe6e82adc2795f8590c15a473c2
--- /dev/null
+++ b/common/observer/dispatcher/direct_event_dispatcher_test.go
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dispatcher
+
+import (
+ "fmt"
+ "reflect"
+ "testing"
+)
+
+import (
+ "github.com/apache/dubbo-go/common/observer"
+)
+
+func TestDirectEventDispatcher_Dispatch(t *testing.T) {
+ ded := NewDirectEventDispatcher()
+ ded.AddEventListener(&TestEventListener{})
+ ded.AddEventListener(&TestEventListener1{})
+ ded.Dispatch(&TestEvent{})
+ ded.Dispatch(nil)
+}
+
+type TestEvent struct {
+ observer.BaseEvent
+}
+
+type TestEventListener struct {
+ observer.BaseListenable
+ observer.EventListener
+}
+
+func (tel *TestEventListener) OnEvent(e observer.Event) error {
+ fmt.Println("TestEventListener")
+ return nil
+}
+
+func (tel *TestEventListener) GetPriority() int {
+ return -1
+}
+
+func (tel *TestEventListener) GetEventType() reflect.Type {
+ return reflect.TypeOf(&TestEvent{})
+}
+
+type TestEventListener1 struct {
+ observer.EventListener
+}
+
+func (tel *TestEventListener1) OnEvent(e observer.Event) error {
+ fmt.Println("TestEventListener1")
+ return nil
+}
+
+func (tel *TestEventListener1) GetPriority() int {
+ return 1
+}
+
+func (tel *TestEventListener1) GetEventType() reflect.Type {
+ return reflect.TypeOf(TestEvent{})
+}
diff --git a/common/observer/event.go b/common/observer/event.go
new file mode 100644
index 0000000000000000000000000000000000000000..8c3362feeee3d62315eb734460f486dcdbfe2f36
--- /dev/null
+++ b/common/observer/event.go
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package observer
+
+import (
+ "fmt"
+ "math/rand"
+ "time"
+)
+
+func init() {
+ rand.Seed(time.Now().UnixNano())
+}
+
+// Event is align with Event interface in Java.
+// it's the top abstraction
+// Align with 2.7.5
+type Event interface {
+ fmt.Stringer
+ GetSource() interface{}
+ GetTimestamp() time.Time
+}
+
+// BaseEvent is the base implementation of Event
+// You should never use it directly
+type BaseEvent struct {
+ Source interface{}
+ Timestamp time.Time
+}
+
+// GetSource return the source
+func (b *BaseEvent) GetSource() interface{} {
+ return b.Source
+}
+
+// GetTimestamp return the Timestamp when the event is created
+func (b *BaseEvent) GetTimestamp() time.Time {
+ return b.Timestamp
+}
+
+// String return a human readable string representing this event
+func (b *BaseEvent) String() string {
+ return fmt.Sprintf("BaseEvent[source = %#v]", b.Source)
+}
+
+func newBaseEvent(source interface{}) *BaseEvent {
+ return &BaseEvent{
+ Source: source,
+ Timestamp: time.Now(),
+ }
+}
diff --git a/common/observer/event_dispatcher.go b/common/observer/event_dispatcher.go
new file mode 100644
index 0000000000000000000000000000000000000000..17745e68c07065f954d456914fb276fadf6d975d
--- /dev/null
+++ b/common/observer/event_dispatcher.go
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package observer
+
+// EventDispatcher is align with EventDispatcher interface in Java.
+// it's the top abstraction
+// Align with 2.7.5
+type EventDispatcher interface {
+ Listenable
+ // Dispatch event
+ Dispatch(event Event)
+}
diff --git a/common/observer/event_listener.go b/common/observer/event_listener.go
new file mode 100644
index 0000000000000000000000000000000000000000..8db60d8475da49262947329fc71fd8e364d8d0af
--- /dev/null
+++ b/common/observer/event_listener.go
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package observer
+
+import (
+ "reflect"
+)
+
+import (
+ gxsort "github.com/dubbogo/gost/sort"
+)
+
+// EventListener is an new interface used to align with dubbo 2.7.5
+// It contains the Prioritized means that the listener has its priority
+type EventListener interface {
+ gxsort.Prioritizer
+ // OnEvent handle this event
+ OnEvent(e Event) error
+ // GetEventType listen which event type
+ GetEventType() reflect.Type
+}
+
+// ConditionalEventListener only handle the event which it can handle
+type ConditionalEventListener interface {
+ EventListener
+ // Accept will make the decision whether it should handle this event
+ Accept(e Event) bool
+}
+
+// TODO (implement ConditionalEventListener)
+type ServiceInstancesChangedListener struct {
+ ServiceName string
+}
diff --git a/common/observer/listenable.go b/common/observer/listenable.go
new file mode 100644
index 0000000000000000000000000000000000000000..7b64aa8f2d263793c7aa4dc5e294466c48cd7b36
--- /dev/null
+++ b/common/observer/listenable.go
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package observer
+
+import (
+ "reflect"
+ "sort"
+ "sync"
+)
+
+// Listenable could add and remove the event listener
+type Listenable interface {
+ AddEventListener(listener EventListener)
+ AddEventListeners(listenersSlice []EventListener)
+ RemoveEventListener(listener EventListener)
+ RemoveEventListeners(listenersSlice []EventListener)
+ GetAllEventListeners() []EventListener
+ RemoveAllEventListeners()
+}
+
+// BaseListenable base listenable
+type BaseListenable struct {
+ Listenable
+ ListenersCache sync.Map
+ Mutex sync.Mutex
+}
+
+// NewBaseListenable a constructor of base listenable
+func NewBaseListenable() Listenable {
+ return &BaseListenable{}
+}
+
+// AddEventListener add event listener
+func (bl *BaseListenable) AddEventListener(listener EventListener) {
+ eventType := listener.GetEventType()
+ if eventType.Kind() == reflect.Ptr {
+ eventType = eventType.Elem()
+ }
+ bl.Mutex.Lock()
+ defer bl.Mutex.Unlock()
+ value, loaded := bl.ListenersCache.LoadOrStore(eventType, make([]EventListener, 0, 8))
+ listenersSlice := value.([]EventListener)
+ // return if listenersSlice already has this listener
+ if loaded && containListener(listenersSlice, listener) {
+ return
+ }
+ listenersSlice = append(listenersSlice, listener)
+ sort.Slice(listenersSlice, func(i, j int) bool {
+ return listenersSlice[i].GetPriority() < listenersSlice[j].GetPriority()
+ })
+ bl.ListenersCache.Store(eventType, listenersSlice)
+}
+
+// AddEventListeners add the slice of event listener
+func (bl *BaseListenable) AddEventListeners(listenersSlice []EventListener) {
+ for _, listener := range listenersSlice {
+ bl.AddEventListener(listener)
+ }
+}
+
+// RemoveEventListener remove the event listener
+func (bl *BaseListenable) RemoveEventListener(listener EventListener) {
+ eventType := listener.GetEventType()
+ if eventType.Kind() == reflect.Ptr {
+ eventType = eventType.Elem()
+ }
+ bl.Mutex.Lock()
+ defer bl.Mutex.Unlock()
+ value, loaded := bl.ListenersCache.Load(eventType)
+ if !loaded {
+ return
+ }
+ listenersSlice := value.([]EventListener)
+ for i, l := range listenersSlice {
+ if l == listener {
+ listenersSlice = append(listenersSlice[:i], listenersSlice[i+1:]...)
+ }
+ }
+ bl.ListenersCache.Store(eventType, listenersSlice)
+}
+
+// RemoveEventListeners remove the slice of event listener
+func (bl *BaseListenable) RemoveEventListeners(listenersSlice []EventListener) {
+ for _, listener := range listenersSlice {
+ bl.RemoveEventListener(listener)
+ }
+}
+
+// RemoveAllEventListeners remove all
+func (bl *BaseListenable) RemoveAllEventListeners() {
+ bl.Mutex.Lock()
+ defer bl.Mutex.Unlock()
+ bl.ListenersCache = sync.Map{}
+}
+
+// GetAllEventListeners get all
+func (bl *BaseListenable) GetAllEventListeners() []EventListener {
+ allListenersSlice := make([]EventListener, 0, 16)
+ bl.ListenersCache.Range(func(_, value interface{}) bool {
+ listenersSlice := value.([]EventListener)
+ allListenersSlice = append(allListenersSlice, listenersSlice...)
+ return true
+ })
+ sort.Slice(allListenersSlice, func(i, j int) bool {
+ return allListenersSlice[i].GetPriority() < allListenersSlice[j].GetPriority()
+ })
+ return allListenersSlice
+}
+
+// containListener true if contain listener
+func containListener(listenersSlice []EventListener, listener EventListener) bool {
+ for _, loadListener := range listenersSlice {
+ if loadListener == listener {
+ return true
+ }
+ }
+ return false
+}
diff --git a/common/observer/listenable_test.go b/common/observer/listenable_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..df46bfc2ba47f6e447074b44208a809949f7ae3d
--- /dev/null
+++ b/common/observer/listenable_test.go
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package observer
+
+import (
+ "reflect"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+func TestListenable(t *testing.T) {
+ el := &TestEventListener{}
+ b := &BaseListenable{}
+ b.AddEventListener(el)
+ b.AddEventListener(el)
+ al := b.GetAllEventListeners()
+ assert.Equal(t, len(al), 1)
+ assert.Equal(t, al[0].GetEventType(), reflect.TypeOf(TestEvent{}))
+ b.RemoveEventListener(el)
+ assert.Equal(t, len(b.GetAllEventListeners()), 0)
+ var ts []EventListener
+ ts = append(ts, el)
+ b.AddEventListeners(ts)
+ assert.Equal(t, len(al), 1)
+
+}
+
+type TestEvent struct {
+ BaseEvent
+}
+
+type TestEventListener struct {
+ EventListener
+}
+
+func (tel *TestEventListener) OnEvent(e Event) error {
+ return nil
+}
+
+func (tel *TestEventListener) GetPriority() int {
+ return -1
+}
+
+func (tel *TestEventListener) GetEventType() reflect.Type {
+ return reflect.TypeOf(TestEvent{})
+}
diff --git a/common/rpc_service.go b/common/rpc_service.go
index ebd1d02f843bc339c3a37d977e2138798307475d..3a25d322391bf3ee52fabadd511677a52dcb169a 100644
--- a/common/rpc_service.go
+++ b/common/rpc_service.go
@@ -133,6 +133,11 @@ func (s *Service) Method() map[string]*MethodType {
return s.methods
}
+// Name will return service name
+func (s *Service) Name() string {
+ return s.name
+}
+
// RcvrType ...
func (s *Service) RcvrType() reflect.Type {
return s.rcvrType
diff --git a/config/base_config.go b/config/base_config.go
index 93c0ce6a6692193e7ea7b1b9f2f74e9eaed0c858..f58138d2e58a1b3da06894d3afa4728ea2ecf8fb 100644
--- a/config/base_config.go
+++ b/config/base_config.go
@@ -42,14 +42,13 @@ type multiConfiger interface {
// BaseConfig is the common configuration for provider and consumer
type BaseConfig struct {
- ConfigCenterConfig *ConfigCenterConfig `yaml:"config_center" json:"config_center,omitempty"`
- configCenterUrl *common.URL
- prefix string
- fatherConfig interface{}
-
- MetricConfig *MetricConfig `yaml:"metrics" json:"metrics,omitempty"`
-
- fileStream *bytes.Buffer
+ ConfigCenterConfig *ConfigCenterConfig `yaml:"config_center" json:"config_center,omitempty"`
+ configCenterUrl *common.URL
+ prefix string
+ fatherConfig interface{}
+ eventDispatcherType string `default:"direct" yaml:"event_dispatcher_type" json:"event_dispatcher_type,omitempty"`
+ MetricConfig *MetricConfig `yaml:"metrics" json:"metrics,omitempty"`
+ fileStream *bytes.Buffer
}
// startConfigCenter will start the config center.
diff --git a/config/base_config_test.go b/config/base_config_test.go
index d16b2420922ece60ef2135729cd47d5aa73a3760..60eccfb1836dccbec0e8dc593a0954005117c28e 100644
--- a/config/base_config_test.go
+++ b/config/base_config_test.go
@@ -53,15 +53,6 @@ func Test_refresh(t *testing.T) {
Owner: "dubbo",
Environment: "test"},
Registries: map[string]*RegistryConfig{
- //"shanghai_reg1": {
- // id: "shanghai_reg1",
- // Protocol: "mock",
- // TimeoutStr: "2s",
- // Group: "shanghai_idc",
- // Address: "127.0.0.1:2181",
- // Username: "user1",
- // Password: "pwd1",
- //},
"shanghai_reg2": {
Protocol: "mock",
TimeoutStr: "2s",
@@ -156,15 +147,6 @@ func Test_appExternal_refresh(t *testing.T) {
Owner: "dubbo",
Environment: "test"},
Registries: map[string]*RegistryConfig{
- //"shanghai_reg1": {
- // id: "shanghai_reg1",
- // Protocol: "mock",
- // TimeoutStr: "2s",
- // Group: "shanghai_idc",
- // Address: "127.0.0.1:2181",
- // Username: "user1",
- // Password: "pwd1",
- //},
"shanghai_reg2": {
Protocol: "mock",
TimeoutStr: "2s",
@@ -251,15 +233,6 @@ func Test_appExternalWithoutId_refresh(t *testing.T) {
Owner: "dubbo",
Environment: "test"},
Registries: map[string]*RegistryConfig{
- //"shanghai_reg1": {
- // id: "shanghai_reg1",
- // Protocol: "mock",
- // TimeoutStr: "2s",
- // Group: "shanghai_idc",
- // Address: "127.0.0.1:2181",
- // Username: "user1",
- // Password: "pwd1",
- //},
"shanghai_reg2": {
Protocol: "mock",
TimeoutStr: "2s",
@@ -408,15 +381,6 @@ func Test_refreshProvider(t *testing.T) {
Owner: "dubbo",
Environment: "test"},
Registries: map[string]*RegistryConfig{
- //"shanghai_reg1": {
- // id: "shanghai_reg1",
- // Protocol: "mock",
- // TimeoutStr: "2s",
- // Group: "shanghai_idc",
- // Address: "127.0.0.1:2181",
- // Username: "user1",
- // Password: "pwd1",
- //},
"shanghai_reg2": {
Protocol: "mock",
TimeoutStr: "2s",
diff --git a/config/config_loader.go b/config/config_loader.go
index 61cb49457b7a2435cb22f02a9df0a02a71ae68cb..6666409dd3f6e9eabb9b858e7691b44c3f9041bd 100644
--- a/config/config_loader.go
+++ b/config/config_loader.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/dubbo-go/common/constant"
"github.com/apache/dubbo-go/common/extension"
"github.com/apache/dubbo-go/common/logger"
+ _ "github.com/apache/dubbo-go/common/observer/dispatcher"
)
var (
@@ -91,6 +92,17 @@ func Load() {
}
}
+ var eventDispatcherType string
+ if consumerConfig != nil {
+ eventDispatcherType = consumerConfig.eventDispatcherType
+ }
+ // notice consumerConfig.eventDispatcherType will be replaced
+ if providerConfig != nil {
+ eventDispatcherType = providerConfig.eventDispatcherType
+ }
+ // init EventDispatcher should before everything
+ extension.SetAndInitGlobalDispatcher(eventDispatcherType)
+
// reference config
if consumerConfig == nil {
logger.Warnf("consumerConfig is nil!")
@@ -198,6 +210,7 @@ func Load() {
}
svs.id = key
svs.Implement(rpcService)
+ svs.Protocols = providerConfig.Protocols
if err := svs.Export(); err != nil {
panic(fmt.Sprintf("service %s export failed! err: %#v", key, err))
}
diff --git a/config/config_loader_test.go b/config/config_loader_test.go
index 6368fcbd2c7bc675231e7b7835750f26743708af..0192b4c8a06263266cb80b344a0792ea2f6af8c8 100644
--- a/config/config_loader_test.go
+++ b/config/config_loader_test.go
@@ -24,6 +24,7 @@ import (
import (
"github.com/stretchr/testify/assert"
+ "go.uber.org/atomic"
)
import (
@@ -90,7 +91,7 @@ func TestLoad(t *testing.T) {
func TestLoadWithSingleReg(t *testing.T) {
doInitConsumerWithSingleRegistry()
- doInitProviderWithSingleRegistry()
+ mockInitProviderWithSingleRegistry()
ms := &MockService{}
SetConsumerService(ms)
@@ -233,3 +234,55 @@ func TestConfigLoaderWithConfigCenterSingleRegistry(t *testing.T) {
assert.Equal(t, "mock://127.0.0.1:2182", consumerConfig.Registries[constant.DEFAULT_KEY].Address)
}
+
+// mockInitProviderWithSingleRegistry will init a mocked providerConfig
+func mockInitProviderWithSingleRegistry() {
+ providerConfig = &ProviderConfig{
+ ApplicationConfig: &ApplicationConfig{
+ Organization: "dubbo_org",
+ Name: "dubbo",
+ Module: "module",
+ Version: "1.0.0",
+ Owner: "dubbo",
+ Environment: "test"},
+ Registry: &RegistryConfig{
+ Address: "mock://127.0.0.1:2181",
+ Username: "user1",
+ Password: "pwd1",
+ },
+ Registries: map[string]*RegistryConfig{},
+ Services: map[string]*ServiceConfig{
+ "MockService": {
+ InterfaceName: "com.MockService",
+ Protocol: "mock",
+ Cluster: "failover",
+ Loadbalance: "random",
+ Retries: "3",
+ Group: "huadong_idc",
+ Version: "1.0.0",
+ Methods: []*MethodConfig{
+ {
+ Name: "GetUser",
+ Retries: "2",
+ Loadbalance: "random",
+ Weight: 200,
+ },
+ {
+ Name: "GetUser1",
+ Retries: "2",
+ Loadbalance: "random",
+ Weight: 200,
+ },
+ },
+ exported: new(atomic.Bool),
+ },
+ },
+ Protocols: map[string]*ProtocolConfig{
+ "mock": {
+ Name: "mock",
+ Ip: "127.0.0.1",
+ Port: "20000",
+ },
+ },
+ }
+}
diff --git a/config/instance/metedata_report.go b/config/instance/metedata_report.go
index cd54b0a7940df166c88f02234ab1a4e3bf384163..9cf435bc9debf09d0707d0a3b5d699c79d2cafa7 100644
--- a/config/instance/metedata_report.go
+++ b/config/instance/metedata_report.go
@@ -24,16 +24,16 @@ import (
import (
"github.com/apache/dubbo-go/common"
"github.com/apache/dubbo-go/common/extension"
- "github.com/apache/dubbo-go/metadata"
+ "github.com/apache/dubbo-go/metadata/report"
)
var (
- instance metadata.MetadataReport
+ instance report.MetadataReport
once sync.Once
)
// GetMetadataReportInstance ...
-func GetMetadataReportInstance(url *common.URL) metadata.MetadataReport {
+func GetMetadataReportInstance(url *common.URL) report.MetadataReport {
once.Do(func() {
instance = extension.GetMetadataReportFactory(url.Protocol).CreateMetadataReport(url)
})
diff --git a/config/metadata_report_config_test.go b/config/metadata_report_config_test.go
index d6b08d5fb0c51495940d4dc021a0796c1d577923..635feecc2d433366534566d184e058eb54a881ed 100644
--- a/config/metadata_report_config_test.go
+++ b/config/metadata_report_config_test.go
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package config
import "testing"
diff --git a/config/service_config.go b/config/service_config.go
index 50bf5e12c3247340f177a84c72446383ec5c3450..9cf60c8fb30f74e78f44e736604394fda779eb88 100644
--- a/config/service_config.go
+++ b/config/service_config.go
@@ -71,11 +71,15 @@ type ServiceConfig struct {
ParamSign string `yaml:"param.sign" json:"param.sign,omitempty" property:"param.sign"`
Tag string `yaml:"tag" json:"tag,omitempty" property:"tag"`
+ Protocols map[string]*ProtocolConfig
unexported *atomic.Bool
exported *atomic.Bool
rpcService common.RPCService
- cacheProtocol protocol.Protocol
cacheMutex sync.Mutex
+ cacheProtocol protocol.Protocol
+
+ exportersLock sync.Mutex
+ exporters []protocol.Exporter
}
// Prefix ...
@@ -92,6 +96,8 @@ func (c *ServiceConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {
if err := unmarshal((*plain)(c)); err != nil {
return err
}
+ c.exported = atomic.NewBool(false)
+ c.unexported = atomic.NewBool(false)
return nil
}
@@ -105,6 +111,16 @@ func NewServiceConfig(id string, context context.Context) *ServiceConfig {
}
}
+// InitExported will set exported as false atom bool
+func (c *ServiceConfig) InitExported() {
+ c.exported = atomic.NewBool(false)
+}
+
+// IsExport will return whether the service config is exported or not
+func (c *ServiceConfig) IsExport() bool {
+ return c.exported.Load()
+}
+
// Export ...
func (c *ServiceConfig) Export() error {
// TODO: config center start here
@@ -122,7 +138,7 @@ func (c *ServiceConfig) Export() error {
regUrls := loadRegistries(c.Registry, providerConfig.Registries, common.PROVIDER)
urlMap := c.getUrlMap()
- protocolConfigs := loadProtocol(c.Protocol, providerConfig.Protocols)
+ protocolConfigs := loadProtocol(c.Protocol, c.Protocols)
if len(protocolConfigs) == 0 {
logger.Warnf("The service %v's '%v' protocols don't has right protocolConfigs ", c.InterfaceName, c.Protocol)
return nil
@@ -148,6 +164,9 @@ func (c *ServiceConfig) Export() error {
if len(c.Tag) > 0 {
ivkURL.AddParam(constant.Tagkey, c.Tag)
}
+
+ var exporter protocol.Exporter
+
if len(regUrls) > 0 {
for _, regUrl := range regUrls {
regUrl.SubURL = ivkURL
@@ -160,22 +179,46 @@ func (c *ServiceConfig) Export() error {
c.cacheMutex.Unlock()
invoker := extension.GetProxyFactory(providerConfig.ProxyFactory).GetInvoker(*regUrl)
- exporter := c.cacheProtocol.Export(invoker)
+ exporter = c.cacheProtocol.Export(invoker)
if exporter == nil {
panic(perrors.New(fmt.Sprintf("Registry protocol new exporter error,registry is {%v},url is {%v}", regUrl, ivkURL)))
}
}
} else {
invoker := extension.GetProxyFactory(providerConfig.ProxyFactory).GetInvoker(*ivkURL)
- exporter := extension.GetProtocol(protocolwrapper.FILTER).Export(invoker)
+ exporter = extension.GetProtocol(protocolwrapper.FILTER).Export(invoker)
if exporter == nil {
panic(perrors.New(fmt.Sprintf("Filter protocol without registry new exporter error,url is {%v}", ivkURL)))
}
}
+ c.exporters = append(c.exporters, exporter)
}
+ c.exported.Store(true)
return nil
}
+// Unexport will call unexport of all exporters service config exported
+func (c *ServiceConfig) Unexport() {
+ if !c.exported.Load() {
+ return
+ }
+ if c.unexported.Load() {
+ return
+ }
+
+ func() {
+ c.exportersLock.Lock()
+ defer c.exportersLock.Unlock()
+ for _, exporter := range c.exporters {
+ exporter.Unexport()
+ }
+ c.exporters = nil
+ }()
+
+ c.exported.Store(false)
+ c.unexported.Store(true)
+}
+
// Implement ...
func (c *ServiceConfig) Implement(s common.RPCService) {
c.rpcService = s
@@ -245,3 +288,16 @@ func (c *ServiceConfig) getUrlMap() url.Values {
return urlMap
}
+
+// GetExportedUrls will return the url in service config's exporter
+func (c *ServiceConfig) GetExportedUrls() []*common.URL {
+ if c.exported.Load() {
+ var urls []*common.URL
+ for _, exporter := range c.exporters {
+ url := exporter.GetInvoker().GetUrl()
+ urls = append(urls, &url)
+ }
+ return urls
+ }
+ return nil
+}
diff --git a/config/service_config_test.go b/config/service_config_test.go
index 6f3230890348e77ea26c9c0eaf9165090c8cd09f..fb3b4a7dc20e32fbcaf6f9514991fba9a7af0d95 100644
--- a/config/service_config_test.go
+++ b/config/service_config_test.go
@@ -21,6 +21,10 @@ import (
"testing"
)
+import (
+ "go.uber.org/atomic"
+)
+
import (
"github.com/apache/dubbo-go/common/extension"
)
@@ -92,6 +96,7 @@ func doInitProvider() {
Weight: 200,
},
},
+ exported: new(atomic.Bool),
},
"MockServiceNoRightProtocol": {
InterfaceName: "com.MockService",
@@ -116,56 +121,7 @@ func doInitProvider() {
Weight: 200,
},
},
- },
- },
- Protocols: map[string]*ProtocolConfig{
- "mock": {
- Name: "mock",
- Ip: "127.0.0.1",
- Port: "20000",
- },
- },
- }
-}
-
-func doInitProviderWithSingleRegistry() {
- providerConfig = &ProviderConfig{
- ApplicationConfig: &ApplicationConfig{
- Organization: "dubbo_org",
- Name: "dubbo",
- Module: "module",
- Version: "2.6.0",
- Owner: "dubbo",
- Environment: "test"},
- Registry: &RegistryConfig{
- Address: "mock://127.0.0.1:2181",
- Username: "user1",
- Password: "pwd1",
- },
- Registries: map[string]*RegistryConfig{},
- Services: map[string]*ServiceConfig{
- "MockService": {
- InterfaceName: "com.MockService",
- Protocol: "mock",
- Cluster: "failover",
- Loadbalance: "random",
- Retries: "3",
- Group: "huadong_idc",
- Version: "1.0.0",
- Methods: []*MethodConfig{
- {
- Name: "GetUser",
- Retries: "2",
- Loadbalance: "random",
- Weight: 200,
- },
- {
- Name: "GetUser1",
- Retries: "2",
- Loadbalance: "random",
- Weight: 200,
- },
- },
+ exported: new(atomic.Bool),
},
},
Protocols: map[string]*ProtocolConfig{
diff --git a/go.mod b/go.mod
index 54d532eac06a56057f815e3c8e91fdd267c9c6ad..fe1891ea6e70ee3f8c605243e4d464d4b22dc73f 100644
--- a/go.mod
+++ b/go.mod
@@ -1,7 +1,7 @@
module github.com/apache/dubbo-go
require (
- github.com/Workiva/go-datastructures v1.0.50
+ github.com/Workiva/go-datastructures v1.0.52
github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5
github.com/aliyun/alibaba-cloud-sdk-go v0.0.0-20190802083043-4cd0c391755e // indirect
github.com/apache/dubbo-go-hessian2 v1.4.0
diff --git a/go.sum b/go.sum
index e499992eb0a0335ca0a1e1f746caca3418af7655..73c3da87a0bb784ef17fed34687a024cb18b7f22 100644
--- a/go.sum
+++ b/go.sum
@@ -27,8 +27,8 @@ github.com/SermoDigital/jose v0.0.0-20180104203859-803625baeddc h1:LkkwnbY+S8Wmw
github.com/SermoDigital/jose v0.0.0-20180104203859-803625baeddc/go.mod h1:ARgCUhI1MHQH+ONky/PAtmVHQrP5JlGY0F3poXOp/fA=
github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 h1:fLjPD/aNc3UIOA6tDi6QXUemppXK3P9BI7mr2hd6gx8=
github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
-github.com/Workiva/go-datastructures v1.0.50 h1:slDmfW6KCHcC7U+LP3DDBbm4fqTwZGn1beOFPfGaLvo=
-github.com/Workiva/go-datastructures v1.0.50/go.mod h1:Z+F2Rca0qCsVYDS8z7bAGm8f3UkzuWYS/oBZz5a7VVA=
+github.com/Workiva/go-datastructures v1.0.52 h1:PLSK6pwn8mYdaoaCZEMsXBpBotr4HHn9abU0yMQt0NI=
+github.com/Workiva/go-datastructures v1.0.52/go.mod h1:Z+F2Rca0qCsVYDS8z7bAGm8f3UkzuWYS/oBZz5a7VVA=
github.com/abdullin/seq v0.0.0-20160510034733-d5467c17e7af h1:DBNMBMuMiWYu0b+8KMJuWmfCkcxl09JwdlqwDZZ6U14=
github.com/abdullin/seq v0.0.0-20160510034733-d5467c17e7af/go.mod h1:5Jv4cbFiHJMsVxt52+i0Ha45fjshj6wxYr1r19tB9bw=
github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5 h1:rFw4nCn9iMW+Vajsk51NtYIcwSTkXr+JGrMd36kTDJw=
@@ -111,10 +111,7 @@ github.com/dubbogo/getty v1.3.3/go.mod h1:U92BDyJ6sW9Jpohr2Vlz8w2uUbIbNZ3d+6rJvF
github.com/dubbogo/go-zookeeper v1.0.0 h1:RsYdlGwhDW+iKXM3eIIcvt34P2swLdmQfuIJxsHlGoM=
github.com/dubbogo/go-zookeeper v1.0.0/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c=
github.com/dubbogo/gost v1.5.1/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
-github.com/dubbogo/gost v1.5.2 h1:ri/03971hdpnn3QeCU+4UZgnRNGDXLDGDucR/iozZm8=
github.com/dubbogo/gost v1.5.2/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
-github.com/dubbogo/gost v1.7.0 h1:lWNBIE2hk1Aj2be2uXkyRTpZG0RQZj0/xbXnkIq6EHE=
-github.com/dubbogo/gost v1.7.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
github.com/dubbogo/gost v1.8.0 h1:9ACbQe5OwMjqtinQcNJC5xp16kky27OsfSGw5L9A6vw=
github.com/dubbogo/gost v1.8.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
github.com/duosecurity/duo_api_golang v0.0.0-20190308151101-6c680f768e74 h1:2MIhn2R6oXQbgW5yHfS+d6YqyMfXiu2L55rFZC4UD/M=
@@ -385,8 +382,6 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/nacos-group/nacos-sdk-go v0.0.0-20190723125407-0242d42e3dbb h1:lbmvw8r9W55w+aQgWn35W1nuleRIECMoqUrmwAOAvoI=
-github.com/nacos-group/nacos-sdk-go v0.0.0-20190723125407-0242d42e3dbb/go.mod h1:CEkSvEpoveoYjA81m4HNeYQ0sge0LFGKSEqO3JKHllo=
github.com/nacos-group/nacos-sdk-go v0.0.0-20191128082542-fe1b325b125c h1:WoCa3AvgQMVKNs+RIFlWPRgY9QVJwUxJDrGxHs0fcRo=
github.com/nacos-group/nacos-sdk-go v0.0.0-20191128082542-fe1b325b125c/go.mod h1:CEkSvEpoveoYjA81m4HNeYQ0sge0LFGKSEqO3JKHllo=
github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2 h1:BQ1HW7hr4IVovMwWg0E0PYcyW8CzqDcVmaew9cujU4s=
diff --git a/metadata/definition/definition.go b/metadata/definition/definition.go
index ead984345efde1ddd1d54b7599fd9d5584947ea2..8d4a584ee53c7c5ebbc4cc2222134ca9e957068f 100644
--- a/metadata/definition/definition.go
+++ b/metadata/definition/definition.go
@@ -17,6 +17,16 @@
package definition
+import (
+ "bytes"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/constant"
+)
+
+// ServiceDefinition is the describer of service definition
type ServiceDefinition struct {
CanonicalName string
CodeSource string
@@ -24,6 +34,7 @@ type ServiceDefinition struct {
Types []TypeDefinition
}
+// MethodDefinition is the describer of method definition
type MethodDefinition struct {
Name string
ParameterTypes []string
@@ -31,6 +42,7 @@ type MethodDefinition struct {
Parameters []TypeDefinition
}
+// TypeDefinition is the describer of type definition
type TypeDefinition struct {
Id string
Type string
@@ -39,3 +51,39 @@ type TypeDefinition struct {
Properties map[string]TypeDefinition
TypeBuilderName string
}
+
+// BuildServiceDefinition can build service definition which will be used to describe a service
+func BuildServiceDefinition(service common.Service, url common.URL) ServiceDefinition {
+ sd := ServiceDefinition{}
+ sd.CanonicalName = url.Service()
+
+ for k, m := range service.Method() {
+ var paramTypes []string
+ for _, t := range m.ArgsType() {
+ paramTypes = append(paramTypes, t.Kind().String())
+ }
+ methodD := MethodDefinition{
+ Name: k,
+ ParameterTypes: paramTypes,
+ ReturnType: m.ReplyType().Kind().String(),
+ }
+ sd.Methods = append(sd.Methods, methodD)
+ }
+
+ return sd
+}
+
+// ServiceDescriperBuild: build the service key, format is `group/serviceName:version` which be same as URL's service key
+func ServiceDescriperBuild(serviceName string, group string, version string) string {
+ buf := &bytes.Buffer{}
+ if group != "" {
+ buf.WriteString(group)
+ buf.WriteString(constant.PATH_SEPARATOR)
+ }
+ buf.WriteString(serviceName)
+ if version != "" && version != "0.0.0" {
+ buf.WriteString(constant.KEY_SEPARATOR)
+ buf.WriteString(version)
+ }
+ return buf.String()
+}
diff --git a/metadata/identifier/base_metadata_identifier.go b/metadata/identifier/base_metadata_identifier.go
index a314671055be523844fd7d8f9589b8b6031632bc..64290c668f14277a5f2c8b9e7603ca50e7713fd6 100644
--- a/metadata/identifier/base_metadata_identifier.go
+++ b/metadata/identifier/base_metadata_identifier.go
@@ -25,19 +25,21 @@ import (
"github.com/apache/dubbo-go/common/constant"
)
-type BaseMetadataIdentifier interface {
- getFilePathKey(params ...string) string
- getIdentifierKey(params ...string) string
+// BaseMetadataIdentifier defined for describe the Metadata base identify
+type IMetadataIdentifier interface {
+ GetFilePathKey() string
+ GetIdentifierKey() string
}
-type BaseServiceMetadataIdentifier struct {
- serviceInterface string
- version string
- group string
- side string
+// BaseMetadataIdentifier is the base implement of BaseMetadataIdentifier interface
+type BaseMetadataIdentifier struct {
+ ServiceInterface string
+ Version string
+ Group string
+ Side string
}
-// joinParams...
+// joinParams will join the specified char in slice, and build as string
func joinParams(joinChar string, params []string) string {
var joinedStr string
for _, param := range params {
@@ -47,24 +49,24 @@ func joinParams(joinChar string, params []string) string {
return joinedStr
}
-// getIdentifierKey...
-func (mdi *BaseServiceMetadataIdentifier) getIdentifierKey(params ...string) string {
- return mdi.serviceInterface +
- constant.KEY_SEPARATOR + mdi.version +
- constant.KEY_SEPARATOR + mdi.group +
- constant.KEY_SEPARATOR + mdi.side +
+// getIdentifierKey will return string format as service:Version:Group:Side:param1:param2...
+func (mdi *BaseMetadataIdentifier) getIdentifierKey(params ...string) string {
+ return mdi.ServiceInterface +
+ constant.KEY_SEPARATOR + mdi.Version +
+ constant.KEY_SEPARATOR + mdi.Group +
+ constant.KEY_SEPARATOR + mdi.Side +
joinParams(constant.KEY_SEPARATOR, params)
}
-// getFilePathKey...
-func (mdi *BaseServiceMetadataIdentifier) getFilePathKey(params ...string) string {
- path := serviceToPath(mdi.serviceInterface)
+// getFilePathKey will return string format as metadata/path/Version/Group/Side/param1/param2...
+func (mdi *BaseMetadataIdentifier) getFilePathKey(params ...string) string {
+ path := serviceToPath(mdi.ServiceInterface)
return constant.DEFAULT_PATH_TAG +
withPathSeparator(path) +
- withPathSeparator(mdi.version) +
- withPathSeparator(mdi.group) +
- withPathSeparator(mdi.side) +
+ withPathSeparator(mdi.Version) +
+ withPathSeparator(mdi.Group) +
+ withPathSeparator(mdi.Side) +
joinParams(constant.PATH_SEPARATOR, params)
}
diff --git a/metadata/service.go b/metadata/identifier/base_metadata_identifier_test.go
similarity index 55%
rename from metadata/service.go
rename to metadata/identifier/base_metadata_identifier_test.go
index d85703c95a57183d5c0a5b2445839e946dc6a59b..5b60992ab6132ecb306245af31bba7e3d0f09117 100644
--- a/metadata/service.go
+++ b/metadata/identifier/base_metadata_identifier_test.go
@@ -15,23 +15,27 @@
* limitations under the License.
*/
-package metadata
+package identifier
import (
- "github.com/apache/dubbo-go/common"
- gxset "github.com/dubbogo/gost/container/set"
+ "testing"
)
-type MetadataService interface {
- ServiceName() string
- ExportURL(url *common.URL) bool
- UnexportURL(url *common.URL) bool
- RefreshMetadata(exportedRevision string, subscribedRevision string) bool
- SubscribeURL(url *common.URL) bool
- UnsubscribeURL(url *common.URL) bool
- PublishServiceDefinition(url *common.URL)
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+var baseId = &BaseMetadataIdentifier{
+ ServiceInterface: "org.apache.pkg.mockService",
+ Version: "1.0.0",
+ Group: "Group",
+ Side: "provider",
+}
+
+func TestBaseGetFilePathKey(t *testing.T) {
+ assert.Equal(t, "metadata/1.0.0/Group/provider/a/b/c", baseId.getFilePathKey("a", "b", "c"))
+}
- GetExportedURLs(serviceInterface string, group string, version string, protocol string) gxset.HashSet
- GetServiceDefinition(interfaceName string, version string, group string) string
- GetServiceDefinitionByServiceKey(serviceKey string) string
+func TestBaseGetIdentifierKey(t *testing.T) {
+ assert.Equal(t, "org.apache.pkg.mockService:1.0.0:Group:provider:a:b:c", baseId.getIdentifierKey("a", "b", "c"))
}
diff --git a/metadata/identifier/metadata_identifier.go b/metadata/identifier/metadata_identifier.go
index f3df8f36546093a826279c4e9ec1546f78d444bd..18b330ae083d55cf77330d19c144b2d4a6bde862 100644
--- a/metadata/identifier/metadata_identifier.go
+++ b/metadata/identifier/metadata_identifier.go
@@ -17,17 +17,18 @@
package identifier
+// MetadataIdentifier is inherit baseMetaIdentifier with Application name
type MetadataIdentifier struct {
- application string
+ Application string
BaseMetadataIdentifier
}
-// getIdentifierKey...
-func (mdi *MetadataIdentifier) getIdentifierKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.application)
+// GetIdentifierKey will return string format as service:Version:Group:Side:Application
+func (mdi *MetadataIdentifier) GetIdentifierKey() string {
+ return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.Application)
}
-// getIdentifierKey...
-func (mdi *MetadataIdentifier) getFilePathKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.application)
+// GetFilePathKey will return string format as metadata/path/Version/Group/Side/Application
+func (mdi *MetadataIdentifier) GetFilePathKey() string {
+ return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.Application)
}
diff --git a/metadata/identifier/metadata_identifier_test.go b/metadata/identifier/metadata_identifier_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..cba3c0dd76a01f2125b87db4478f99501bf2c284
--- /dev/null
+++ b/metadata/identifier/metadata_identifier_test.go
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package identifier
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+var metadataId = &MetadataIdentifier{
+ Application: "app",
+ BaseMetadataIdentifier: BaseMetadataIdentifier{
+ ServiceInterface: "org.apache.pkg.mockService",
+ Version: "1.0.0",
+ Group: "Group",
+ Side: "provider",
+ },
+}
+
+func TestGetFilePathKey(t *testing.T) {
+ assert.Equal(t, "metadata/1.0.0/Group/provider/app", metadataId.GetFilePathKey())
+}
+
+func TestGetIdentifierKey(t *testing.T) {
+ assert.Equal(t, "org.apache.pkg.mockService:1.0.0:Group:provider:app", metadataId.GetIdentifierKey())
+}
diff --git a/metadata/identifier/service_metadata_identifier.go b/metadata/identifier/service_metadata_identifier.go
index 373df0130dd1f87e3175918bde50060c4be89616..92c15704db3bb873b3aff26297643119f8835f45 100644
--- a/metadata/identifier/service_metadata_identifier.go
+++ b/metadata/identifier/service_metadata_identifier.go
@@ -21,18 +21,19 @@ import (
"github.com/apache/dubbo-go/common/constant"
)
+// ServiceMetadataIdentifier is inherit baseMetaIdentifier with service params: Revision and Protocol
type ServiceMetadataIdentifier struct {
- revision string
- protocol string
+ Revision string
+ Protocol string
BaseMetadataIdentifier
}
-// getIdentifierKey...
-func (mdi *ServiceMetadataIdentifier) getIdentifierKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.protocol + constant.KEY_REVISON_PREFIX + mdi.revision)
+// GetIdentifierKey will return string format as service:Version:Group:Side:Protocol:"revision"+Revision
+func (mdi *ServiceMetadataIdentifier) GetIdentifierKey() string {
+ return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.Protocol, constant.KEY_REVISON_PREFIX+mdi.Revision)
}
-// getIdentifierKey...
-func (mdi *ServiceMetadataIdentifier) getFilePathKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.protocol + constant.KEY_REVISON_PREFIX + mdi.revision)
+// GetFilePathKey will return string format as metadata/path/Version/Group/Side/Protocol/"revision"+Revision
+func (mdi *ServiceMetadataIdentifier) GetFilePathKey() string {
+ return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.Protocol, constant.KEY_REVISON_PREFIX+mdi.Revision)
}
diff --git a/metadata/identifier/service_metadata_identifier_test.go b/metadata/identifier/service_metadata_identifier_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..d7ef44a4bbc7611b6391122f8f5841db349eb036
--- /dev/null
+++ b/metadata/identifier/service_metadata_identifier_test.go
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package identifier
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+var serviceMetadataId = &ServiceMetadataIdentifier{
+ Revision: "1.0",
+ Protocol: "dubbo",
+ BaseMetadataIdentifier: BaseMetadataIdentifier{
+ ServiceInterface: "org.apache.pkg.mockService",
+ Version: "1.0.0",
+ Group: "Group",
+ Side: "provider",
+ },
+}
+
+func TestServiceGetFilePathKey(t *testing.T) {
+ assert.Equal(t, "metadata/1.0.0/Group/provider/dubbo/revision1.0", serviceMetadataId.GetFilePathKey())
+}
+
+func TestServiceGetIdentifierKey(t *testing.T) {
+ assert.Equal(t, "org.apache.pkg.mockService:1.0.0:Group:provider:dubbo:revision1.0", serviceMetadataId.GetIdentifierKey())
+}
diff --git a/metadata/identifier/subscribe_metadata_identifier.go b/metadata/identifier/subscribe_metadata_identifier.go
index fd3a290b41e870674366943e12a396c3dae7e238..e599fc9e0da1962d60d0bde2646eed552e26e95d 100644
--- a/metadata/identifier/subscribe_metadata_identifier.go
+++ b/metadata/identifier/subscribe_metadata_identifier.go
@@ -1,16 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package identifier
+// SubscriberMetadataIdentifier is inherit baseMetaIdentifier with service params: Revision
type SubscriberMetadataIdentifier struct {
- revision string
+ Revision string
BaseMetadataIdentifier
}
-// getIdentifierKey...
-func (mdi *SubscriberMetadataIdentifier) getIdentifierKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.revision)
+// GetIdentifierKey will return string format as service:Version:Group:Side:Revision
+func (mdi *SubscriberMetadataIdentifier) GetIdentifierKey() string {
+ return mdi.BaseMetadataIdentifier.getIdentifierKey(mdi.Revision)
}
-// getIdentifierKey...
-func (mdi *SubscriberMetadataIdentifier) getFilePathKey(params ...string) string {
- return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.revision)
+// GetFilePathKey will return string format as metadata/path/Version/Group/Side/Revision
+func (mdi *SubscriberMetadataIdentifier) GetFilePathKey() string {
+ return mdi.BaseMetadataIdentifier.getFilePathKey(mdi.Revision)
}
diff --git a/metadata/identifier/subscribe_metadata_identifier_test.go b/metadata/identifier/subscribe_metadata_identifier_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..9c9ef70641c52222a09475e97a2afbb604a467ff
--- /dev/null
+++ b/metadata/identifier/subscribe_metadata_identifier_test.go
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package identifier
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+var subscribeMetadataId = &SubscriberMetadataIdentifier{
+ Revision: "1.0",
+ BaseMetadataIdentifier: BaseMetadataIdentifier{
+ ServiceInterface: "org.apache.pkg.mockService",
+ Version: "1.0.0",
+ Group: "Group",
+ Side: "provider",
+ },
+}
+
+func TestSubscribeGetFilePathKey(t *testing.T) {
+ assert.Equal(t, "metadata/1.0.0/Group/provider/1.0", subscribeMetadataId.GetFilePathKey())
+}
+
+func TestSubscribeGetIdentifierKey(t *testing.T) {
+ assert.Equal(t, "org.apache.pkg.mockService:1.0.0:Group:provider:1.0", subscribeMetadataId.GetIdentifierKey())
+}
diff --git a/metadata/namemapping/dynamic/service_name_mapping.go b/metadata/mapping/dynamic/service_name_mapping.go
similarity index 97%
rename from metadata/namemapping/dynamic/service_name_mapping.go
rename to metadata/mapping/dynamic/service_name_mapping.go
index e93c256fe093b4a3e3c431e1d012038b2bb7976b..4cfac8f82887d0b101beaf55ae9ca84c124d30b3 100644
--- a/metadata/namemapping/dynamic/service_name_mapping.go
+++ b/metadata/mapping/dynamic/service_name_mapping.go
@@ -31,7 +31,7 @@ import (
"github.com/apache/dubbo-go/common/constant"
"github.com/apache/dubbo-go/config"
"github.com/apache/dubbo-go/config_center"
- "github.com/apache/dubbo-go/metadata"
+ "github.com/apache/dubbo-go/metadata/mapping"
)
const (
@@ -77,6 +77,6 @@ func (d *DynamicConfigurationServiceNameMapping) buildGroup(serviceInterface str
}
// NewServiceNameMapping will create an instance of DynamicConfigurationServiceNameMapping
-func NewServiceNameMapping(dc config_center.DynamicConfiguration) metadata.ServiceNameMapping {
+func NewServiceNameMapping(dc config_center.DynamicConfiguration) mapping.ServiceNameMapping {
return &DynamicConfigurationServiceNameMapping{dc: dc}
}
diff --git a/metadata/namemapping/dynamic/service_name_mapping_test.go b/metadata/mapping/dynamic/service_name_mapping_test.go
similarity index 100%
rename from metadata/namemapping/dynamic/service_name_mapping_test.go
rename to metadata/mapping/dynamic/service_name_mapping_test.go
diff --git a/metadata/namemapping/memory/service_name_mapping.go b/metadata/mapping/memory/service_name_mapping.go
similarity index 100%
rename from metadata/namemapping/memory/service_name_mapping.go
rename to metadata/mapping/memory/service_name_mapping.go
diff --git a/metadata/service_name_mapping.go b/metadata/mapping/service_name_mapping.go
similarity index 98%
rename from metadata/service_name_mapping.go
rename to metadata/mapping/service_name_mapping.go
index c14e8ce2e7c40d1573897dfd6ba64c16e18acac7..6caed9f0b48c1bb9c2f0f1026eb642f69bb31113 100644
--- a/metadata/service_name_mapping.go
+++ b/metadata/mapping/service_name_mapping.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package metadata
+package mapping
import (
gxset "github.com/dubbogo/gost/container/set"
diff --git a/metadata/report_factory.go b/metadata/report/factory/report_factory.go
similarity index 76%
rename from metadata/report_factory.go
rename to metadata/report/factory/report_factory.go
index 19b1004eee57073acec13c7f114179c47c73f145..8769ebdd2fd1f088415232bd4463d02f7ebd730f 100644
--- a/metadata/report_factory.go
+++ b/metadata/report/factory/report_factory.go
@@ -15,16 +15,21 @@
* limitations under the License.
*/
-package metadata
+package factory
import (
"github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/metadata/report"
)
var (
- MetadataReportInstance MetadataReport
+ MetadataReportInstance report.MetadataReport
)
+// MetadataReportFactory interface will create metadata report
type MetadataReportFactory interface {
- CreateMetadataReport(*common.URL) MetadataReport
+ CreateMetadataReport(*common.URL) report.MetadataReport
+}
+
+type BaseMetadataReportFactory struct {
}
diff --git a/metadata/report.go b/metadata/report/report.go
similarity index 90%
rename from metadata/report.go
rename to metadata/report/report.go
index 3fcc71241411d4a8f9577bb5fb3233e67942cd52..81227e0c765b61df7edc8a5d025b9cd1921d1113 100644
--- a/metadata/report.go
+++ b/metadata/report/report.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package metadata
+package report
import (
"github.com/apache/dubbo-go/common"
@@ -23,9 +23,10 @@ import (
"github.com/apache/dubbo-go/metadata/identifier"
)
+// MetadataReport is an interface of remote metadata report
type MetadataReport interface {
StoreProviderMetadata(*identifier.MetadataIdentifier, *definition.ServiceDefinition)
- StoreConsumeretadata(*identifier.MetadataIdentifier, map[string]string)
+ StoreConsumerMetadata(*identifier.MetadataIdentifier, map[string]string)
SaveServiceMetadata(*identifier.ServiceMetadataIdentifier, *common.URL)
RemoveServiceMetadata(*identifier.ServiceMetadataIdentifier)
GetExportedURLs(*identifier.ServiceMetadataIdentifier) []string
diff --git a/metadata/service/exporter/configurable/exporter.go b/metadata/service/exporter/configurable/exporter.go
new file mode 100644
index 0000000000000000000000000000000000000000..3d12e0ecd4def4b9d99f346a4f556fc3d781d1b2
--- /dev/null
+++ b/metadata/service/exporter/configurable/exporter.go
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package configurable
+
+import (
+ "context"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/constant"
+ "github.com/apache/dubbo-go/common/logger"
+ "github.com/apache/dubbo-go/config"
+ "github.com/apache/dubbo-go/metadata/service"
+ "github.com/apache/dubbo-go/metadata/service/exporter"
+)
+
+// MetadataServiceExporter is the ConfigurableMetadataServiceExporter which implement MetadataServiceExporter interface
+type MetadataServiceExporter struct {
+ serviceConfig *config.ServiceConfig
+ lock sync.RWMutex
+ metadataService service.MetadataService
+}
+
+// NewMetadataServiceExporter will return a service_exporter.MetadataServiceExporter with the specified metadata service
+func NewMetadataServiceExporter(metadataService service.MetadataService) exporter.MetadataServiceExporter {
+ return &MetadataServiceExporter{
+ metadataService: metadataService,
+ }
+}
+
+// Export will export the metadataService
+func (exporter *MetadataServiceExporter) Export() error {
+ if !exporter.IsExported() {
+
+ serviceConfig := config.NewServiceConfig("MetadataService", context.Background())
+ serviceConfig.Protocol = constant.DEFAULT_PROTOCOL
+ serviceConfig.Protocols = map[string]*config.ProtocolConfig{
+ constant.DEFAULT_PROTOCOL: generateMetadataProtocol(),
+ }
+ serviceConfig.InterfaceName = constant.METADATA_SERVICE_NAME
+ serviceConfig.Group = config.GetApplicationConfig().Name
+ serviceConfig.Version = exporter.metadataService.Version()
+
+ var err error
+ func() {
+ exporter.lock.Lock()
+ defer exporter.lock.Unlock()
+ exporter.serviceConfig = serviceConfig
+ exporter.serviceConfig.Implement(exporter.metadataService)
+ err = exporter.serviceConfig.Export()
+ }()
+
+ logger.Infof("The MetadataService exports urls : %v ", exporter.serviceConfig.GetExportedUrls())
+ return err
+ }
+ logger.Warnf("The MetadataService has been exported : %v ", exporter.serviceConfig.GetExportedUrls())
+ return nil
+}
+
+// Unexport will unexport the metadataService
+func (exporter *MetadataServiceExporter) Unexport() {
+ if exporter.IsExported() {
+ exporter.serviceConfig.Unexport()
+ }
+}
+
+// GetExportedURLs will return the urls that export use.
+// Noticeļ¼The exported url is not same as url in registry , for example it lack the ip.
+func (exporter *MetadataServiceExporter) GetExportedURLs() []*common.URL {
+ return exporter.serviceConfig.GetExportedUrls()
+}
+
+// isExported will return is metadataServiceExporter exported or not
+func (exporter *MetadataServiceExporter) IsExported() bool {
+ exporter.lock.RLock()
+ defer exporter.lock.RUnlock()
+ return exporter.serviceConfig != nil && exporter.serviceConfig.IsExport()
+}
+
+// generateMetadataProtocol will return a default ProtocolConfig
+func generateMetadataProtocol() *config.ProtocolConfig {
+ return &config.ProtocolConfig{
+ Name: constant.DEFAULT_PROTOCOL,
+ Port: "20000",
+ }
+}
diff --git a/metadata/service/exporter/configurable/exporter_test.go b/metadata/service/exporter/configurable/exporter_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..220ef71daca47f46bdcd4b88b215970399a5da31
--- /dev/null
+++ b/metadata/service/exporter/configurable/exporter_test.go
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package configurable
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ _ "github.com/apache/dubbo-go/common/proxy/proxy_factory"
+ "github.com/apache/dubbo-go/config"
+ _ "github.com/apache/dubbo-go/filter/filter_impl"
+ "github.com/apache/dubbo-go/metadata/service/inmemory"
+ "github.com/apache/dubbo-go/protocol/dubbo"
+ _ "github.com/apache/dubbo-go/protocol/dubbo"
+)
+
+func TestConfigurableExporter(t *testing.T) {
+ dubbo.SetServerConfig(dubbo.ServerConfig{
+ SessionNumber: 700,
+ SessionTimeout: "20s",
+ GettySessionParam: dubbo.GettySessionParam{
+ CompressEncoding: false,
+ TcpNoDelay: true,
+ TcpKeepAlive: true,
+ KeepAlivePeriod: "120s",
+ TcpRBufSize: 262144,
+ TcpWBufSize: 65536,
+ PkgWQSize: 512,
+ TcpReadTimeout: "1s",
+ TcpWriteTimeout: "5s",
+ WaitTimeout: "1s",
+ MaxMsgLen: 10240000000,
+ SessionName: "server",
+ }})
+ mockInitProviderWithSingleRegistry()
+ metadataService := inmemory.NewMetadataService()
+ exported := NewMetadataServiceExporter(metadataService)
+ assert.Equal(t, false, exported.IsExported())
+ assert.NoError(t, exported.Export())
+ assert.Equal(t, true, exported.IsExported())
+ assert.Regexp(t, "dubbo://:20000/MetadataService*", exported.GetExportedURLs()[0].String())
+ exported.Unexport()
+ assert.Equal(t, false, exported.IsExported())
+}
+
+// mockInitProviderWithSingleRegistry will init a mocked providerConfig
+func mockInitProviderWithSingleRegistry() {
+ providerConfig := &config.ProviderConfig{
+ ApplicationConfig: &config.ApplicationConfig{
+ Organization: "dubbo_org",
+ Name: "dubbo",
+ Module: "module",
+ Version: "1.0.0",
+ Owner: "dubbo",
+ Environment: "test"},
+ Registry: &config.RegistryConfig{
+ Address: "mock://127.0.0.1:2181",
+ Username: "user1",
+ Password: "pwd1",
+ },
+ Registries: map[string]*config.RegistryConfig{},
+ Services: map[string]*config.ServiceConfig{
+ "MockService": {
+ InterfaceName: "com.MockService",
+ Protocol: "mock",
+ Cluster: "failover",
+ Loadbalance: "random",
+ Retries: "3",
+ Group: "huadong_idc",
+ Version: "1.0.0",
+ Methods: []*config.MethodConfig{
+ {
+ Name: "GetUser",
+ Retries: "2",
+ Loadbalance: "random",
+ Weight: 200,
+ },
+ {
+ Name: "GetUser1",
+ Retries: "2",
+ Loadbalance: "random",
+ Weight: 200,
+ },
+ },
+ },
+ },
+ Protocols: map[string]*config.ProtocolConfig{
+ "mock": {
+ Name: "mock",
+ Ip: "127.0.0.1",
+ Port: "20000",
+ },
+ },
+ }
+ providerConfig.Services["MockService"].InitExported()
+ config.SetProviderConfig(*providerConfig)
+}
diff --git a/metadata/exporter.go b/metadata/service/exporter/exporter.go
similarity index 81%
rename from metadata/exporter.go
rename to metadata/service/exporter/exporter.go
index 5d47f8bd808ec802ba73c7db73d22c78c675d12a..cfdef3a0e79d29ce31717c0fc3c575e9e4ba1759 100644
--- a/metadata/exporter.go
+++ b/metadata/service/exporter/exporter.go
@@ -15,15 +15,16 @@
* limitations under the License.
*/
-package metadata
+package exporter
import (
"github.com/apache/dubbo-go/common"
)
-type MetadataExporter interface {
- Export() MetadataExporter
- Unexport() MetadataExporter
+// MetadataServiceExporter will export & unexport the metadata service, get exported url, and return is exported or not
+type MetadataServiceExporter interface {
+ Export() error
+ Unexport()
GetExportedURLs() []*common.URL
IsExported() bool
}
diff --git a/metadata/service/inmemory/service.go b/metadata/service/inmemory/service.go
new file mode 100644
index 0000000000000000000000000000000000000000..c59949401f419b44ce155a914a7afff7c327a8fe
--- /dev/null
+++ b/metadata/service/inmemory/service.go
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package inmemory
+
+import (
+ "encoding/json"
+ "sync"
+)
+
+import (
+ cm "github.com/Workiva/go-datastructures/common"
+ "github.com/Workiva/go-datastructures/slice/skip"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/constant"
+ "github.com/apache/dubbo-go/common/logger"
+ "github.com/apache/dubbo-go/metadata/definition"
+ "github.com/apache/dubbo-go/metadata/service"
+)
+
+// MetadataService is store and query the metadata info in memory when each service registry
+type MetadataService struct {
+ service.BaseMetadataService
+ exportedServiceURLs *sync.Map
+ subscribedServiceURLs *sync.Map
+ serviceDefinitions *sync.Map
+ lock *sync.RWMutex
+}
+
+// NewMetadataService: initiate a metadata service
+func NewMetadataService() *MetadataService {
+ return &MetadataService{
+ exportedServiceURLs: &sync.Map{},
+ subscribedServiceURLs: &sync.Map{},
+ serviceDefinitions: &sync.Map{},
+ lock: &sync.RWMutex{},
+ }
+}
+
+// comparator is defined as Comparator for skip list to compare the URL
+type comparator common.URL
+
+// Compare is defined as Comparator for skip list to compare the URL
+func (c comparator) Compare(comp cm.Comparator) int {
+ a := common.URL(c).String()
+ b := common.URL(comp.(comparator)).String()
+ switch {
+ case a > b:
+ return 1
+ case a < b:
+ return -1
+ default:
+ return 0
+ }
+}
+
+// addURL will add URL in memory
+func (mts *MetadataService) addURL(targetMap *sync.Map, url *common.URL) bool {
+ var (
+ urlSet interface{}
+ loaded bool
+ )
+ logger.Debug(url.ServiceKey())
+ if urlSet, loaded = targetMap.LoadOrStore(url.ServiceKey(), skip.New(uint64(0))); loaded {
+ mts.lock.RLock()
+ wantedUrl := urlSet.(*skip.SkipList).Get(comparator(*url))
+ if len(wantedUrl) > 0 && wantedUrl[0] != nil {
+ mts.lock.RUnlock()
+ return false
+ }
+ mts.lock.RUnlock()
+ }
+ mts.lock.Lock()
+ //double chk
+ wantedUrl := urlSet.(*skip.SkipList).Get(comparator(*url))
+ if len(wantedUrl) > 0 && wantedUrl[0] != nil {
+ mts.lock.Unlock()
+ return false
+ }
+ urlSet.(*skip.SkipList).Insert(comparator(*url))
+ mts.lock.Unlock()
+ return true
+}
+
+// removeURL is used to remove specified url
+func (mts *MetadataService) removeURL(targetMap *sync.Map, url *common.URL) {
+ if value, loaded := targetMap.Load(url.ServiceKey()); loaded {
+ mts.lock.Lock()
+ value.(*skip.SkipList).Delete(comparator(*url))
+ mts.lock.Unlock()
+ mts.lock.RLock()
+ defer mts.lock.RUnlock()
+ if value.(*skip.SkipList).Len() == 0 {
+ targetMap.Delete(url.ServiceKey())
+ }
+ }
+}
+
+// getAllService can return all the exportedUrlString except for metadataService
+func (mts *MetadataService) getAllService(services *sync.Map) *skip.SkipList {
+ skipList := skip.New(uint64(0))
+ services.Range(func(key, value interface{}) bool {
+ urls := value.(*skip.SkipList)
+ for i := uint64(0); i < urls.Len(); i++ {
+ url := common.URL(urls.ByPosition(i).(comparator))
+ if url.GetParam(constant.INTERFACE_KEY, url.Path) != "MetadataService" {
+ skipList.Insert(comparator(url))
+ }
+ }
+ return true
+ })
+ return skipList
+}
+
+// getSpecifiedService can return specified service url by serviceKey
+func (mts *MetadataService) getSpecifiedService(services *sync.Map, serviceKey string, protocol string) *skip.SkipList {
+ skipList := skip.New(uint64(0))
+ serviceList, loaded := services.Load(serviceKey)
+ if loaded {
+ urls := serviceList.(*skip.SkipList)
+ for i := uint64(0); i < urls.Len(); i++ {
+ url := common.URL(urls.ByPosition(i).(comparator))
+ if len(protocol) == 0 || url.Protocol == protocol || url.GetParam(constant.PROTOCOL_KEY, "") == protocol {
+ skipList.Insert(comparator(url))
+ }
+ }
+ }
+ return skipList
+}
+
+// ExportURL can store the in memory
+func (mts *MetadataService) ExportURL(url common.URL) (bool, error) {
+ return mts.addURL(mts.exportedServiceURLs, &url), nil
+}
+
+// UnexportURL can remove the url store in memory
+func (mts *MetadataService) UnexportURL(url common.URL) error {
+ mts.removeURL(mts.exportedServiceURLs, &url)
+ return nil
+}
+
+// SubscribeURL can store the in memory
+func (mts *MetadataService) SubscribeURL(url common.URL) (bool, error) {
+ return mts.addURL(mts.subscribedServiceURLs, &url), nil
+}
+
+// UnsubscribeURL can remove the url store in memory
+func (mts *MetadataService) UnsubscribeURL(url common.URL) error {
+ mts.removeURL(mts.subscribedServiceURLs, &url)
+ return nil
+}
+
+// PublishServiceDefinition: publish url's service metadata info, and write into memory
+func (mts *MetadataService) PublishServiceDefinition(url common.URL) error {
+ interfaceName := url.GetParam(constant.INTERFACE_KEY, "")
+ isGeneric := url.GetParamBool(constant.GENERIC_KEY, false)
+ if len(interfaceName) > 0 && !isGeneric {
+ //judge is consumer or provider
+ //side := url.GetParam(constant.SIDE_KEY, "")
+ //var service common.RPCService
+ service := common.ServiceMap.GetService(url.Protocol, url.GetParam(constant.BEAN_NAME_KEY, url.Service()))
+ //if side == common.RoleType(common.CONSUMER).Role() {
+ // //TODO:generate the service definition and store it
+ //
+ //} else if side == common.RoleType(common.PROVIDER).Role() {
+ // //TODO:generate the service definition and store it
+ //}
+ sd := definition.BuildServiceDefinition(*service, url)
+ data, err := json.Marshal(sd)
+ if err != nil {
+ logger.Errorf("publishProvider getServiceDescriptor error. providerUrl:%v , error: ", url, err)
+ }
+ mts.serviceDefinitions.Store(url.ServiceKey(), string(data))
+ return nil
+ }
+ logger.Errorf("publishProvider interfaceName is empty . providerUrl:%v ", url)
+ return nil
+}
+
+// GetExportedURLs get all exported urls
+func (mts *MetadataService) GetExportedURLs(serviceInterface string, group string, version string, protocol string) (*skip.SkipList, error) {
+ if serviceInterface == constant.ANY_VALUE {
+ return mts.getAllService(mts.exportedServiceURLs), nil
+ } else {
+ serviceKey := definition.ServiceDescriperBuild(serviceInterface, group, version)
+ return mts.getSpecifiedService(mts.exportedServiceURLs, serviceKey, protocol), nil
+ }
+}
+
+// GetSubscribedURLs get all subscribedUrl
+func (mts *MetadataService) GetSubscribedURLs() (*skip.SkipList, error) {
+ return mts.getAllService(mts.subscribedServiceURLs), nil
+}
+
+// GetServiceDefinition can get service definition by interfaceName, group and version
+func (mts *MetadataService) GetServiceDefinition(interfaceName string, group string, version string) (string, error) {
+ serviceKey := definition.ServiceDescriperBuild(interfaceName, group, version)
+ v, _ := mts.serviceDefinitions.Load(serviceKey)
+ return v.(string), nil
+}
+
+// GetServiceDefinition can get service definition by serviceKey
+func (mts *MetadataService) GetServiceDefinitionByServiceKey(serviceKey string) (string, error) {
+ v, _ := mts.serviceDefinitions.Load(serviceKey)
+ return v.(string), nil
+}
+
+// Version will return the version of metadata service
+func (mts *MetadataService) Version() string {
+ return "1.0.0"
+}
+
+// Version will return the version of metadata service
+func (mts *MetadataService) Reference() string {
+ return "MetadataService"
+}
diff --git a/metadata/service/inmemory/service_test.go b/metadata/service/inmemory/service_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..9e593db282e7f4fa55d52c49129a15a9b389c67f
--- /dev/null
+++ b/metadata/service/inmemory/service_test.go
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package inmemory
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/metadata/definition"
+)
+
+type User struct {
+ Id string
+ Name string
+ Age int32
+ Time time.Time
+}
+
+type UserProvider struct {
+}
+
+func (u *UserProvider) GetUser(ctx context.Context, req []interface{}) (*User, error) {
+ rsp := User{"A001", "Alex Stocks", 18, time.Now()}
+ return &rsp, nil
+}
+
+func (u *UserProvider) Reference() string {
+ return "UserProvider"
+}
+
+func (u User) JavaClassName() string {
+ return "com.ikurento.user.User"
+}
+
+func TestMetadataService(t *testing.T) {
+ mts := NewMetadataService()
+ serviceName := "com.ikurento.user.UserProvider"
+ group := "group1"
+ version := "0.0.1"
+ protocol := "dubbo"
+ beanName := "UserProvider"
+
+ u2, err := common.NewURL(fmt.Sprintf(
+ "%v://127.0.0.1:20000/com.ikurento.user.UserProvider2?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=%v&ip=192.168.56.1&methods=GetUser&module=dubbogo+user-info+server&org=ikurento.com&"+
+ "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v",
+ protocol, serviceName, group, version, beanName))
+ assert.NoError(t, err)
+ mts.ExportURL(u2)
+
+ u3, err := common.NewURL(fmt.Sprintf(
+ "%v://127.0.0.1:20000/com.ikurento.user.UserProvider3?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=%v&ip=192.168.56.1&methods=GetUser&module=dubbogo+user-info+server&org=ikurento.com&"+
+ "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v",
+ protocol, serviceName, group, version, beanName))
+ assert.NoError(t, err)
+ mts.ExportURL(u3)
+
+ u, err := common.NewURL(fmt.Sprintf(
+ "%v://127.0.0.1:20000/com.ikurento.user.UserProvider1?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=%v&ip=192.168.56.1&methods=GetUser&module=dubbogo+user-info+server&org=ikurento.com&"+
+ "owner=ZX&pid=1447&revision=0.0.1&side=provider&timeout=3000×tamp=1556509797245&group=%v&version=%v&bean.name=%v",
+ protocol, serviceName, group, version, beanName))
+ assert.NoError(t, err)
+ mts.ExportURL(u)
+ list, _ := mts.GetExportedURLs(serviceName, group, version, protocol)
+ assert.Equal(t, uint64(3), list.Len())
+ iter := list.IterAtPosition(0)
+ for iter.Next() {
+ comparator := iter.Value()
+ fmt.Println(comparator)
+ }
+ mts.SubscribeURL(u)
+
+ mts.SubscribeURL(u)
+ list2, _ := mts.GetSubscribedURLs()
+ assert.Equal(t, uint64(1), list2.Len())
+
+ mts.UnexportURL(u)
+ list3, _ := mts.GetExportedURLs(serviceName, group, version, protocol)
+ assert.Equal(t, uint64(2), list3.Len())
+
+ mts.UnsubscribeURL(u)
+ list4, _ := mts.GetSubscribedURLs()
+ assert.Equal(t, uint64(0), list4.Len())
+
+ userProvider := &UserProvider{}
+ common.ServiceMap.Register(serviceName, protocol, userProvider)
+ mts.PublishServiceDefinition(u)
+ expected := "{\"CanonicalName\":\"com.ikurento.user.UserProvider\",\"CodeSource\":\"\"," +
+ "\"Methods\":[{\"Name\":\"GetUser\",\"ParameterTypes\":[\"slice\"],\"ReturnType\":\"ptr\"," +
+ "\"Parameters\":null}],\"Types\":null}"
+ def1, _ := mts.GetServiceDefinition(serviceName, group, version)
+ assert.Equal(t, expected, def1)
+ serviceKey := definition.ServiceDescriperBuild(serviceName, group, version)
+ def2, _ := mts.GetServiceDefinitionByServiceKey(serviceKey)
+ assert.Equal(t, expected, def2)
+}
diff --git a/metadata/service/service.go b/metadata/service/service.go
new file mode 100644
index 0000000000000000000000000000000000000000..bc526c5411383f0d5cee971cef4f84d6f4f48f59
--- /dev/null
+++ b/metadata/service/service.go
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package service
+
+import (
+ "github.com/Workiva/go-datastructures/slice/skip"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/config"
+)
+
+// Metadataservice is used to define meta data related behaviors
+type MetadataService interface {
+ common.RPCService
+ // ServiceName will get the service's name in meta service , which is application name
+ ServiceName() (string, error)
+ // ExportURL will store the exported url in metadata
+ ExportURL(url common.URL) (bool, error)
+ // UnexportURL will delete the exported url in metadata
+ UnexportURL(url common.URL) error
+ // SubscribeURL will store the subscribed url in metadata
+ SubscribeURL(url common.URL) (bool, error)
+ // UnsubscribeURL will delete the subscribed url in metadata
+ UnsubscribeURL(url common.URL) error
+ // PublishServiceDefinition will generate the target url's code info
+ PublishServiceDefinition(url common.URL) error
+ // GetExportedURLs will get the target exported url in metadata
+ GetExportedURLs(serviceInterface string, group string, version string, protocol string) (*skip.SkipList, error)
+ // GetExportedURLs will get the target subscribed url in metadata
+ GetSubscribedURLs() (*skip.SkipList, error)
+ // GetServiceDefinition will get the target service info store in metadata
+ GetServiceDefinition(interfaceName string, group string, version string) (string, error)
+ // GetServiceDefinition will get the target service info store in metadata by service key
+ GetServiceDefinitionByServiceKey(serviceKey string) (string, error)
+ // Version will return the metadata service version
+ Version() string
+}
+
+// BaseMetadataService is used for the common logic for struct who will implement interface MetadataService
+type BaseMetadataService struct {
+}
+
+// ServiceName can get the service's name in meta service , which is application name
+func (mts *BaseMetadataService) ServiceName() (string, error) {
+ return config.GetApplicationConfig().Name, nil
+}
diff --git a/registry/event.go b/registry/event.go
index be9f11d00bb5a70b0d787d15bcdc98471aad0a4b..6f647185cc213b80b9ab25b4702f91b36aa8ad4b 100644
--- a/registry/event.go
+++ b/registry/event.go
@@ -25,6 +25,7 @@ import (
import (
"github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/observer"
"github.com/apache/dubbo-go/remoting"
)
@@ -47,47 +48,9 @@ func (e ServiceEvent) String() string {
return fmt.Sprintf("ServiceEvent{Action{%s}, Path{%s}}", e.Action, e.Service)
}
-// Event is align with Event interface in Java.
-// it's the top abstraction
-// Align with 2.7.5
-type Event interface {
- fmt.Stringer
- GetSource() interface{}
- GetTimestamp() time.Time
-}
-
-// baseEvent is the base implementation of Event
-// You should never use it directly
-type baseEvent struct {
- source interface{}
- timestamp time.Time
-}
-
-// GetSource return the source
-func (b *baseEvent) GetSource() interface{} {
- return b.source
-}
-
-// GetTimestamp return the timestamp when the event is created
-func (b *baseEvent) GetTimestamp() time.Time {
- return b.timestamp
-}
-
-// String return a human readable string representing this event
-func (b *baseEvent) String() string {
- return fmt.Sprintf("baseEvent[source = %#v]", b.source)
-}
-
-func newBaseEvent(source interface{}) *baseEvent {
- return &baseEvent{
- source: source,
- timestamp: time.Now(),
- }
-}
-
// ServiceInstancesChangedEvent represents service instances make some changing
type ServiceInstancesChangedEvent struct {
- baseEvent
+ observer.BaseEvent
ServiceName string
Instances []ServiceInstance
}
@@ -100,9 +63,9 @@ func (s *ServiceInstancesChangedEvent) String() string {
// NewServiceInstancesChangedEvent will create the ServiceInstanceChangedEvent instance
func NewServiceInstancesChangedEvent(serviceName string, instances []ServiceInstance) *ServiceInstancesChangedEvent {
return &ServiceInstancesChangedEvent{
- baseEvent: baseEvent{
- source: serviceName,
- timestamp: time.Now(),
+ BaseEvent: observer.BaseEvent{
+ Source: serviceName,
+ Timestamp: time.Now(),
},
ServiceName: serviceName,
Instances: instances,
diff --git a/registry/event_listener.go b/registry/event_listener.go
index b8d6148442d9e10e210958dead690c4a95b33fb6..0aaa081a4488de5c291485277cdc0fc14ef4e380 100644
--- a/registry/event_listener.go
+++ b/registry/event_listener.go
@@ -18,25 +18,27 @@
package registry
import (
- gxsort "github.com/dubbogo/gost/sort"
+ "reflect"
)
-// EventListener is an new interface used to align with dubbo 2.7.5
-// It contains the Prioritized means that the listener has its priority
-type EventListener interface {
- gxsort.Prioritizer
- // OnEvent handle this event
- OnEvent(e Event) error
-}
-
-// ConditionalEventListener only handle the event which it can handle
-type ConditionalEventListener interface {
- EventListener
- // Accept will make the decision whether it should handle this event
- Accept(e Event) bool
-}
+import (
+ "github.com/apache/dubbo-go/common/observer"
+)
// TODO (implement ConditionalEventListener)
type ServiceInstancesChangedListener struct {
ServiceName string
+ observer.EventListener
+}
+
+func (sicl *ServiceInstancesChangedListener) OnEvent(e observer.Event) error {
+ return nil
+}
+
+func (sicl *ServiceInstancesChangedListener) GetPriority() int {
+ return -1
+}
+
+func (sicl *ServiceInstancesChangedListener) GetEventType() reflect.Type {
+ return reflect.TypeOf(&ServiceInstancesChangedEvent{})
}
diff --git a/registry/inmemory/service_discovery.go b/registry/inmemory/service_discovery.go
new file mode 100644
index 0000000000000000000000000000000000000000..3dac35cd381a7e0e1c0694d6376c85eb7762afc4
--- /dev/null
+++ b/registry/inmemory/service_discovery.go
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package inmemory
+
+import (
+ "github.com/dubbogo/gost/container/set"
+ "github.com/dubbogo/gost/page"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/registry"
+)
+
+const (
+ name = "in-memory"
+)
+
+func init() {
+
+ instance := &InMemoryServiceDiscovery{
+ instances: make(map[string]registry.ServiceInstance, 4),
+ listeners: make([]*registry.ServiceInstancesChangedListener, 0, 2),
+ }
+
+ extension.SetServiceDiscovery(name, func(url *common.URL) (discovery registry.ServiceDiscovery, err error) {
+ return instance, nil
+ })
+}
+
+// InMemoryServiceDiscovery is an implementation based on memory.
+// Usually you will not use this implementation except for tests.
+type InMemoryServiceDiscovery struct {
+ instances map[string]registry.ServiceInstance
+ listeners []*registry.ServiceInstancesChangedListener
+}
+
+func (i *InMemoryServiceDiscovery) String() string {
+ return name
+}
+
+// Destroy doesn't destroy the instance, it just clear the instances
+func (i *InMemoryServiceDiscovery) Destroy() error {
+ // reset to empty
+ i.instances = make(map[string]registry.ServiceInstance, 4)
+ i.listeners = make([]*registry.ServiceInstancesChangedListener, 0, 2)
+ return nil
+}
+
+// Register will store the instance using its id as key
+func (i *InMemoryServiceDiscovery) Register(instance registry.ServiceInstance) error {
+ i.instances[instance.GetId()] = instance
+ return nil
+}
+
+// Update will act like register
+func (i *InMemoryServiceDiscovery) Update(instance registry.ServiceInstance) error {
+ return i.Register(instance)
+}
+
+// Unregister will remove the instance
+func (i *InMemoryServiceDiscovery) Unregister(instance registry.ServiceInstance) error {
+ delete(i.instances, instance.GetId())
+ return nil
+}
+
+// GetDefaultPageSize will return the default page size
+func (i *InMemoryServiceDiscovery) GetDefaultPageSize() int {
+ return registry.DefaultPageSize
+}
+
+// GetServices will return all service names
+func (i *InMemoryServiceDiscovery) GetServices() *gxset.HashSet {
+ result := gxset.NewSet()
+ for _, value := range i.instances {
+ result.Add(value.GetServiceName())
+ }
+ return result
+}
+
+// GetInstances will find out all instances with serviceName
+func (i *InMemoryServiceDiscovery) GetInstances(serviceName string) []registry.ServiceInstance {
+ result := make([]registry.ServiceInstance, 0, len(i.instances))
+ for _, value := range i.instances {
+ if value.GetServiceName() == serviceName {
+ result = append(result, value)
+ }
+ }
+ return result
+}
+
+// GetInstancesByPage will return the part of instances
+func (i *InMemoryServiceDiscovery) GetInstancesByPage(serviceName string, offset int, pageSize int) gxpage.Pager {
+ instances := i.GetInstances(serviceName)
+ // we can not use []registry.ServiceInstance since New(...) received []interface{} as parameter
+ result := make([]interface{}, 0, pageSize)
+ for i := offset; i < len(instances) && i < offset+pageSize; i++ {
+ result = append(result, instances[i])
+ }
+ return gxpage.New(offset, pageSize, result, len(instances))
+}
+
+// GetHealthyInstancesByPage will return the instances
+func (i *InMemoryServiceDiscovery) GetHealthyInstancesByPage(serviceName string, offset int, pageSize int, healthy bool) gxpage.Pager {
+ instances := i.GetInstances(serviceName)
+ // we can not use []registry.ServiceInstance since New(...) received []interface{} as parameter
+ result := make([]interface{}, 0, pageSize)
+ count := 0
+ for i := offset; i < len(instances) && count < pageSize; i++ {
+ if instances[i].IsHealthy() == healthy {
+ result = append(result, instances[i])
+ count++
+ }
+ }
+ return gxpage.New(offset, pageSize, result, len(instances))
+}
+
+// GetRequestInstances will iterate the serviceName and aggregate them
+func (i *InMemoryServiceDiscovery) GetRequestInstances(serviceNames []string, offset int, requestedSize int) map[string]gxpage.Pager {
+ res := make(map[string]gxpage.Pager, len(serviceNames))
+ for _, name := range serviceNames {
+ res[name] = i.GetInstancesByPage(name, offset, requestedSize)
+ }
+ return res
+}
+
+// AddListener will save the listener inside the memory
+func (i *InMemoryServiceDiscovery) AddListener(listener *registry.ServiceInstancesChangedListener) error {
+ i.listeners = append(i.listeners, listener)
+ return nil
+}
+
+// DispatchEventByServiceName will do nothing
+func (i *InMemoryServiceDiscovery) DispatchEventByServiceName(serviceName string) error {
+ return nil
+}
+
+// DispatchEventForInstances will do nothing
+func (i *InMemoryServiceDiscovery) DispatchEventForInstances(serviceName string, instances []registry.ServiceInstance) error {
+ return nil
+}
+
+// DispatchEvent will do nothing
+func (i *InMemoryServiceDiscovery) DispatchEvent(event *registry.ServiceInstancesChangedEvent) error {
+ return nil
+}
diff --git a/registry/inmemory/service_discovery_test.go b/registry/inmemory/service_discovery_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..a934dbabfff7f4041df6dcca77ecc825b3ce391b
--- /dev/null
+++ b/registry/inmemory/service_discovery_test.go
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package inmemory
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/registry"
+)
+
+func TestInMemoryServiceDiscovery(t *testing.T) {
+ discovery, _ := extension.GetServiceDiscovery(name, nil)
+ serviceName := "my-service"
+ err := discovery.Register(®istry.DefaultServiceInstance{
+ ServiceName: serviceName,
+ Id: "1",
+ Healthy: true,
+ })
+ assert.Nil(t, err)
+
+ err = discovery.Register(®istry.DefaultServiceInstance{
+ Id: "2",
+ ServiceName: "mock-service",
+ Healthy: false,
+ })
+
+ assert.Nil(t, err)
+
+ services := discovery.GetServices()
+ assert.Equal(t, 2, services.Size())
+ assert.Equal(t, registry.DefaultPageSize, discovery.GetDefaultPageSize())
+
+ reqInstances := discovery.GetRequestInstances([]string{serviceName, "mock-service"}, 0, 10)
+ assert.Equal(t, 2, len(reqInstances))
+
+ page := discovery.GetInstancesByPage(serviceName, 0, 10)
+ assert.Equal(t, 1, page.GetDataSize())
+
+ discovery.GetHealthyInstancesByPage(serviceName, 0, 10, true)
+ page = discovery.GetInstancesByPage(serviceName, 0, 10)
+ assert.Equal(t, 1, page.GetDataSize())
+
+ err = discovery.AddListener(®istry.ServiceInstancesChangedListener{})
+ assert.Nil(t, err)
+
+ err = discovery.DispatchEvent(®istry.ServiceInstancesChangedEvent{})
+ assert.Nil(t, err)
+
+ err = discovery.DispatchEventForInstances(serviceName, nil)
+ assert.Nil(t, err)
+
+ err = discovery.DispatchEventByServiceName(serviceName)
+ assert.Nil(t, err)
+
+ err = discovery.Unregister(®istry.DefaultServiceInstance{
+ Id: "2",
+ })
+ assert.Nil(t, err)
+
+ services = discovery.GetServices()
+ assert.Equal(t, 1, services.Size())
+
+ err = discovery.Update(®istry.DefaultServiceInstance{
+ Id: "3",
+ })
+ assert.Nil(t, err)
+
+ services = discovery.GetServices()
+ assert.Equal(t, 2, services.Size())
+
+ err = discovery.Destroy()
+ assert.Nil(t, err)
+
+ services = discovery.GetServices()
+ assert.Equal(t, 0, services.Size())
+}
diff --git a/registry/nacos/service_discovery.go b/registry/nacos/service_discovery.go
index 8ef72c1b11dca257103c155534b68cbd522d358f..7d3406cac233dd5293c7522b4f12148fdcdd704e 100644
--- a/registry/nacos/service_discovery.go
+++ b/registry/nacos/service_discovery.go
@@ -280,6 +280,6 @@ func newNacosServiceDiscovery(url *common.URL) (registry.ServiceDiscovery, error
}
return &nacosServiceDiscovery{
nacosBaseRegistry: base,
- group: url.GetParam(constant.NACOS_GROUP, defaultGroup),
+ group: url.GetParam(constant.NACOS_GROUP, defaultGroup),
}, nil
}