diff --git a/common/rpc_service.go b/common/rpc_service.go
index eb68a49f72a35d31a5908071ab383a094a591a21..5ed4df6d717db31021d8585a88d5576f59756f8a 100644
--- a/common/rpc_service.go
+++ b/common/rpc_service.go
@@ -169,7 +169,7 @@ func (sm *serviceMap) GetService(protocol, name string) *Service {
return nil
}
-// GetInterface gets an interface defination by interface name
+// GetInterface gets an interface definition by interface name
func (sm *serviceMap) GetInterface(interfaceName string) []*Service {
sm.mutex.RLock()
defer sm.mutex.RUnlock()
diff --git a/config/config_loader.go b/config/config_loader.go
index 75b82628d68a23e575cfa637b3603d09e09ea9d6..c66e526921e7e5ab017105e2f4ea2baa62563205 100644
--- a/config/config_loader.go
+++ b/config/config_loader.go
@@ -141,19 +141,18 @@ func loadConsumerConfig() {
// wait for invoker is available, if wait over default 3s, then panic
var count int
- checkok := true
for {
+ checkok := true
for _, refconfig := range consumerConfig.References {
if (refconfig.Check != nil && *refconfig.Check) ||
(refconfig.Check == nil && consumerConfig.Check != nil && *consumerConfig.Check) ||
(refconfig.Check == nil && consumerConfig.Check == nil) { // default to true
- if refconfig.invoker != nil &&
- !refconfig.invoker.IsAvailable() {
+ if refconfig.invoker != nil && !refconfig.invoker.IsAvailable() {
checkok = false
count++
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)
+ 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)
}
@@ -161,14 +160,13 @@ func loadConsumerConfig() {
break
}
if refconfig.invoker == nil {
- logger.Warnf("The interface %s invoker not exist , may you should check your interface config.", refconfig.InterfaceName)
+ logger.Warnf("The interface %s invoker not exist, may you should check your interface config.", refconfig.InterfaceName)
}
}
}
if checkok {
break
}
- checkok = true
}
}
diff --git a/config/consumer_config.go b/config/consumer_config.go
index 9d283eeca7bbaf5a82f71357853c6b53560b2fe4..c8083603e12570a4492dd63a749adb4aa89663c8 100644
--- a/config/consumer_config.go
+++ b/config/consumer_config.go
@@ -60,8 +60,8 @@ type ConsumerConfig struct {
References map[string]*ReferenceConfig `yaml:"references" json:"references,omitempty" property:"references"`
ProtocolConf interface{} `yaml:"protocol_conf" json:"protocol_conf,omitempty" property:"protocol_conf"`
- FilterConf interface{} `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf" `
- ShutdownConfig *ShutdownConfig `yaml:"shutdown_conf" json:"shutdown_conf,omitempty" property:"shutdown_conf" `
+ FilterConf interface{} `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf"`
+ ShutdownConfig *ShutdownConfig `yaml:"shutdown_conf" json:"shutdown_conf,omitempty" property:"shutdown_conf"`
ConfigType map[string]string `yaml:"config_type" json:"config_type,omitempty" property:"config_type"`
}
diff --git a/config/provider_config.go b/config/provider_config.go
index c710e48dc233a62837b31a89828e9c612eaff093..fcb429b6400936f72c09035968b4d4c72387246c 100644
--- a/config/provider_config.go
+++ b/config/provider_config.go
@@ -43,9 +43,9 @@ type ProviderConfig struct {
ProxyFactory string `yaml:"proxy_factory" default:"default" json:"proxy_factory,omitempty" property:"proxy_factory"`
Services map[string]*ServiceConfig `yaml:"services" json:"services,omitempty" property:"services"`
Protocols map[string]*ProtocolConfig `yaml:"protocols" json:"protocols,omitempty" property:"protocols"`
- ProtocolConf interface{} `yaml:"protocol_conf" json:"protocol_conf,omitempty" property:"protocol_conf" `
- FilterConf interface{} `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf" `
- ShutdownConfig *ShutdownConfig `yaml:"shutdown_conf" json:"shutdown_conf,omitempty" property:"shutdown_conf" `
+ ProtocolConf interface{} `yaml:"protocol_conf" json:"protocol_conf,omitempty" property:"protocol_conf"`
+ FilterConf interface{} `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf"`
+ ShutdownConfig *ShutdownConfig `yaml:"shutdown_conf" json:"shutdown_conf,omitempty" property:"shutdown_conf"`
ConfigType map[string]string `yaml:"config_type" json:"config_type,omitempty" property:"config_type"`
Registry *RegistryConfig `yaml:"registry" json:"registry,omitempty" property:"registry"`
diff --git a/config/service_config.go b/config/service_config.go
index 54383e4791bf0d0749aea08d7ba9a613e4cfe70b..8968bac88f55c614841462517c5a5a4ecbe6f594 100644
--- a/config/service_config.go
+++ b/config/service_config.go
@@ -133,7 +133,7 @@ func getRandomPort(protocolConfigs []*ProtocolConfig) *list.List {
tcp, err := gxnet.ListenOnTCPRandomPort(proto.Ip)
if err != nil {
- panic(perrors.New(fmt.Sprintf("Get tcp port error,err is {%v}", err)))
+ panic(perrors.New(fmt.Sprintf("Get tcp port error, err is {%v}", err)))
}
defer tcp.Close()
ports.PushBack(strings.Split(tcp.Addr().String(), ":")[1])
@@ -145,14 +145,14 @@ func getRandomPort(protocolConfigs []*ProtocolConfig) *list.List {
func (c *ServiceConfig) Export() error {
// TODO: config center start here
- // TODO:delay export
+ // TODO: delay export
if c.unexported != nil && c.unexported.Load() {
- err := perrors.Errorf("The service %v has already unexported! ", c.InterfaceName)
+ err := perrors.Errorf("The service %v has already unexported!", c.InterfaceName)
logger.Errorf(err.Error())
return err
}
if c.unexported != nil && c.exported.Load() {
- logger.Warnf("The service %v has already exported! ", c.InterfaceName)
+ logger.Warnf("The service %v has already exported!", c.InterfaceName)
return nil
}
@@ -160,23 +160,23 @@ func (c *ServiceConfig) Export() error {
urlMap := c.getUrlMap()
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)
+ logger.Warnf("The service %v's '%v' protocols don't has right protocolConfigs", c.InterfaceName, c.Protocol)
return nil
}
ports := getRandomPort(protocolConfigs)
nextPort := ports.Front()
+ proxyFactory := extension.GetProxyFactory(providerConfig.ProxyFactory)
for _, proto := range protocolConfigs {
// registry the service reflect
methods, err := common.ServiceMap.Register(c.InterfaceName, proto.Name, c.rpcService)
if err != nil {
- formatErr := perrors.Errorf("The service %v export the protocol %v error! Error message is %v .", c.InterfaceName, proto.Name, err.Error())
+ formatErr := perrors.Errorf("The service %v export the protocol %v error! Error message is %v.", c.InterfaceName, proto.Name, err.Error())
logger.Errorf(formatErr.Error())
return formatErr
}
port := proto.Port
-
if len(proto.Port) == 0 {
port = nextPort.Value.(string)
nextPort = nextPort.Next()
@@ -196,33 +196,31 @@ func (c *ServiceConfig) Export() error {
ivkURL.AddParam(constant.Tagkey, c.Tag)
}
- var exporter protocol.Exporter
-
if len(regUrls) > 0 {
+ c.cacheMutex.Lock()
+ if c.cacheProtocol == nil {
+ logger.Infof(fmt.Sprintf("First load the registry protocol, url is {%v}!", ivkURL))
+ c.cacheProtocol = extension.GetProtocol("registry")
+ }
+ c.cacheMutex.Unlock()
+
for _, regUrl := range regUrls {
regUrl.SubURL = ivkURL
-
- c.cacheMutex.Lock()
- if c.cacheProtocol == nil {
- logger.Infof(fmt.Sprintf("First load the registry protocol , url is {%v}!", ivkURL))
- c.cacheProtocol = extension.GetProtocol("registry")
- }
- c.cacheMutex.Unlock()
-
- invoker := extension.GetProxyFactory(providerConfig.ProxyFactory).GetInvoker(*regUrl)
- exporter = c.cacheProtocol.Export(invoker)
+ invoker := proxyFactory.GetInvoker(*regUrl)
+ 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)))
+ return perrors.New(fmt.Sprintf("Registry protocol new exporter error, registry is {%v}, url is {%v}", regUrl, ivkURL))
}
+ c.exporters = append(c.exporters, exporter)
}
} else {
- invoker := extension.GetProxyFactory(providerConfig.ProxyFactory).GetInvoker(*ivkURL)
- exporter = extension.GetProtocol(protocolwrapper.FILTER).Export(invoker)
+ invoker := proxyFactory.GetInvoker(*ivkURL)
+ 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)))
+ return perrors.New(fmt.Sprintf("Filter protocol without registry new exporter error, url is {%v}", ivkURL))
}
+ c.exporters = append(c.exporters, exporter)
}
- c.exporters = append(c.exporters, exporter)
}
c.exported.Store(true)
return nil
@@ -314,7 +312,6 @@ func (c *ServiceConfig) getUrlMap() url.Values {
urlMap.Set(constant.EXECUTE_LIMIT_KEY, v.ExecuteLimit)
urlMap.Set(constant.EXECUTE_REJECTED_EXECUTION_HANDLER_KEY, v.ExecuteLimitRejectedHandler)
-
}
return urlMap
diff --git a/metadata/service/exporter/configurable/exporter.go b/metadata/service/exporter/configurable/exporter.go
index f8b4b0c0174cb0e5a8753b814f89ed4d332e2fbe..5a930c5e953aead15e4673f1b9537197128a7b35 100644
--- a/metadata/service/exporter/configurable/exporter.go
+++ b/metadata/service/exporter/configurable/exporter.go
@@ -48,7 +48,6 @@ func NewMetadataServiceExporter(metadataService service.MetadataService) exporte
// Export will export the metadataService
func (exporter *MetadataServiceExporter) Export() error {
if !exporter.IsExported() {
-
serviceConfig := config.NewServiceConfig(constant.SIMPLE_METADATA_SERVICE_NAME, context.Background())
serviceConfig.Protocol = constant.DEFAULT_PROTOCOL
serviceConfig.Protocols = map[string]*config.ProtocolConfig{
diff --git a/protocol/jsonrpc/jsonrpc_protocol.go b/protocol/jsonrpc/jsonrpc_protocol.go
index 90a6bf5ef7aa017488f723804b22cc613850bcf2..1778d99d4078058dc43a630b89b9d256350da0f3 100644
--- a/protocol/jsonrpc/jsonrpc_protocol.go
+++ b/protocol/jsonrpc/jsonrpc_protocol.go
@@ -59,7 +59,7 @@ func NewJsonrpcProtocol() *JsonrpcProtocol {
}
}
-// Export JSON RPC service for remote invocation
+// Export JSON RPC service for remote invocation
func (jp *JsonrpcProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
url := invoker.GetUrl()
serviceKey := strings.TrimPrefix(url.Path, "/")
diff --git a/protocol/jsonrpc/server.go b/protocol/jsonrpc/server.go
index aa458a1614df29997b05ac4462200f9e9ffffc25..29eba0223eb49f2cdf91d20ac023b87fa2717f31 100644
--- a/protocol/jsonrpc/server.go
+++ b/protocol/jsonrpc/server.go
@@ -127,10 +127,10 @@ func (s *Server) handlePkg(conn net.Conn) {
}
reqBody, err := ioutil.ReadAll(r.Body)
+ r.Body.Close()
if err != nil {
return
}
- r.Body.Close()
reqHeader := make(map[string]string)
for k := range r.Header {
@@ -263,8 +263,7 @@ func (s *Server) Stop() {
})
}
-func serveRequest(ctx context.Context,
- header map[string]string, body []byte, conn net.Conn) error {
+func serveRequest(ctx context.Context, header map[string]string, body []byte, conn net.Conn) error {
sendErrorResp := func(header map[string]string, body []byte) error {
rsp := &http.Response{
Header: make(http.Header),
@@ -324,13 +323,12 @@ func serveRequest(ctx context.Context,
if err == io.EOF || err == io.ErrUnexpectedEOF {
return perrors.WithStack(err)
}
-
return perrors.New("server cannot decode request: " + err.Error())
}
+
path := header["Path"]
methodName := codec.req.Method
if len(path) == 0 || len(methodName) == 0 {
- codec.ReadBody(nil)
return perrors.New("service/method request ill-formed: " + path + "/" + methodName)
}
diff --git a/protocol/protocolwrapper/protocol_filter_wrapper.go b/protocol/protocolwrapper/protocol_filter_wrapper.go
index 87f90d3b7c73328a30e883da8251ac8364a63b2d..4b2702b99f9793d4e567de65bd9555fb20381b1d 100644
--- a/protocol/protocolwrapper/protocol_filter_wrapper.go
+++ b/protocol/protocolwrapper/protocol_filter_wrapper.go
@@ -68,21 +68,19 @@ func (pfw *ProtocolFilterWrapper) Destroy() {
}
func buildInvokerChain(invoker protocol.Invoker, key string) protocol.Invoker {
- filtName := invoker.GetUrl().GetParam(key, "")
- if filtName == "" {
+ filterName := invoker.GetUrl().GetParam(key, "")
+ if filterName == "" {
return invoker
}
- filtNames := strings.Split(filtName, ",")
- next := invoker
+ filterNames := strings.Split(filterName, ",")
// The order of filters is from left to right, so loading from right to left
-
- for i := len(filtNames) - 1; i >= 0; i-- {
- flt := extension.GetFilter(filtNames[i])
+ next := invoker
+ for i := len(filterNames) - 1; i >= 0; i-- {
+ flt := extension.GetFilter(filterNames[i])
fi := &FilterInvoker{next: next, invoker: invoker, filter: flt}
next = fi
}
-
return next
}
diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go
index 659afe86b1c8f06773a11332c8b81e6be0ac824a..5b2a5d66f67991cc6e7ead610b3259bb5962870d 100644
--- a/registry/protocol/protocol.go
+++ b/registry/protocol/protocol.go
@@ -22,6 +22,7 @@ import (
"strings"
"sync"
)
+
import (
gxset "github.com/dubbogo/gost/container/set"
)
@@ -54,9 +55,10 @@ var (
type registryProtocol struct {
invokers []protocol.Invoker
- // Registry Map<RegistryAddress, Registry>
+ // Registry Map<RegistryAddress, Registry>
registries *sync.Map
- // To solve the problem of RMI repeated exposure port conflicts, the services that have been exposed are no longer exposed.
+ // To solve the problem of RMI repeated exposure port conflicts,
+ // the services that have been exposed are no longer exposed.
// providerurl <--> exporter
bounds *sync.Map
overrideListeners *sync.Map
@@ -100,7 +102,6 @@ func getUrlToRegistry(providerUrl *common.URL, registryUrl *common.URL) *common.
// filterHideKey filter the parameters that do not need to be output in url(Starting with .)
func filterHideKey(url *common.URL) *common.URL {
-
// be careful params maps in url is map type
removeSet := gxset.NewSet()
for k := range url.GetParams() {
@@ -139,7 +140,6 @@ func (proto *registryProtocol) Refer(url common.URL) protocol.Invoker {
}
var reg registry.Registry
-
if regI, loaded := proto.registries.Load(registryUrl.Key()); !loaded {
reg = getRegistry(®istryUrl)
proto.registries.Store(registryUrl.Key(), reg)
@@ -150,7 +150,7 @@ func (proto *registryProtocol) Refer(url common.URL) protocol.Invoker {
// new registry directory for store service url from registry
directory, err := extension.GetDefaultRegistryDirectory(®istryUrl, reg)
if err != nil {
- logger.Errorf("consumer service %v create registry directory error, error message is %s, and will return nil invoker!",
+ logger.Errorf("consumer service %v create registry directory error, error message is %s, and will return nil invoker!",
serviceUrl.String(), err.Error())
return nil
}
@@ -163,7 +163,6 @@ func (proto *registryProtocol) Refer(url common.URL) protocol.Invoker {
// new cluster invoker
cluster := extension.GetCluster(serviceUrl.GetParam(constant.CLUSTER_KEY, constant.DEFAULT_CLUSTER))
-
invoker := cluster.Join(directory)
proto.invokers = append(proto.invokers, invoker)
return invoker
@@ -204,7 +203,7 @@ func (proto *registryProtocol) Export(invoker protocol.Invoker) protocol.Exporte
}
key := getCacheKey(providerUrl)
- logger.Infof("The cached exporter keys is %v !", key)
+ logger.Infof("The cached exporter keys is %v!", key)
cachedExporter, loaded := proto.bounds.Load(key)
if loaded {
logger.Infof("The exporter has been cached, and will return cached exporter!")
@@ -217,7 +216,6 @@ func (proto *registryProtocol) Export(invoker protocol.Invoker) protocol.Exporte
go reg.Subscribe(overriderUrl, overrideSubscribeListener)
return cachedExporter.(protocol.Exporter)
-
}
func (proto *registryProtocol) reExport(invoker protocol.Invoker, newUrl *common.URL) {
@@ -229,7 +227,6 @@ func (proto *registryProtocol) reExport(invoker protocol.Invoker, newUrl *common
proto.bounds.Delete(key)
proto.Export(wrappedNewInvoker)
// TODO: unregister & unsubscribe
-
}
}
@@ -366,7 +363,7 @@ func (proto *registryProtocol) Destroy() {
func getRegistryUrl(invoker protocol.Invoker) *common.URL {
// here add * for return a new url
url := invoker.GetUrl()
- // if the protocol == registry ,set protocol the registry value in url.params
+ // if the protocol == registry, set protocol the registry value in url.params
if url.Protocol == constant.REGISTRY_PROTOCOL {
protocol := url.GetParam(constant.REGISTRY_KEY, "")
url.Protocol = protocol
diff --git a/registry/registry.go b/registry/registry.go
index 5e77eab186680671f27b44bbe2e6a6b964a28721..855b487d4766d011eaa9d97d06b672df075e60b9 100644
--- a/registry/registry.go
+++ b/registry/registry.go
@@ -30,30 +30,38 @@ import (
// Registry Extension - Registry
type Registry interface {
common.Node
- //used for service provider calling , register services to registry
- //And it is also used for service consumer calling , register services cared about ,for dubbo's admin monitoring.
+
+ // Register is used for service provider calling, register services
+ // to registry. And it is also used for service consumer calling, register
+ // services cared about, for dubbo's admin monitoring.
Register(url common.URL) error
// UnRegister is required to support the contract:
- // 1. If it is the persistent stored data of dynamic=false, the registration data can not be found, then the IllegalStateException is thrown, otherwise it is ignored.
+ // 1. If it is the persistent stored data of dynamic=false, the
+ // registration data can not be found, then the IllegalStateException
+ // is thrown, otherwise it is ignored.
// 2. Unregister according to the full url match.
- // url Registration information , is not allowed to be empty, e.g: dubbo://10.20.153.10/org.apache.dubbo.foo.BarService?version=1.0.0&application=kylin
+ // url Registration information, is not allowed to be empty, e.g:
+ // dubbo://10.20.153.10/org.apache.dubbo.foo.BarService?version=1.0.0&application=kylin
UnRegister(url common.URL) error
- //When creating new registry extension,pls select one of the following modes.
- //Will remove in dubbogo version v1.1.0
- //mode1 : return Listener with Next function which can return subscribe service event from registry
- //Deprecated!
- //subscribe(event.URL) (Listener, error)
-
- //Will replace mode1 in dubbogo version v1.1.0
- //mode2 : callback mode, subscribe with notify(notify listener).
+ // Subscribe is required to support the contract:
+ // When creating new registry extension, pls select one of the
+ // following modes.
+ // Will remove in dubbogo version v1.1.0
+ // mode1: return Listener with Next function which can return
+ // subscribe service event from registry
+ // Deprecated!
+ // subscribe(event.URL) (Listener, error)
+ // Will replace mode1 in dubbogo version v1.1.0
+ // mode2: callback mode, subscribe with notify(notify listener).
Subscribe(*common.URL, NotifyListener) error
// UnSubscribe is required to support the contract:
// 1. If don't subscribe, ignore it directly.
// 2. Unsubscribe by full URL match.
- // url Subscription condition, not allowed to be empty, e.g. consumer://10.20.153.10/org.apache.dubbo.foo.BarService?version=1.0.0&application=kylin
+ // url Subscription condition, not allowed to be empty, e.g.
+ // consumer://10.20.153.10/org.apache.dubbo.foo.BarService?version=1.0.0&application=kylin
// listener A listener of the change event, not allowed to be empty
UnSubscribe(*common.URL, NotifyListener) error
}