feat: 调整代码

application/rpcserver.go

@@ -1,20 +1,13 @@
 package application
 
 import (
-	"context"
 	"fmt"
 	routing "github.com/gly-hub/fasthttp-routing"
 	"github.com/gly-hub/toolbox/ip"
-	"github.com/gly-hub/toolbox/stringx"
-	jsoniter "github.com/json-iterator/go"
 	"github.com/smallnest/rpcx/server"
 	"github.com/team-dandelion/go-dandelion/config"
-	error_support "github.com/team-dandelion/go-dandelion/error-support"
 	"github.com/team-dandelion/go-dandelion/logger"
-	"github.com/team-dandelion/go-dandelion/server/http"
 	"github.com/team-dandelion/go-dandelion/server/rpcx"
-	"github.com/team-dandelion/go-dandelion/telemetry"
-	"reflect"
 )
 
 var (
@@ -25,8 +18,8 @@ var (
 
 type RpcClient struct {
 	ClientName string
-	clientPool *rpcx.ClientPool
-	headerFunc func(ctx *routing.Context, header map[string]string) map[string]string
+	ClientPool *rpcx.ClientPool
+	HeaderFunc func(ctx *routing.Context, header map[string]string) map[string]string
 }
 
 func initRpcClient() {
@@ -47,97 +40,12 @@ func initRpcClient() {
 	}
 	rpcClient = &RpcClient{
 		ClientName: config.Conf.RpcClient.ClientName,
-		clientPool: client,
+		ClientPool: client,
 	}
 }
 
 func RegisterHeaderFunc(f func(ctx *routing.Context, header map[string]string) map[string]string) {
-	rpcClient.headerFunc = f
-}
-
-// RpcCall rpc请求
-func RpcCall(ctx *routing.Context, serverName, funcName string, args interface{}, reply interface{}) error {
-	if rpcClient.clientPool == nil {
-		panic("请配置rpcx参数")
-	}
-	content, _ := jsoniter.MarshalToString(args)
-	var traceId string
-	if telemetry.GetSpanTraceId() != nil {
-		traceId = telemetry.GetSpanTraceId().(string)
-	}
-	requestHeader := map[string]string{
-		"request_id":    stringx.Strval(logger.GetRequestId()),
-		"span_trace_id": traceId,
-		"client_name":   rpcClient.ClientName,
-		"content":       content,
-	}
-
-	requestHeader = rpcClient.headerFunc(ctx, requestHeader)
-	c := rpcx.Header().Set(context.Background(), requestHeader)
-	err := rpcClient.clientPool.Client().Call(c, serverName, funcName, args, reply)
-	if err != nil {
-		logger.Error("ServerName: ", serverName, ", FuncName: ", funcName, ", Err: ", err)
-		return &error_support.Error{Code: 5001, Msg: "服务器异常"}
-	}
-
-	rv := reflect.ValueOf(reply)
-	if rv.Kind() == reflect.Ptr {
-		rv = rv.Elem()
-	}
-	if rv.FieldByName("Code").Int() != int64(0) {
-		return &error_support.Error{Code: int(rv.FieldByName("Code").Int()), Msg: rv.FieldByName("Msg").String()}
-	}
-	return nil
-}
-
-// SRpcCall rpc请求拓展
-func SRpcCall(ctx *routing.Context, serverName, funcName string, args interface{}, reply interface{}) error {
-	if rpcClient.clientPool == nil {
-		panic("请配置rpcx参数")
-	}
-	var hc http.HttpController
-	if err := hc.ReadJson(ctx, args); err != nil {
-		return hc.Fail(ctx, &error_support.Error{Code: 5000, Msg: "数据解析失败"})
-	}
-
-	content, _ := jsoniter.MarshalToString(args)
-	var traceId string
-	if telemetry.GetSpanTraceId() != nil {
-		traceId = telemetry.GetSpanTraceId().(string)
-	}
-	requestHeader := map[string]string{
-		"request_id":    stringx.Strval(logger.GetRequestId()),
-		"span_trace_id": traceId,
-		"client_name":   rpcClient.ClientName,
-		"content":       content,
-	}
-	requestHeader = rpcClient.headerFunc(ctx, requestHeader)
-	c := rpcx.Header().Set(context.Background(), requestHeader)
-	err := rpcClient.clientPool.Client().Call(c, serverName, funcName, args, reply)
-	if err != nil {
-		logger.Error("ServerName: ", serverName, ", FuncName: ", funcName, ", Err: ", err)
-		return hc.Fail(ctx, &error_support.Error{Code: 5001, Msg: "服务器异常"})
-	}
-
-	rt := reflect.TypeOf(reply)
-	if rt.Kind() == reflect.Ptr {
-		rt = rt.Elem()
-	}
-	_, cOk := rt.FieldByName("Code")
-	_, mOk := rt.FieldByName("Msg")
-	if mOk && cOk {
-		rv := reflect.ValueOf(reply)
-		if rv.Kind() == reflect.Ptr {
-			rv = rv.Elem()
-		}
-
-		if rv.FieldByName("Code").Int() != int64(0) {
-			return hc.Fail(ctx, &error_support.Error{Code: int(rv.FieldByName("Code").Int()), Msg: rv.FieldByName("Msg").String()})
-		}
-		return hc.Success(ctx, reply, rv.FieldByName("Msg").String())
-	}
-
-	return hc.Success(ctx, reply, "")
+	rpcClient.HeaderFunc = f
 }
 
 func RegisterRpcPlugin(plugins ...server.Plugin) {
@@ -174,3 +82,7 @@ func RpcServer(handler interface{}, auth ...rpcx.AuthFunc) {
 	}
 	rpcServer.Start()
 }
+
+func GetRpcClient() *RpcClient {
+	return rpcClient
+}

go.mod

@@ -156,5 +156,6 @@ require (
 	google.golang.org/grpc v1.53.0 // indirect
 	google.golang.org/protobuf v1.29.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1199,6 +1199,8 @@ gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMy
 gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=
 gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
 gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
+gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 h1:VpOs+IwYnYBaFnrNAeB8UUWtL3vEUnzSCL1nVjPhqrw=
+gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/warnings.v0 v0.1.2/go.mod h1:jksf8JmL6Qr/oQM2OXTHunEvvTAsrWBLb6OOjuVWRNI=

tools/rpccall/Makefile

@@ -0,0 +1,4 @@
+proto:
+	protoc -I. -I${GOPATH}/src -I ../../ -I ./ \
+	  --gofast_out=. --gofast_opt=paths=source_relative \
+	  --rpcx_out=. --rpcx_opt=paths=source_relative *.proto