ws
CI / changes (push) Successful in 1m29s
CI / ad (push) Successful in 0s
CI / admin (push) Successful in 0s
CI / bff (push) Successful in 35s
CI / chore (push) Successful in 1s
CI / equipment (push) Successful in 0s
CI / express (push) Successful in 0s
CI / factory (push) Successful in 1s
CI / message (push) Successful in 0s
CI / order (push) Successful in 0s
CI / payment (push) Successful in 1s
CI / percutaneous (push) Successful in 21s
CI / product (push) Successful in 0s
CI / record (push) Successful in 0s
CI / sale (push) Successful in 0s
CI / support (push) Successful in 0s
CI / user (push) Successful in 0s
CI / task (push) Successful in 0s
CI / wecom (push) Successful in 0s
CI / changes (push) Successful in 1m29s
CI / ad (push) Successful in 0s
CI / admin (push) Successful in 0s
CI / bff (push) Successful in 35s
CI / chore (push) Successful in 1s
CI / equipment (push) Successful in 0s
CI / express (push) Successful in 0s
CI / factory (push) Successful in 1s
CI / message (push) Successful in 0s
CI / order (push) Successful in 0s
CI / payment (push) Successful in 1s
CI / percutaneous (push) Successful in 21s
CI / product (push) Successful in 0s
CI / record (push) Successful in 0s
CI / sale (push) Successful in 0s
CI / support (push) Successful in 0s
CI / user (push) Successful in 0s
CI / task (push) Successful in 0s
CI / wecom (push) Successful in 0s
This commit is contained in:
@@ -15,17 +15,14 @@ import (
|
|||||||
|
|
||||||
func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc {
|
func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc {
|
||||||
return request.UserReadMiddleware(func(w http.ResponseWriter, r *http.Request) {
|
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()
|
cli, ready := rpcMgr.GetClient()
|
||||||
if !ready {
|
if !ready {
|
||||||
logx.Error("BFF: rpc client not ready, return 503")
|
logx.Error("BFF: rpc client not ready, return 503")
|
||||||
http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable)
|
http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
logx.Info("BFF: rpc client ready")
|
|
||||||
|
|
||||||
hostCode := r.URL.Query().Get("host_code")
|
hostCode := r.URL.Query().Get("host_code")
|
||||||
logx.Infof("BFF: get host_code from query, val=%s", hostCode)
|
|
||||||
if hostCode == "" {
|
if hostCode == "" {
|
||||||
logx.Error("BFF: host_code is empty")
|
logx.Error("BFF: host_code is empty")
|
||||||
http.Error(w, "missing host_code query param", http.StatusBadRequest)
|
http.Error(w, "missing host_code query param", http.StatusBadRequest)
|
||||||
@@ -34,16 +31,13 @@ func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc {
|
|||||||
|
|
||||||
proxyHandler := wsproxy.NewWsProxyHandler(
|
proxyHandler := wsproxy.NewWsProxyHandler(
|
||||||
func(parentCtx context.Context) (grpc.ClientStream, error) {
|
func(parentCtx context.Context) (grpc.ClientStream, error) {
|
||||||
logx.Info("BFF: enter grpc stream create callback")
|
|
||||||
streamCli := percutaneous.NewPercutaneousClient(cli.Conn())
|
streamCli := percutaneous.NewPercutaneousClient(cli.Conn())
|
||||||
stream, err := streamCli.WsAddUserStream(parentCtx)
|
stream, err := streamCli.WsAddUserStream(parentCtx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logx.Errorf("BFF: create WsAddUserStream failed, err=%v", err)
|
logx.Errorf("BFF: create WsAddUserStream failed, err=%v", err)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
logx.Info("BFF: grpc stream create SUCCESS")
|
|
||||||
|
|
||||||
// =========核心新增:BFF自动发送第一条host_code消息给percutaneous=========
|
|
||||||
jsonStr := fmt.Sprintf(`{"host_code":"%s"}`, hostCode)
|
jsonStr := fmt.Sprintf(`{"host_code":"%s"}`, hostCode)
|
||||||
reqMsg := &percutaneous.WsStreamRequest{
|
reqMsg := &percutaneous.WsStreamRequest{
|
||||||
RawBody: []byte(jsonStr),
|
RawBody: []byte(jsonStr),
|
||||||
@@ -53,16 +47,12 @@ func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc {
|
|||||||
logx.Errorf("BFF: auto send host_code msg failed, err=%v, json=%s", err, jsonStr)
|
logx.Errorf("BFF: auto send host_code msg failed, err=%v, json=%s", err, jsonStr)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
logx.Infof("BFF: auto send host_code message success, content=%s", jsonStr)
|
|
||||||
// =====================================================================
|
|
||||||
return stream, nil
|
return stream, nil
|
||||||
},
|
},
|
||||||
func() protoreflect.ProtoMessage {
|
func() protoreflect.ProtoMessage {
|
||||||
return &percutaneous.WsStreamRequest{}
|
return &percutaneous.WsStreamRequest{}
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
logx.Info("BFF: start run proxyHandler")
|
|
||||||
proxyHandler(w, r)
|
proxyHandler(w, r)
|
||||||
logx.Info("BFF: proxyHandler finished, ws stream closed")
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -48,7 +48,6 @@ func NewWsAddUserStreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *W
|
|||||||
|
|
||||||
// 添加用户
|
// 添加用户
|
||||||
func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error {
|
func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error {
|
||||||
l.Logger.Info("in coming ws tunnel stream")
|
|
||||||
ctx := stream.Context()
|
ctx := stream.Context()
|
||||||
timeout := time.Duration(utils.GetConfigInt("ws.out_time")) * time.Second
|
timeout := time.Duration(utils.GetConfigInt("ws.out_time")) * time.Second
|
||||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||||
@@ -65,12 +64,10 @@ func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_
|
|||||||
var hostCode struct {
|
var hostCode struct {
|
||||||
Code string `json:"host_code"`
|
Code string `json:"host_code"`
|
||||||
}
|
}
|
||||||
l.Logger.Errorf(" host_code raw msg is: %v", string(recvFrame.RawBody))
|
|
||||||
// 保留你确认的 !err 判断
|
// 保留你确认的 !err 判断
|
||||||
if ok := utils.JsonStringToStruct(string(recvFrame.RawBody), &hostCode); !ok {
|
if ok := utils.JsonStringToStruct(string(recvFrame.RawBody), &hostCode); !ok {
|
||||||
return status.Errorf(codes.InvalidArgument, "device code parse error")
|
return status.Errorf(codes.InvalidArgument, "device code parse error")
|
||||||
}
|
}
|
||||||
l.Logger.Infof("device connected, code=%s", hostCode.Code)
|
|
||||||
|
|
||||||
msgCh := make(chan []byte, 1)
|
msgCh := make(chan []byte, 1)
|
||||||
wrapper := &grpcStreamWrapper{stream: stream}
|
wrapper := &grpcStreamWrapper{stream: stream}
|
||||||
@@ -82,7 +79,6 @@ func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_
|
|||||||
}
|
}
|
||||||
l.svcCtx.SessionMgr.Register(hostCode.Code, sess)
|
l.svcCtx.SessionMgr.Register(hostCode.Code, sess)
|
||||||
defer l.svcCtx.SessionMgr.Remove(hostCode.Code)
|
defer l.svcCtx.SessionMgr.Remove(hostCode.Code)
|
||||||
l.Logger.Info("session register success, waiting for scan event")
|
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
|
|||||||
Reference in New Issue
Block a user