Ftr: zk/consul Metadata

.gitignore

@@ -20,15 +20,13 @@ classes
 
 # go mod, go test
 vendor/
-coverage.txt
-
 logs/
 .vscode/
-coverage.txt
 
 # unit test
 remoting/zookeeper/zookeeper-4unittest/
 config_center/zookeeper/zookeeper-4unittest/
 registry/zookeeper/zookeeper-4unittest/
+metadata/report/zookeeper/zookeeper-4unittest/
 registry/consul/agent*
 config_center/apollo/mockDubbog.properties.json

before_ut.bat

@@ -34,4 +34,7 @@ md cluster\router\chain\zookeeper-4unittest\contrib\fatjar
 xcopy /f "%zkJar%" "cluster/router/chain/zookeeper-4unittest/contrib/fatjar/"
 
 md cluster\router\condition\zookeeper-4unittest\contrib\fatjar
-xcopy /f "%zkJar%" "cluster/router/condition/zookeeper-4unittest/contrib/fatjar/"
+xcopy /f "%zkJar%" "cluster/router/condition/zookeeper-4unittest/contrib/fatjar/"
+
+md metadata\report\zookeeper\zookeeper-4unittest\contrib\fatjar
+xcopy /f "%zkJar%" "metadata/report/zookeeper/zookeeper-4unittest/contrib/fatjar/"

before_ut.sh

@@ -25,13 +25,16 @@ if [ ! -f "${zkJar}" ]; then
 fi
 
 mkdir -p config_center/zookeeper/zookeeper-4unittest/contrib/fatjar
-cp ${zkJar} config_center/zookeeper/zookeeper-4unittest/contrib/fatjar/
+cp ${zkJar} config_center/zookeeper/zookeeper-4unittest/contrib/fatjar
 
 mkdir -p registry/zookeeper/zookeeper-4unittest/contrib/fatjar
-cp ${zkJar} registry/zookeeper/zookeeper-4unittest/contrib/fatjar/
+cp ${zkJar} registry/zookeeper/zookeeper-4unittest/contrib/fatjar
 
 mkdir -p cluster/router/chain/zookeeper-4unittest/contrib/fatjar
 cp ${zkJar} cluster/router/chain/zookeeper-4unittest/contrib/fatjar
 
 mkdir -p cluster/router/condition/zookeeper-4unittest/contrib/fatjar
-cp ${zkJar} cluster/router/condition/zookeeper-4unittest/contrib/fatjar
+cp ${zkJar} cluster/router/condition/zookeeper-4unittest/contrib/fatjar
+
+mkdir -p metadata/report/zookeeper/zookeeper-4unittest/contrib/fatjar
+cp ${zkJar} metadata/report/zookeeper/zookeeper-4unittest/contrib/fatjar

metadata/report/consul/report.go

