Merge branch 'develop' of git.ailuowan.com:zzw/lone-services into develop
This commit is contained in:
+33
-3
@@ -8,11 +8,17 @@ import (
|
||||
"lone-services/bff/internal/dialer"
|
||||
"lone-services/bff/internal/request"
|
||||
"lone-services/bff/internal/response"
|
||||
"lone-services/bff/internal/wsClient"
|
||||
"lone-services/bff/internal/wsproxy"
|
||||
"lone-services/pkg/discovery"
|
||||
"lone-services/pkg/utils"
|
||||
"net/http"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/conf"
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
"github.com/zeromicro/go-zero/gateway"
|
||||
"github.com/zeromicro/go-zero/rest"
|
||||
"github.com/zeromicro/go-zero/zrpc"
|
||||
)
|
||||
|
||||
var configFile = flag.String("f", "etc/bff.yaml", "the config file")
|
||||
@@ -54,16 +60,40 @@ func main() {
|
||||
}
|
||||
}()
|
||||
|
||||
// Gateway 仅处理普通接口 /api /admin，端口 c.Port=10000
|
||||
gw := gateway.MustNewServer(c.GatewayConf,
|
||||
gateway.WithDialer(dialer.Nacos),
|
||||
|
||||
gateway.WithMiddleware(request.FormUploadMiddleware),
|
||||
// 权鉴+穿透
|
||||
gateway.WithMiddleware(request.UserReadMiddleware),
|
||||
gateway.WithMiddleware(response.Wrap),
|
||||
)
|
||||
defer gw.Stop()
|
||||
|
||||
fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
|
||||
rpcMgr := wsproxy.NewRpcClientManager(c.WsPercutaneousRpc,
|
||||
zrpc.WithUnaryClientInterceptor(request.UserClientInterceptor),
|
||||
zrpc.WithStreamClientInterceptor(request.UserStreamInterceptor),
|
||||
)
|
||||
|
||||
wsRest := rest.MustNewServer(rest.RestConf{
|
||||
Host: c.Host,
|
||||
Port: c.WsPort,
|
||||
})
|
||||
defer wsRest.Stop()
|
||||
|
||||
// 注册ws路由到独立rest服务
|
||||
wsRest.AddRoutes([]rest.Route{
|
||||
{
|
||||
Method: http.MethodGet,
|
||||
Path: "/ws/percutaneous/add/user",
|
||||
Handler: wsClient.PercutaneousAddUser(rpcMgr),
|
||||
},
|
||||
})
|
||||
|
||||
go func() {
|
||||
logx.Info("WS rest server start on :", c.WsPort)
|
||||
wsRest.Start()
|
||||
}()
|
||||
|
||||
fmt.Printf("Gateway server at %s:%d\n", c.Host, c.Port)
|
||||
gw.Start()
|
||||
}
|
||||
|
||||
@@ -2,11 +2,14 @@ package config
|
||||
|
||||
import (
|
||||
"github.com/zeromicro/go-zero/gateway"
|
||||
"github.com/zeromicro/go-zero/zrpc"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
gateway.GatewayConf
|
||||
Nacos NacosConf
|
||||
Nacos NacosConf
|
||||
WsPercutaneousRpc zrpc.RpcClientConf `json:",optional"`
|
||||
WsPort int
|
||||
}
|
||||
|
||||
type NacosConf struct {
|
||||
|
||||
@@ -205,3 +205,40 @@ func UserClientInterceptor(ctx context.Context, method string, req, reply any, c
|
||||
ctx = metadata.NewOutgoingContext(ctx, md)
|
||||
return invoker(ctx, method, req, reply, cc, opts...)
|
||||
}
|
||||
|
||||
// UserStreamInterceptor gRPC Stream客户端拦截器，用于ws桥接的双向流透传metadata
|
||||
func UserStreamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
|
||||
user := GetUserInfo(ctx)
|
||||
md := metadata.New(map[string]string{})
|
||||
|
||||
if user.Valid {
|
||||
md.Set("x-user-id", user.RawUID)
|
||||
md.Set("x-user-name", url.QueryEscape(user.Name))
|
||||
}
|
||||
|
||||
setMDIfNotEmpty(md, "x-raw-body", user.RawBody)
|
||||
setMDIfNotEmpty(md, "x-refresh", user.Refresh)
|
||||
setMDIfNotEmpty(md, "x-client-ip", user.ClientIP)
|
||||
if user.UserAgent != utils.StringEmpty {
|
||||
md.Set("x-user-agent", url.QueryEscape(user.UserAgent))
|
||||
}
|
||||
setMDIfNotEmpty(md, "x-user-type", user.Type)
|
||||
setMDIfNotEmpty(md, "x-client-code", user.ClientCode)
|
||||
setMDIfNotEmpty(md, "x-sale-id", user.SaleId)
|
||||
if user.SaleName != utils.StringEmpty {
|
||||
md.Set("x-sale-name", url.QueryEscape(user.SaleName))
|
||||
}
|
||||
setMDIfNotEmpty(md, "x-sale-mobile", user.SaleMobile)
|
||||
setMDIfNotEmpty(md, "x-sale-province", user.SaleProvince)
|
||||
setMDIfNotEmpty(md, "x-store-id", user.StoreId)
|
||||
if user.StoreName != utils.StringEmpty {
|
||||
md.Set("x-store-name", url.QueryEscape(user.StoreName))
|
||||
}
|
||||
setMDIfNotEmpty(md, "x-group-id", user.GroupId)
|
||||
if user.GroupName != utils.StringEmpty {
|
||||
md.Set("x-group-name", url.QueryEscape(user.GroupName))
|
||||
}
|
||||
|
||||
newCtx := metadata.NewOutgoingContext(ctx, md)
|
||||
return streamer(newCtx, desc, cc, method, opts...)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package wsClient
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"lone-services/bff/internal/request"
|
||||
"lone-services/bff/internal/wsproxy"
|
||||
percutaneous "lone-services/rpc/percutaneous/pb"
|
||||
"net/http"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
)
|
||||
|
||||
func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc {
|
||||
return request.UserReadMiddleware(func(w http.ResponseWriter, r *http.Request) {
|
||||
logx.Infof("BFF ws handler start, raw url=%s", r.URL.String())
|
||||
cli, ready := rpcMgr.GetClient()
|
||||
if !ready {
|
||||
logx.Error("BFF: rpc client not ready, return 503")
|
||||
http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
logx.Info("BFF: rpc client ready")
|
||||
|
||||
hostCode := r.URL.Query().Get("host_code")
|
||||
logx.Infof("BFF: get host_code from query, val=%s", hostCode)
|
||||
if hostCode == "" {
|
||||
logx.Error("BFF: host_code is empty")
|
||||
http.Error(w, "missing host_code query param", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
proxyHandler := wsproxy.NewWsProxyHandler(
|
||||
func(parentCtx context.Context) (grpc.ClientStream, error) {
|
||||
logx.Info("BFF: enter grpc stream create callback")
|
||||
streamCli := percutaneous.NewPercutaneousClient(cli.Conn())
|
||||
stream, err := streamCli.WsAddUserStream(parentCtx)
|
||||
if err != nil {
|
||||
logx.Errorf("BFF: create WsAddUserStream failed, err=%v", err)
|
||||
return nil, err
|
||||
}
|
||||
logx.Info("BFF: grpc stream create SUCCESS")
|
||||
|
||||
// =========核心新增：BFF自动发送第一条host_code消息给percutaneous=========
|
||||
jsonStr := fmt.Sprintf(`{"host_code":"%s"}`, hostCode)
|
||||
reqMsg := &percutaneous.WsStreamRequest{
|
||||
RawBody: []byte(jsonStr),
|
||||
}
|
||||
err = stream.SendMsg(reqMsg)
|
||||
if err != nil {
|
||||
logx.Errorf("BFF: auto send host_code msg failed, err=%v, json=%s", err, jsonStr)
|
||||
return nil, err
|
||||
}
|
||||
logx.Infof("BFF: auto send host_code message success, content=%s", jsonStr)
|
||||
// =====================================================================
|
||||
return stream, nil
|
||||
},
|
||||
func() protoreflect.ProtoMessage {
|
||||
return &percutaneous.WsStreamRequest{}
|
||||
},
|
||||
)
|
||||
logx.Info("BFF: start run proxyHandler")
|
||||
proxyHandler(w, r)
|
||||
logx.Info("BFF: proxyHandler finished, ws stream closed")
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package wsproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
"github.com/zeromicro/go-zero/zrpc"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
)
|
||||
|
||||
type StreamCreator func(ctx context.Context) (grpc.ClientStream, error)
|
||||
|
||||
var upgrader = websocket.Upgrader{
|
||||
ReadBufferSize: 1024,
|
||||
WriteBufferSize: 1024,
|
||||
CheckOrigin: func(r *http.Request) bool {
|
||||
return true
|
||||
},
|
||||
}
|
||||
|
||||
func NewWsProxyHandler(createStream StreamCreator, newFrame func() protoreflect.ProtoMessage) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
conn, err := upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
stream, err := createStream(r.Context())
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer stream.CloseSend()
|
||||
|
||||
// 上行 ws -> grpc stream
|
||||
go func() {
|
||||
for {
|
||||
_, rawBytes, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
msg := newFrame()
|
||||
fd := msg.ProtoReflect().Descriptor().Fields().ByName("raw_body")
|
||||
msg.ProtoReflect().Set(fd, protoreflect.ValueOfBytes(rawBytes))
|
||||
_ = stream.SendMsg(msg)
|
||||
}
|
||||
}()
|
||||
|
||||
// 下行 grpc stream -> ws
|
||||
for {
|
||||
msg := newFrame()
|
||||
err := stream.RecvMsg(msg)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
fd := msg.ProtoReflect().Descriptor().Fields().ByName("raw_body")
|
||||
val := msg.ProtoReflect().Get(fd)
|
||||
_ = conn.WriteMessage(websocket.TextMessage, val.Bytes())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 全局rpc客户端管理器，线程安全
|
||||
type RpcClientManager struct {
|
||||
mu sync.RWMutex
|
||||
client zrpc.Client
|
||||
ready bool
|
||||
cfg zrpc.RpcClientConf
|
||||
streamOpts []zrpc.ClientOption
|
||||
}
|
||||
|
||||
func NewRpcClientManager(cfg zrpc.RpcClientConf, streamOpts ...zrpc.ClientOption) *RpcClientManager {
|
||||
m := &RpcClientManager{
|
||||
cfg: cfg,
|
||||
streamOpts: streamOpts,
|
||||
}
|
||||
go m.loopReconnect()
|
||||
return m
|
||||
}
|
||||
|
||||
// 尝试建立一次连接
|
||||
func (m *RpcClientManager) tryConnect() error {
|
||||
cli, err := zrpc.NewClient(m.cfg, m.streamOpts...)
|
||||
if err != nil {
|
||||
logx.Errorf("tryConnect: create client failed, err=%v", err)
|
||||
return err
|
||||
}
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.client = cli
|
||||
m.ready = true
|
||||
return nil
|
||||
}
|
||||
|
||||
// 后台循环重试
|
||||
func (m *RpcClientManager) loopReconnect() {
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
m.mu.RLock()
|
||||
ready := m.ready
|
||||
m.mu.RUnlock()
|
||||
if ready {
|
||||
logx.Debug("rpc client already ready, skip reconnect")
|
||||
continue
|
||||
}
|
||||
if err := m.tryConnect(); err != nil {
|
||||
logx.Errorf("loopReconnect reconnect failed: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 获取client，判断是否就绪
|
||||
func (m *RpcClientManager) GetClient() (zrpc.Client, bool) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.client, m.ready
|
||||
}
|
||||
Reference in New Issue
Block a user