diff --git a/common/extension/cluster.go b/common/extension/cluster.go
index 1fd76a4756fbce1a198e6ea410acb052cd8dc537..553209ebadb0ad5d544bc8bd96f4d571364b5de4 100644
--- a/common/extension/cluster.go
+++ b/common/extension/cluster.go
@@ -13,5 +13,8 @@ func SetCluster(name string, fcn func() cluster.Cluster) {
}
func GetCluster(name string) cluster.Cluster {
+ if clusters[name] == nil {
+ panic("cluster for " + name + " is not existing, you must import corresponding package.")
+ }
return clusters[name]()
}
diff --git a/common/extension/filter.go b/common/extension/filter.go
index e602916ecadb2e65df47529e09ba0ffaa59dfc30..2586a28c0780f3c48ebcb9a449ccfde5b60f9e7a 100644
--- a/common/extension/filter.go
+++ b/common/extension/filter.go
@@ -17,5 +17,8 @@ func SetFilter(name string, v func() filter.Filter) {
}
func GetFilterExtension(name string) filter.Filter {
+ if filters[name] == nil {
+ panic("filter for " + name + " is not existing, you must import corresponding package.")
+ }
return filters[name]()
}
diff --git a/common/extension/loadbalance.go b/common/extension/loadbalance.go
index ae1998fb3f404cc5c8d3f43058657c8b9328da8a..97d22cffbc6f54e001cbf0a7fed6f507ce7ab377 100644
--- a/common/extension/loadbalance.go
+++ b/common/extension/loadbalance.go
@@ -11,5 +11,8 @@ func SetLoadbalance(name string, fcn func() cluster.LoadBalance) {
}
func GetLoadbalance(name string) cluster.LoadBalance {
+ if loadbalances[name] == nil {
+ panic("loadbalance for " + name + " is not existing, you must import corresponding package.")
+ }
return loadbalances[name]()
}
diff --git a/common/extension/protocol.go b/common/extension/protocol.go
index 2cc14abf82eca77d8796b8708d2ffa56eba2224f..2ac43676b7ca58e2934241504f2095f0d7f2715a 100644
--- a/common/extension/protocol.go
+++ b/common/extension/protocol.go
@@ -17,5 +17,8 @@ func SetProtocol(name string, v func() protocol.Protocol) {
}
func GetProtocolExtension(name string) protocol.Protocol {
+ if protocols[name] == nil {
+ panic("protocol for " + name + " is not existing, you must import corresponding package.")
+ }
return protocols[name]()
}
diff --git a/common/extension/registry.go b/common/extension/registry.go
index a3238d65816c0f7242918db673b1c4696967375c..9fc97912adf3a8e0970f656a5d5e2f9356823182 100644
--- a/common/extension/registry.go
+++ b/common/extension/registry.go
@@ -22,6 +22,9 @@ func SetRegistry(name string, v func(config *common.URL) (registry.Registry, err
}
func GetRegistryExtension(name string, config *common.URL) (registry.Registry, error) {
+ if registrys[name] == nil {
+ panic("registry for " + name + " is not existing, you must import corresponding package.")
+ }
return registrys[name](config)
}
diff --git a/common/rpc_service.go b/common/rpc_service.go
index 8b1243794e480d2ff9323feb54d715f95925a1ef..e94bb7b891c9d25329709e522cc00e0c1fcd2c86 100644
--- a/common/rpc_service.go
+++ b/common/rpc_service.go
@@ -2,6 +2,7 @@ package common
import (
"context"
+ "errors"
"reflect"
"strings"
"sync"
@@ -25,6 +26,7 @@ var (
// because Typeof takes an empty interface value. This is annoying.
typeOfError = reflect.TypeOf((*error)(nil)).Elem()
+ // todo: lowerecas?
ServiceMap = &serviceMap{
serviceMap: make(map[string]map[string]*Service),
}
@@ -145,6 +147,31 @@ func (sm *serviceMap) Register(protocol string, rcvr RPCService) (string, error)
return strings.TrimSuffix(methods, ","), nil
}
+func (sm *serviceMap) UnRegister(protocol, serviceName string) error {
+ if protocol == "" || serviceName == "" {
+ return errors.New("protocol or serviceName is nil")
+ }
+ sm.mutex.RLock()
+ svcs, ok := sm.serviceMap[protocol]
+ if !ok {
+ sm.mutex.RUnlock()
+ return errors.New("no services for " + protocol)
+ }
+ _, ok = svcs[serviceName]
+ if !ok {
+ sm.mutex.RUnlock()
+ return errors.New("no service for " + serviceName)
+ }
+ sm.mutex.RUnlock()
+
+ sm.mutex.Lock()
+ defer sm.mutex.Unlock()
+ delete(svcs, serviceName)
+ delete(sm.serviceMap, protocol)
+
+ return nil
+}
+
// Is this an exported - upper case - name
func isExported(name string) bool {
rune, _ := utf8.DecodeRuneInString(name)
diff --git a/common/rpc_service_test.go b/common/rpc_service_test.go
index 6d72bf5e56f4eefe1404f92555dcb7fb91265d9f..7b38236f2960a0681d726a7aba181b331ba21ca8 100644
--- a/common/rpc_service_test.go
+++ b/common/rpc_service_test.go
@@ -81,6 +81,30 @@ func TestServiceMap_Register(t *testing.T) {
s1 := &TestService1{}
methods, err = ServiceMap.Register("testporotocol", s1)
assert.EqualError(t, err, "type com.test.Path1 has no exported methods of suitable type")
+
+ ServiceMap = &serviceMap{
+ serviceMap: make(map[string]map[string]*Service),
+ }
+}
+
+func TestServiceMap_UnRegister(t *testing.T) {
+ s := &TestService{}
+ _, err := ServiceMap.Register("testprotocol", s)
+ assert.NoError(t, err)
+ assert.NotNil(t, ServiceMap.GetService("testprotocol", "com.test.Path"))
+
+ err = ServiceMap.UnRegister("", "com.test.Path")
+ assert.EqualError(t, err, "protocol or serviceName is nil")
+
+ err = ServiceMap.UnRegister("protocol", "com.test.Path")
+ assert.EqualError(t, err, "no services for protocol")
+
+ err = ServiceMap.UnRegister("testprotocol", "com.test.Path1")
+ assert.EqualError(t, err, "no service for com.test.Path1")
+
+ // succ
+ err = ServiceMap.UnRegister("testprotocol", "com.test.Path")
+ assert.NoError(t, err)
}
func TestMethodType_SuiteContext(t *testing.T) {
diff --git a/go.mod b/go.mod
index 45cf064bdf992486ca2371775517bd309687bf19..793fe406ad4358f3509994741c33ececfb275611 100644
--- a/go.mod
+++ b/go.mod
@@ -4,10 +4,8 @@ require (
github.com/AlexStocks/getty v0.0.0-20190513203438-4a52b6874223
github.com/AlexStocks/goext v0.3.2
github.com/AlexStocks/log4go v1.0.2
- github.com/dubbogo/hessian2 v0.0.0-20190513195500-efce02944002
+ github.com/dubbogo/hessian2 v0.0.0-20190515104320-57ac2e777dc6
github.com/juju/errors v0.0.0-20190207033735-e65537c515d7
- github.com/montanaflynn/stats v0.5.0
- github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec
github.com/stretchr/testify v1.3.0
github.com/tevino/abool v0.0.0-20170917061928-9b9efcf221b5
diff --git a/go.sum b/go.sum
index 46219102e3c2f445853950af42d65eeca5d3dfed..bc22846cb1e5d7ea6b0d0b6b337d9cbe35694f3e 100644
--- a/go.sum
+++ b/go.sum
@@ -24,8 +24,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
-github.com/dubbogo/hessian2 v0.0.0-20190513195500-efce02944002 h1:qtXH4fHzFh1ezGWIdVrfIYxbaCUqbDpUwu2xIjnkw6I=
-github.com/dubbogo/hessian2 v0.0.0-20190513195500-efce02944002/go.mod h1:v+gfInE8fm/k3Fjkb2oUCKSO9LKbWvf+PtweEI89BmI=
+github.com/dubbogo/hessian2 v0.0.0-20190515104320-57ac2e777dc6 h1:qFg1/KkHq4W/c5/3sagBNuMT7VjTiMQGIqknIlnPhu0=
+github.com/dubbogo/hessian2 v0.0.0-20190515104320-57ac2e777dc6/go.mod h1:v+gfInE8fm/k3Fjkb2oUCKSO9LKbWvf+PtweEI89BmI=
github.com/fatih/set v0.2.1/go.mod h1:+RKtMCH+favT2+3YecHGxcc0b4KyVWA1QWWJUs4E0CI=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
@@ -82,8 +82,6 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0j
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/montanaflynn/stats v0.5.0 h1:2EkzeTSqBB4V4bJwWrt5gIIrZmpJBcoIRGS2kWLgzmk=
-github.com/montanaflynn/stats v0.5.0/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd h1:22rhYEzttbrnKjgYh5pifnDluXHHcJ3uSOi2l8Nw+9A=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd/go.mod h1:JrOIxq3vDxvtuEI7Kmm2yqkuBfuT9DMLFMnCyYHLaKM=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd/go.mod h1:JrOIxq3vDxvtuEI7Kmm2yqkuBfuT9DMLFMnCyYHLaKM=
diff --git a/protocol/dubbo/dubbo_exporter.go b/protocol/dubbo/dubbo_exporter.go
index 9a4ed023f05c221ea731e6816ef771cedb494e11..3e73211cb11e6f206be11413c50b9343a89ca80c 100644
--- a/protocol/dubbo/dubbo_exporter.go
+++ b/protocol/dubbo/dubbo_exporter.go
@@ -5,6 +5,12 @@ import (
)
import (
+ log "github.com/AlexStocks/log4go"
+)
+
+import (
+ "github.com/dubbo/go-for-apache-dubbo/common"
+ "github.com/dubbo/go-for-apache-dubbo/common/constant"
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
@@ -17,3 +23,12 @@ func NewDubboExporter(key string, invoker protocol.Invoker, exporterMap *sync.Ma
BaseExporter: *protocol.NewBaseExporter(key, invoker, exporterMap),
}
}
+
+func (de *DubboExporter) Unexport() {
+ service := de.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
+ de.BaseExporter.Unexport()
+ err := common.ServiceMap.UnRegister(DUBBO, service)
+ if err != nil {
+ log.Error("[DubboExporter.Unexport] error: %v", err)
+ }
+}
diff --git a/protocol/jsonrpc/http_test.go b/protocol/jsonrpc/http_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..8c681e8d06ce978ead14624604b1d5916ae4f20d
--- /dev/null
+++ b/protocol/jsonrpc/http_test.go
@@ -0,0 +1,99 @@
+package jsonrpc
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "github.com/dubbo/go-for-apache-dubbo/common"
+ "github.com/dubbo/go-for-apache-dubbo/common/constant"
+ "github.com/dubbo/go-for-apache-dubbo/protocol"
+)
+
+type (
+ User struct {
+ Id string `json:"id"`
+ Name string `json:"name"`
+ }
+
+ UserProvider struct {
+ user map[string]User
+ }
+)
+
+func TestHTTPClient_Call(t *testing.T) {
+
+ methods, err := common.ServiceMap.Register("jsonrpc", &UserProvider{})
+ assert.NoError(t, err)
+ assert.Equal(t, "GetUser,GetUser1", methods)
+
+ // Export
+ proto := GetProtocol()
+ url, err := common.NewURL(context.Background(), "jsonrpc://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
+ "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
+ "side=provider&timeout=3000×tamp=1556509797245")
+ assert.NoError(t, err)
+ proto.Export(protocol.NewBaseInvoker(url))
+
+ client := NewHTTPClient(&HTTPOptions{
+ HandshakeTimeout: time.Second,
+ HTTPTimeout: time.Second,
+ })
+
+ // call GetUser
+ ctx := context.WithValue(context.Background(), constant.DUBBOGO_CTX_KEY, map[string]string{
+ "X-Proxy-Id": "dubbogo",
+ "X-Services": url.Path,
+ "X-Method": "GetUser",
+ })
+ req := client.NewRequest(url, "GetUser", []interface{}{"1", "username"})
+ reply := &User{}
+ err = client.Call(ctx, url, req, reply)
+ assert.NoError(t, err)
+ assert.Equal(t, "1", reply.Id)
+ assert.Equal(t, "username", reply.Name)
+
+ // call GetUser1
+ ctx = context.WithValue(context.Background(), constant.DUBBOGO_CTX_KEY, map[string]string{
+ "X-Proxy-Id": "dubbogo",
+ "X-Services": url.Path,
+ "X-Method": "GetUser1",
+ })
+ req = client.NewRequest(url, "GetUser1", []interface{}{""})
+ reply = &User{}
+ err = client.Call(ctx, url, req, reply)
+ assert.True(t, strings.Contains(err.Error(), "500 Internal Server Error"))
+ assert.True(t, strings.Contains(err.Error(), "\\\"result\\\":{},\\\"error\\\":{\\\"code\\\":-32000,\\\"message\\\":\\\"error\\\"}"))
+
+ // destroy
+ proto.Destroy()
+
+}
+
+func (u *UserProvider) GetUser(ctx context.Context, req []interface{}, rsp *User) error {
+ rsp.Id = req[0].(string)
+ rsp.Name = req[1].(string)
+ return nil
+}
+
+func (u *UserProvider) GetUser1(ctx context.Context, req []interface{}, rsp *User) error {
+ return errors.New("error")
+}
+
+func (u *UserProvider) Service() string {
+ return "com.ikurento.user.UserProvider"
+}
+
+func (u *UserProvider) Version() string {
+ return ""
+}
diff --git a/protocol/jsonrpc/json_test.go b/protocol/jsonrpc/json_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..b14afe7778a626fdeeeb71916ff8a51613a38258
--- /dev/null
+++ b/protocol/jsonrpc/json_test.go
@@ -0,0 +1,70 @@
+package jsonrpc
+
+import (
+ "encoding/json"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+type TestData struct {
+ Test string
+}
+
+func TestJsonClientCodec_Write(t *testing.T) {
+ cd := &CodecData{
+ ID: 1,
+ Method: "GetUser",
+ Args: []interface{}{"args", 2},
+ }
+ codec := newJsonClientCodec()
+ data, err := codec.Write(cd)
+ assert.NoError(t, err)
+ assert.Equal(t, "{\"jsonrpc\":\"2.0\",\"method\":\"GetUser\",\"params\":[\"args\",2],\"id\":1}\n", string(data))
+}
+
+func TestJsonClientCodec_Read(t *testing.T) {
+ codec := newJsonClientCodec()
+ codec.pending[1] = "GetUser"
+ rsp := &TestData{}
+ err := codec.Read([]byte("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"Test\":\"test\"}}\n"), rsp)
+ assert.NoError(t, err)
+ assert.Equal(t, "test", rsp.Test)
+
+ //error
+ codec.pending[1] = "GetUser"
+ err = codec.Read([]byte("{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\":-32000,\"message\":\"error\"}}\n"), rsp)
+ assert.EqualError(t, err, "{\"code\":-32000,\"message\":\"error\"}")
+}
+
+func TestServerCodec_Write(t *testing.T) {
+ codec := newServerCodec()
+ a := json.RawMessage([]byte("1"))
+ codec.req = serverRequest{Version: "1.0", Method: "GetUser", ID: &a}
+ data, err := codec.Write("error", &TestData{Test: "test"})
+ assert.NoError(t, err)
+ assert.Equal(t, "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"Test\":\"test\"},\"error\":{\"code\":-32000,\"message\":\"error\"}}\n", string(data))
+}
+
+func TestServerCodec_Read(t *testing.T) {
+ codec := newServerCodec()
+ header := map[string]string{}
+ err := codec.ReadHeader(header, []byte("{\"jsonrpc\":\"2.0\",\"method\":\"GetUser\",\"params\":[\"args\",2],\"id\":1}\n"))
+ assert.EqualError(t, err, "{\"code\":-32601,\"message\":\"Method not found\"}")
+
+ header["HttpMethod"] = "POST"
+ err = codec.ReadHeader(header, []byte("{\"jsonrpc\":\"2.0\",\"method\":\"GetUser\",\"params\":[\"args\",2],\"id\":1}\n"))
+ assert.NoError(t, err)
+ assert.Equal(t, "1", string([]byte(*codec.req.ID)))
+ assert.Equal(t, "GetUser", codec.req.Method)
+ assert.Equal(t, "2.0", codec.req.Version)
+ assert.Equal(t, "[\"args\",2]", string([]byte(*codec.req.Params)))
+
+ req := []interface{}{}
+ err = codec.ReadBody(&req)
+ assert.NoError(t, err)
+ assert.Equal(t, "args", req[0])
+ assert.Equal(t, float64(2), req[1])
+}
diff --git a/protocol/jsonrpc/jsonrpc_exporter.go b/protocol/jsonrpc/jsonrpc_exporter.go
index 3401d3fcb8774f4cc60c8b9148369a5b756afd32..eedb4044d570087269927aeeb157c4c547b1e808 100644
--- a/protocol/jsonrpc/jsonrpc_exporter.go
+++ b/protocol/jsonrpc/jsonrpc_exporter.go
@@ -5,6 +5,12 @@ import (
)
import (
+ log "github.com/AlexStocks/log4go"
+)
+
+import (
+ "github.com/dubbo/go-for-apache-dubbo/common"
+ "github.com/dubbo/go-for-apache-dubbo/common/constant"
"github.com/dubbo/go-for-apache-dubbo/protocol"
)
@@ -17,3 +23,12 @@ func NewJsonrpcExporter(key string, invoker protocol.Invoker, exporterMap *sync.
BaseExporter: *protocol.NewBaseExporter(key, invoker, exporterMap),
}
}
+
+func (je *JsonrpcExporter) Unexport() {
+ service := je.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
+ je.BaseExporter.Unexport()
+ err := common.ServiceMap.UnRegister(JSONRPC, service)
+ if err != nil {
+ log.Error("[JsonrpcExporter.Unexport] error: %v", err)
+ }
+}
diff --git a/protocol/jsonrpc/jsonrpc_protocol_test.go b/protocol/jsonrpc/jsonrpc_protocol_test.go
index e8609da1ab8e94f35ba44abc85b44d1645c074e5..f67a5d8ed93decb336b84f956a0bcbc9a4228a90 100644
--- a/protocol/jsonrpc/jsonrpc_protocol_test.go
+++ b/protocol/jsonrpc/jsonrpc_protocol_test.go
@@ -19,7 +19,7 @@ import (
func TestJsonrpcProtocol_Export(t *testing.T) {
// Export
proto := GetProtocol()
- url, err := common.NewURL(context.Background(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+ url, err := common.NewURL(context.Background(), "jsonrpc://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
@@ -49,7 +49,7 @@ func TestJsonrpcProtocol_Export(t *testing.T) {
func TestJsonrpcProtocol_Refer(t *testing.T) {
// Refer
proto := GetProtocol()
- url, err := common.NewURL(context.Background(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+ url, err := common.NewURL(context.Background(), "jsonrpc://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
"application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
"environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
"module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
diff --git a/registry/zookeeper/zk_client.go b/registry/zookeeper/zk_client.go
index cb417729e30154b6ba2aaadd33c313d0683d5282..d8d351fbdadf61ed80a363567224cb032271203f 100644
--- a/registry/zookeeper/zk_client.go
+++ b/registry/zookeeper/zk_client.go
@@ -220,7 +220,6 @@ func (z *zookeeperClient) unregisterEvent(zkPath string, event *chan struct{}) {
} else {
z.eventRegistry[zkPath] = a
}
- break
}
z.Unlock()
}