liwei_3d/go_dreamfactory
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

修复rpcx系统接口

Browse Source
This commit is contained in:
liwei1dao 2022-06-10 20:12:31 +08:00
parent d4d8c074e0
commit fc50b18c2a
4 changed files with 25 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lego/sys/rpcx/rpcx.go
View File

@ -12,7 +12,7 @@ func newSys(options Options) (sys *RPCX, err error) {
options: options,
metadata: fmt.Sprintf("stype=%s&sid=%s&version=%s&addr=%s", options.ServiceType, options.ServiceId, options.ServiceVersion, "tcp@"+options.ServiceAddr),
}
sys.service = newService(sys)
sys.service, err = newService(sys)
sys.client = newClient(sys)
// if options.RpcxStartType == RpcxStartByAll || options.RpcxStartType == RpcxStartByService { //创建RPCX 服务端

4
lego/sys/rpcx/rpcx_test.go
View File

@ -26,7 +26,7 @@ func Test_Sys(t *testing.T) {
return
}
if sys, err := NewSys(
SetServiceTag("rpcx_test"),
SetServiceTag("dreamfactory"),
SetServiceType("worker"),
SetServiceId("worker_1"),
SetServiceVersion("1.0.0"),
@ -72,7 +72,7 @@ func Test_RPCX(t *testing.T) {
}
go func() {
time.Sleep(time.Second)
s.RegisterName("worker", new(Arith), "stype=worker&sid=worker_1&version=1.0.0")
s.RegisterName("worker", new(Arith), "stype=worker&sid=worker_1&version=1.0.0&addr=tcp@127.0.0.1:9978")
}()
go func() {

6
lego/sys/rpcx/selector.go
View File

@ -3,6 +3,7 @@ package rpcx
import (
"context"
"fmt"
"go_dreamfactory/lego/sys/log"
"strings"
"github.com/smallnest/rpcx/share"
@ -31,8 +32,8 @@ type Selector struct {
///servicePath = [stype] or [stype/sid]
func (this *Selector) Select(ctx context.Context, servicePath, serviceMethod string, args interface{}) string {
fmt.Printf("Select servicePath:%v serviceMethod:%v RoutRules:%v \n", servicePath, serviceMethod, ctx.Value("RoutRules"))
routrules := ctx.Value(share.ReqMetaDataKey).(map[string]string)["RoutRules"]
fmt.Printf("Select servicePath:%v serviceMethod:%v ReqMetaData:%v \n", servicePath, serviceMethod, ctx.Value(share.ReqMetaDataKey))
routrules := ctx.Value(share.ReqMetaDataKey).(map[string]string)[CallRoutRulesKey]
service := strings.Split(routrules, "/")
leng := len(service)
if leng == 1 {
@ -58,6 +59,7 @@ func (this *Selector) UpdateServer(servers map[string]string) {
sst := make(map[string][]*ServiceNode)
for _, v := range servers {
if node, err := smetaToServiceNode(v); err != nil {
log.Errorf("smetaToServiceNode:%s err:%v", v, err)
continue
} else {
ss[node.ServiceId] = node

31
lego/sys/rpcx/service.go
View File

@ -2,6 +2,7 @@ package rpcx
import (
"context"
"go_dreamfactory/lego/sys/log"
"net"
"time"
@ -11,13 +12,25 @@ import (
"github.com/smallnest/rpcx/serverplugin"
)
func newService(rpcx *RPCX) (s *Service) {
func newService(rpcx *RPCX) (s *Service, err error) {
s = &Service{
server: server.NewServer(),
rpcx: rpcx,
// clients: make(map[string]net.Conn),
// clientmeta: make(map[string]string),
}
r := &serverplugin.ConsulRegisterPlugin{
ServiceAddress: "tcp@" + rpcx.options.ServiceAddr,
ConsulServers: []string{"10.0.0.9:8500"},
BasePath: rpcx.options.ServiceTag,
Metrics: metrics.NewRegistry(),
UpdateInterval: time.Minute,
}
if err = r.Start(); err != nil {
return
}
s.server.Plugins.Add(r)
return
}
@ -30,19 +43,11 @@ type Service struct {
}
func (this *Service) Start() (err error) {
r := &serverplugin.ConsulRegisterPlugin{
ServiceAddress: "tcp@" + this.rpcx.options.ServiceAddr,
ConsulServers: []string{"10.0.0.9:8500"},
BasePath: this.rpcx.options.ServiceAddr,
Metrics: metrics.NewRegistry(),
UpdateInterval: time.Minute,
}
if err = r.Start(); err != nil {
return
}
this.server.Plugins.Add(r)
go func() {
this.server.Serve("tcp", this.rpcx.options.ServiceAddr)
if err = this.server.Serve("tcp", this.rpcx.options.ServiceAddr); err != nil {
log.Errorf("rpcx server exit!")
}
}()
return
}