Invoke dubbo client success

config/config_loader.go

@@ -148,7 +148,7 @@ func loadConsumerConfig() {
 					if count > maxWait {
 						errMsg := fmt.Sprintf("Failed to check the status of the service %v . No provider available for the service to the consumer use dubbo version %v", refconfig.InterfaceName, constant.Version)
 						logger.Error(errMsg)
-						// panic(errMsg)
+						panic(errMsg)
 					}
 					time.Sleep(time.Second * 1)
 					break

metadata/service/inmemory/metadata_service_proxy_factory.go

@@ -18,7 +18,12 @@
 package inmemory
 
 import (
+	"encoding/json"
+
+	"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/common/logger"
 	"github.com/apache/dubbo-go/metadata/service"
 	"github.com/apache/dubbo-go/registry"
 )
@@ -31,5 +36,54 @@ func init() {
 }
 
 func createProxy(ins registry.ServiceInstance) service.MetadataService {
-	return nil
+	urls := buildStandardMetadataServiceURL(ins)
+	if len(urls) == 0 {
+		logger.Errorf("metadata service urls not found, %v", ins)
+		return nil
+	}
+	p := extension.GetProtocol(urls[0].Protocol)
+	invoker := p.Refer(*urls[0])
+	return &MetadataServiceProxy{invkr: invoker}
+}
+
+// buildStandardMetadataServiceURL will use standard format to build the metadata service url.
+// Now we don't need to support spring-cloud format metadata service url.
+//
+func buildStandardMetadataServiceURL(ins registry.ServiceInstance) []*common.URL {
+	ps := getMetadataServiceUrlParams(ins)
+	res := make([]*common.URL, 0, len(ps))
+	sn := ins.GetServiceName()
+	host := ins.GetHost()
+	for protocol, params := range ps {
+
+		convertedParams := make(map[string][]string, len(params))
+		for k, v := range params {
+			convertedParams[k] = []string{v}
+		}
+
+		u := common.NewURLWithOptions(common.WithIp(host),
+			common.WithPath(constant.METADATA_SERVICE_NAME),
+			common.WithProtocol(protocol),
+			common.WithPort(params[constant.PORT_KEY]),
+			common.WithParams(convertedParams),
+			common.WithParamsValue(constant.GROUP_KEY, sn))
+		res = append(res, u)
+	}
+	return res
+}
+
+// getMetadataServiceUrlParams this will convert the metadata service url parameters to map structure
+// it looks like:
+// {"dubbo":{"timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"20880"}}
+func getMetadataServiceUrlParams(ins registry.ServiceInstance) map[string]map[string]string {
+	ps := ins.GetMetadata()
+	res := make(map[string]map[string]string, 2)
+	if str, ok := ps[constant.METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME]; ok && len(str) > 0 {
+
+		err := json.Unmarshal([]byte(str), &res)
+		if err != nil {
+			logger.Errorf("could not parse the metadata service url parameters to map", err)
+		}
+	}
+	return res
 }

registry/servicediscovery/proxy/metadata_service_proxy_factory.go → metadata/service/inmemory/metadata_service_proxy_factory_test.go

@@ -15,9 +15,18 @@
  * limitations under the License.
  */
 
-package proxy
+package inmemory
 
 import (
-	"github.com/apache/dubbo-go/metadata/service"
-	"github.com/apache/dubbo-go/registry"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
+
+func TestMetadataService_GetMetadataServiceUrlParams(t *testing.T) {
+	str := `{"dubbo":{"timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"20880"}}`
+	tmp := make(map[string]map[string]string)
+	err := json.Unmarshal([]byte(str), &tmp)
+	assert.Nil(t, err)
+}

metadata/service/inmemory/service_proxy.go

+/*
+ * 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"
+	"reflect"
+
+	"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/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+type MetadataServiceProxy struct {
+	invkr protocol.Invoker
+}
+
+func (m *MetadataServiceProxy) GetExportedURLs(serviceInterface string, group string, version string, protocol string) ([]common.URL, error) {
+
+	siV := reflect.ValueOf(serviceInterface)
+	gV := reflect.ValueOf(group)
+	vV := reflect.ValueOf(version)
+	pV := reflect.ValueOf(protocol)
+
+	inv := invocation.NewRPCInvocationWithOptions(invocation.WithMethodName("getExportedURLs"),
+		invocation.WithArguments([]interface{}{siV.Interface(), gV.Interface(), vV.Interface(), pV.Interface()}),
+		invocation.WithReply(reflect.ValueOf(&[]interface{}{}).Interface()),
+		invocation.WithAttachments(map[string]string{constant.ASYNC_KEY: "false"}),
+		invocation.WithParameterValues([]reflect.Value{siV, gV, vV, pV}))
+
+	res := m.invkr.Invoke(context.Background(), inv)
+	if res.Error() != nil {
+		logger.Errorf("could not get the metadata service from remote provider: %v", res.Error())
+	}
+
+	urlStrs := res.Result().(*[]interface{})
+
+	ret := make([]common.URL, 0, len(*urlStrs))
+
+	for _, s := range *urlStrs {
+		u, err := common.NewURL(s.(string))
+		if err != nil {
+			logger.Errorf("could not convert the string to URL: %s", s)
+			continue
+		}
+		ret = append(ret, u)
+	}
+	return ret, nil
+}
+
+func (m *MetadataServiceProxy) Reference() string {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) ServiceName() (string, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) ExportURL(url common.URL) (bool, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) UnexportURL(url common.URL) error {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) SubscribeURL(url common.URL) (bool, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) UnsubscribeURL(url common.URL) error {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) PublishServiceDefinition(url common.URL) error {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) GetSubscribedURLs() ([]common.URL, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) GetServiceDefinition(interfaceName string, group string, version string) (string, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) GetServiceDefinitionByServiceKey(serviceKey string) (string, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) RefreshMetadata(exportedRevision string, subscribedRevision string) (bool, error) {
+	panic("implement me")
+}
+
+func (m *MetadataServiceProxy) Version() (string, error) {
+	panic("implement me")
+}
+
+type MetadataServiceStub struct {
+	GetExportedURLs func(serviceInterface string, group string, version string, protocol string) ([]interface{}, error)
+}
+
+func (m *MetadataServiceStub) Reference() string {
+	return constant.METADATA_SERVICE_NAME
+}

protocol/dubbo/client.go

@@ -280,7 +280,7 @@ func (c *Client) call(ct CallType, request *Request, response *Response, callbac
 	}
 
 	select {
-	case <-getty.GetTimeWheel().After(c.opts.RequestTimeout):
+	case <-getty.GetTimeWheel().After(3 * time.Second):
 		c.removePendingResponse(SequenceType(rsp.seq))
 		return perrors.WithStack(errClientReadTimeout)
 	case <-rsp.done:

registry/servicediscovery/proxy/service_proxy.go

-/*
- * 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 proxy

registry/servicediscovery/service_discovery_registry.go

@@ -501,6 +501,10 @@ func (s *serviceDiscoveryRegistry) initRevisionExportedURLsByInst(serviceInstanc
 	serviceName := serviceInstance.GetServiceName()
 	revision := getExportedServicesRevision(serviceInstance)
 	revisionExportedURLsMap := s.serviceRevisionExportedURLsCache[serviceName]
+	if revisionExportedURLsMap == nil {
+		revisionExportedURLsMap = make(map[string][]common.URL)
+		s.serviceRevisionExportedURLsCache[serviceName] = revisionExportedURLsMap
+	}
 	revisionExportedURLs := revisionExportedURLsMap[revision]
 	firstGet := false
 	if revisionExportedURLs == nil || len(revisionExportedURLs) == 0 {
@@ -558,8 +562,8 @@ func (s *serviceDiscoveryRegistry) cloneExportedURLs(url common.URL, serviceInsa
 		for _, u := range templateExportURLs {
 			port := strconv.Itoa(getProtocolPort(serviceInstance, u.Protocol))
 			if u.Location != host || u.Port != port {
-				u.Port = port     // reset port
-				u.Location = host // reset host
+				u.Port = port                  // reset port
+				u.Location = host + ":" + port // reset host
 			}
 
 			cloneUrl := u.CloneExceptParams(removeParamSet)