@@ -0,0 +1,135 @@
+/*
+ * 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 consul
+
+import (
+ consul "github.com/hashicorp/consul/api"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/metadata/identifier"
+ "github.com/apache/dubbo-go/metadata/report"
+ "github.com/apache/dubbo-go/metadata/report/factory"
+)
+
+var (
+ emptyStrSlice = make([]string, 0)
+)
+
+func init() {
+ extension.SetMetadataReportFactory("consul", func() factory.MetadataReportFactory {
+ return &consulMetadataReportFactory{}
+ })
+}
+
+// consulMetadataReport is the implementation of
+// MetadataReport based on consul.
+type consulMetadataReport struct {
+ client *consul.Client
+}
+
+// StoreProviderMetadata stores the metadata.
+func (m *consulMetadataReport) StoreProviderMetadata(providerIdentifier *identifier.MetadataIdentifier, serviceDefinitions string) error {
+ kv := &consul.KVPair{Key: providerIdentifier.GetIdentifierKey(), Value: []byte(serviceDefinitions)}
+ _, err := m.client.KV().Put(kv, nil)
+ return err
+}
+
+// StoreConsumerMetadata stores the metadata.
+func (m *consulMetadataReport) StoreConsumerMetadata(consumerMetadataIdentifier *identifier.MetadataIdentifier, serviceParameterString string) error {
+ kv := &consul.KVPair{Key: consumerMetadataIdentifier.GetIdentifierKey(), Value: []byte(serviceParameterString)}
+ _, err := m.client.KV().Put(kv, nil)
+ return err
+}
+
+// SaveServiceMetadata saves the metadata.
+func (m *consulMetadataReport) SaveServiceMetadata(metadataIdentifier *identifier.ServiceMetadataIdentifier, url common.URL) error {
+ kv := &consul.KVPair{Key: metadataIdentifier.GetIdentifierKey(), Value: []byte(url.String())}
+ _, err := m.client.KV().Put(kv, nil)
+ return err
+}
+
+// RemoveServiceMetadata removes the metadata.
+func (m *consulMetadataReport) RemoveServiceMetadata(metadataIdentifier *identifier.ServiceMetadataIdentifier) error {
+ k := metadataIdentifier.GetIdentifierKey()
+ _, err := m.client.KV().Delete(k, nil)
+ return err
+}
+
+// GetExportedURLs gets the urls.
+func (m *consulMetadataReport) GetExportedURLs(metadataIdentifier *identifier.ServiceMetadataIdentifier) ([]string, error) {
+ k := metadataIdentifier.GetIdentifierKey()
+ kv, _, err := m.client.KV().Get(k, nil)
+ if err != nil {
+ return emptyStrSlice, err
+ }
+
+ if kv == nil {
+ return emptyStrSlice, nil
+ }
+ return []string{string(kv.Value)}, nil
+}
+
+// SaveSubscribedData saves the urls.
+func (m *consulMetadataReport) SaveSubscribedData(subscriberMetadataIdentifier *identifier.SubscriberMetadataIdentifier, urls string) error {
+ kv := &consul.KVPair{Key: subscriberMetadataIdentifier.GetIdentifierKey(), Value: []byte(urls)}
+ _, err := m.client.KV().Put(kv, nil)
+ return err
+}
+
+// GetSubscribedURLs gets the urls.
+func (m *consulMetadataReport) GetSubscribedURLs(subscriberMetadataIdentifier *identifier.SubscriberMetadataIdentifier) ([]string, error) {
+ k := subscriberMetadataIdentifier.GetIdentifierKey()
+ kv, _, err := m.client.KV().Get(k, nil)
+ if err != nil {
+ return emptyStrSlice, err
+ }
+
+ if kv == nil {
+ return emptyStrSlice, nil
+ }
+ return []string{string(kv.Value)}, nil
+}
+
+// GetServiceDefinition gets the service definition.
+func (m *consulMetadataReport) GetServiceDefinition(metadataIdentifier *identifier.MetadataIdentifier) (string, error) {
+ k := metadataIdentifier.GetIdentifierKey()
+ kv, _, err := m.client.KV().Get(k, nil)
+ if err != nil {
+ return "", err
+ }
+
+ if kv == nil {
+ return "", nil
+ }
+ return string(kv.Value), nil
+}
+
+type consulMetadataReportFactory struct {
+}
+
+func (mf *consulMetadataReportFactory) CreateMetadataReport(url *common.URL) report.MetadataReport {
+ config := &consul.Config{Address: url.Location}
+ client, err := consul.NewClient(config)
+ if err != nil {
+ panic(err)
+ }
+ return &consulMetadataReport{client: client}
+}

metadata/report/consul/report_test.go

@@ -0,0 +1,166 @@
+/*
+ * 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 consul
+
+import (
+ "encoding/json"
+ "net/url"
+ "strconv"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/constant"
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/metadata/identifier"
+ "github.com/apache/dubbo-go/metadata/report"
+ "github.com/apache/dubbo-go/remoting/consul"
+)
+
+func newProviderRegistryUrl(host string, port int) *common.URL {
+ url1 := common.NewURLWithOptions(
+ common.WithIp(host),
+ common.WithPort(strconv.Itoa(port)),
+ common.WithParams(url.Values{}),
+ common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER)),
+ )
+ return url1
+}
+
+func newBaseMetadataIdentifier(side string) *identifier.BaseMetadataIdentifier {
+ return &identifier.BaseMetadataIdentifier{
+ ServiceInterface: "org.apache.HelloWorld",
+ Version: "1.0.0",
+ Group: "group",
+ Side: side,
+ }
+}
+
+func newMetadataIdentifier(side string) *identifier.MetadataIdentifier {
+ return &identifier.MetadataIdentifier{
+ Application: "application",
+ BaseMetadataIdentifier: *newBaseMetadataIdentifier(side),
+ }
+}
+
+func newServiceMetadataIdentifier(side string) *identifier.ServiceMetadataIdentifier {
+ return &identifier.ServiceMetadataIdentifier{
+ Revision: "1.0",
+ Protocol: "dubbo",
+ BaseMetadataIdentifier: *newBaseMetadataIdentifier(side),
+ }
+}
+
+func newSubscribeMetadataIdentifier(side string) *identifier.SubscriberMetadataIdentifier {
+ return &identifier.SubscriberMetadataIdentifier{
+ Revision: "1.0",
+ MetadataIdentifier: *newMetadataIdentifier(side),
+ }
+}
+
+type consulMetadataReportTestSuite struct {
+ t *testing.T
+ m report.MetadataReport
+}
+
+func newConsulMetadataReportTestSuite(t *testing.T, m report.MetadataReport) *consulMetadataReportTestSuite {
+ return &consulMetadataReportTestSuite{t: t, m: m}
+}
+
+func (suite *consulMetadataReportTestSuite) testStoreProviderMetadata() {
+ providerMi := newMetadataIdentifier("provider")
+ providerMeta := "provider"
+ err := suite.m.StoreProviderMetadata(providerMi, providerMeta)
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testStoreConsumerMetadata() {
+ consumerMi := newMetadataIdentifier("consumer")
+ consumerMeta := "consumer"
+ err := suite.m.StoreProviderMetadata(consumerMi, consumerMeta)
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testSaveServiceMetadata(url common.URL) {
+ serviceMi := newServiceMetadataIdentifier("provider")
+ err := suite.m.SaveServiceMetadata(serviceMi, url)
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testRemoveServiceMetadata() {
+ serviceMi := newServiceMetadataIdentifier("provider")
+ err := suite.m.RemoveServiceMetadata(serviceMi)
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testGetExportedURLs() {
+ serviceMi := newServiceMetadataIdentifier("provider")
+ urls, err := suite.m.GetExportedURLs(serviceMi)
+ assert.Equal(suite.t, 1, len(urls))
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testSaveSubscribedData(url common.URL) {
+ subscribeMi := newSubscribeMetadataIdentifier("provider")
+ urls := []string{url.String()}
+ bytes, _ := json.Marshal(urls)
+ err := suite.m.SaveSubscribedData(subscribeMi, string(bytes))
+ assert.Nil(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testGetSubscribedURLs() {
+ subscribeMi := newSubscribeMetadataIdentifier("provider")
+ urls, err := suite.m.GetSubscribedURLs(subscribeMi)
+ assert.Equal(suite.t, 1, len(urls))
+ assert.NoError(suite.t, err)
+}
+
+func (suite *consulMetadataReportTestSuite) testGetServiceDefinition() {
+ providerMi := newMetadataIdentifier("provider")
+ providerMeta, err := suite.m.GetServiceDefinition(providerMi)
+ assert.Equal(suite.t, "provider", providerMeta)
+ assert.NoError(suite.t, err)
+}
+
+func test1(t *testing.T) {
+ consulAgent := consul.NewConsulAgent(t, 8500)
+ defer consulAgent.Close()
+
+ url := newProviderRegistryUrl("localhost", 8500)
+ mf := extension.GetMetadataReportFactory("consul")
+ m := mf.CreateMetadataReport(url)
+
+ suite := newConsulMetadataReportTestSuite(t, m)
+ suite.testStoreProviderMetadata()
+ suite.testStoreConsumerMetadata()
+ suite.testSaveServiceMetadata(*url)
+ suite.testGetExportedURLs()
+ suite.testRemoveServiceMetadata()
+ suite.testSaveSubscribedData(*url)
+ suite.testGetSubscribedURLs()
+ suite.testGetServiceDefinition()
+}
+
+func TestConsulMetadataReport(t *testing.T) {
+ t.Run("test1", test1)
+}