From 0d8cfc695c89a6a151b2e139a5e1786e2ee00c35 Mon Sep 17 00:00:00 2001 From: gjs Date: Thu, 8 Oct 2026 16:37:33 +0800 Subject: [PATCH] ws --- .gitea/workflows/cd-test.yml | 1 + .gitea/workflows/ci.yml | 3 +- bff/bff.go | 10 - bff/internal/wsClient/client.go | 90 ++++----- pkg/utils/ws.go | 65 ++++++ rpc/percutaneous/pb/percutaneous.pb.go | 41 ++-- rpc/percutaneous/pb/percutaneous_grpc.pb.go | 188 +----------------- rpc/percutaneous/percutaneous.pb | Bin 15363 -> 14865 bytes services/percutaneous/Dockerfile | 37 ++++ .../internal/logic/addUserLogic.go | 13 +- .../internal/logic/wsAddUserStreamLogic.go | 132 ++++++++---- .../internal/server/percutaneousServer.go | 29 --- .../internal/svc/servicecontext.go | 18 +- .../percutaneousclient/percutaneous.go | 38 ---- 14 files changed, 272 insertions(+), 393 deletions(-) create mode 100644 pkg/utils/ws.go create mode 100644 services/percutaneous/Dockerfile diff --git a/.gitea/workflows/cd-test.yml b/.gitea/workflows/cd-test.yml index 9145a65..bdf6afa 100644 --- a/.gitea/workflows/cd-test.yml +++ b/.gitea/workflows/cd-test.yml @@ -49,4 +49,5 @@ jobs: docker pull ${{ vars.REGISTRY }}/lone/factory:latest docker pull ${{ vars.REGISTRY }}/lone/message:latest docker pull ${{ vars.REGISTRY }}/lone/support:latest + docker pull ${{ vars.REGISTRY }}/lone/percutaneous:latest docker compose up -d diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index e532c76..b4d881e 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -26,7 +26,7 @@ jobs: - name: Detect changes id: filter run: | - SERVICES="bff product admin user ad chore order express sale wecom task equipment record payment factory message support" + SERVICES="bff product admin user ad chore order express sale wecom task equipment record payment factory message support percutaneous" pattern_for() { case "$1" in @@ -98,6 +98,7 @@ jobs: - message - order - support + - percutaneous env: BUILD_LIST: ${{ needs.changes.outputs.build_list }} NAME: lone/${{ matrix.name }} diff --git a/bff/bff.go b/bff/bff.go index 9c110c7..2a6364a 100644 --- a/bff/bff.go +++ b/bff/bff.go @@ -82,16 +82,6 @@ func main() { // 注册ws路由到独立rest服务 wsRest.AddRoutes([]rest.Route{ - { - Method: http.MethodGet, - Path: "/ws/percutaneous/login", - Handler: wsClient.PercutaneousLogin(rpcMgr), - }, - { - Method: http.MethodGet, - Path: "/ws/percutaneous/used", - Handler: wsClient.PercutaneousUsed(rpcMgr), - }, { Method: http.MethodGet, Path: "/ws/percutaneous/add/user", diff --git a/bff/internal/wsClient/client.go b/bff/internal/wsClient/client.go index 075e4f0..d829025 100644 --- a/bff/internal/wsClient/client.go +++ b/bff/internal/wsClient/client.go @@ -2,6 +2,7 @@ package wsClient import ( "context" + "fmt" "lone-services/bff/internal/request" "lone-services/bff/internal/wsproxy" percutaneous "lone-services/rpc/percutaneous/pb" @@ -12,69 +13,56 @@ import ( "google.golang.org/protobuf/reflect/protoreflect" ) -// ws handler -func PercutaneousLogin(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc { - wsHandler := request.UserReadMiddleware(func(w http.ResponseWriter, r *http.Request) { - cli, ready := rpcMgr.GetClient() - if !ready { - logx.Error("ws request: rpc client not ready, return 503") - http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable) - return - } - - proxyHandler := wsproxy.NewWsProxyHandler( - func(ctx context.Context) (grpc.ClientStream, error) { - streamCli := percutaneous.NewPercutaneousClient(cli.Conn()) - return streamCli.WsLoginStream(ctx) - }, - func() protoreflect.ProtoMessage { return &percutaneous.WsStreamRequest{} }, - ) - proxyHandler(w, r) - }) - - return wsHandler -} - func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc { - wsHandler := 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() if !ready { - logx.Error("ws request: rpc client not ready, return 503") + 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(ctx 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()) - return streamCli.WsAddUserStream(ctx) + 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{} }, - ) - proxyHandler(w, r) - }) - - return wsHandler -} - -func PercutaneousUsed(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc { - wsHandler := request.UserReadMiddleware(func(w http.ResponseWriter, r *http.Request) { - cli, ready := rpcMgr.GetClient() - if !ready { - logx.Error("ws request: rpc client not ready, return 503") - http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable) - return - } - - proxyHandler := wsproxy.NewWsProxyHandler( - func(ctx context.Context) (grpc.ClientStream, error) { - streamCli := percutaneous.NewPercutaneousClient(cli.Conn()) - return streamCli.WsUsedStream(ctx) + func() protoreflect.ProtoMessage { + return &percutaneous.WsStreamRequest{} }, - func() protoreflect.ProtoMessage { return &percutaneous.WsStreamRequest{} }, ) + logx.Info("BFF: start run proxyHandler") proxyHandler(w, r) + logx.Info("BFF: proxyHandler finished, ws stream closed") }) - - return wsHandler } diff --git a/pkg/utils/ws.go b/pkg/utils/ws.go new file mode 100644 index 0000000..864842f --- /dev/null +++ b/pkg/utils/ws.go @@ -0,0 +1,65 @@ +package utils + +import ( + "context" + "sync" + + "github.com/zeromicro/go-zero/core/logx" +) + +// Stream 通用双向流抽象接口 +type Stream interface { + Send(payload []byte) error +} + +// WaitingSession 通用等待会话 +type WaitingSession struct { + Stream Stream + Ctx context.Context + Cancel context.CancelFunc + MsgCh chan []byte +} + +// WsSessionManager 会话管理器 +type WsSessionManager struct { + mu sync.RWMutex + sessions map[string]*WaitingSession // key: deviceCode +} + +// 全局单例，logic直接用 WsSessionMgr +var WsSessionMgr = NewWsSessionManager() + +func NewWsSessionManager() *WsSessionManager { + return &WsSessionManager{ + sessions: make(map[string]*WaitingSession), + } +} + +// Register 注册会话 +func (m *WsSessionManager) Register(deviceCode string, sess *WaitingSession) { + m.mu.Lock() + defer m.mu.Unlock() + m.sessions[deviceCode] = sess + logx.Infof("register ws session, deviceCode=%s", deviceCode) +} + +// Pop 取出并删除会话 +func (m *WsSessionManager) Pop(deviceCode string) *WaitingSession { + m.mu.Lock() + defer m.mu.Unlock() + sess, ok := m.sessions[deviceCode] + if !ok { + return nil + } + delete(m.sessions, deviceCode) + logx.Infof("pop ws session, deviceCode=%s", deviceCode) + return sess +} + +// Remove 移除会话 +func (m *WsSessionManager) Remove(deviceCode string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.sessions, deviceCode) + logx.Infof("remove ws session, deviceCode=%s", deviceCode) +} diff --git a/rpc/percutaneous/pb/percutaneous.pb.go b/rpc/percutaneous/pb/percutaneous.pb.go index d21dacf..0231d65 100644 --- a/rpc/percutaneous/pb/percutaneous.pb.go +++ b/rpc/percutaneous/pb/percutaneous.pb.go @@ -632,20 +632,15 @@ const file_percutaneous_percutaneous_proto_rawDesc = "" + "\thost_code\x18\x01 \x01(\tR\bhostCode\"\x0e\n" + "\fEmptyRequest\",\n" + "\x0fWsStreamRequest\x12\x19\n" + - "\braw_body\x18\x01 \x01(\fR\arawBody2\xd6\t\n" + + "\braw_body\x18\x01 \x01(\fR\arawBody2\xe4\x05\n" + "\fPercutaneous\x12o\n" + "\tWifiCheck\x12\x1e.percutaneous.WifiCheckRequest\x1a\x16.percutaneous.Response\"*\x82\xd3\xe4\x93\x02$:\x01*\x12\x1f/api/v3/percutaneous/wifi/check\x12i\n" + "\aWifiSet\x12\x1c.percutaneous.WifiSetRequest\x1a\x16.percutaneous.Response\"(\x82\xd3\xe4\x93\x02\":\x01*\"\x1d/api/v3/percutaneous/wifi/set\x12\\\n" + "\x03Use\x12\x18.percutaneous.UseRequest\x1a\x16.percutaneous.Response\"#\x82\xd3\xe4\x93\x02\x1d:\x01*\"\x18/api/v3/percutaneous/use\x12i\n" + "\aUsedLog\x12\x1c.percutaneous.UsedLogRequest\x1a\x16.percutaneous.Response\"(\x82\xd3\xe4\x93\x02\":\x01*\x12\x1d/api/v3/percutaneous/used/log\x12i\n" + "\aAddUser\x12\x1c.percutaneous.AddUserRequest\x1a\x16.percutaneous.Response\"(\x82\xd3\xe4\x93\x02\":\x01*\x12\x1d/api/v3/percutaneous/add/user\x12o\n" + - "\tUserItems\x12\x1e.percutaneous.UserItemsRequest\x1a\x16.percutaneous.Response\"*\x82\xd3\xe4\x93\x02$:\x01*\x12\x1f/api/v3/percutaneous/user/items\x12f\n" + - "\aWeather\x12\x1a.percutaneous.EmptyRequest\x1a\x16.percutaneous.Response\"'\x82\xd3\xe4\x93\x02!:\x01*\x12\x1c/api/v3/percutaneous/weather\x12m\n" + - "\aLoginQR\x12\x1c.percutaneous.LoginQRRequest\x1a\x16.percutaneous.Response\",\x82\xd3\xe4\x93\x02&:\x01*\x12!/api/v3/percutaneous/login/qrcode\x12t\n" + - "\tAddUserQR\x12\x1e.percutaneous.AddUserQRRequest\x1a\x16.percutaneous.Response\"/\x82\xd3\xe4\x93\x02):\x01*\x12$/api/v3/percutaneous/add/user/qrcode\x12Q\n" + - "\rWsLoginStream\x12\x1d.percutaneous.WsStreamRequest\x1a\x1d.percutaneous.WsStreamRequest(\x010\x01\x12S\n" + - "\x0fWsAddUserStream\x12\x1d.percutaneous.WsStreamRequest\x1a\x1d.percutaneous.WsStreamRequest(\x010\x01\x12P\n" + - "\fWsUsedStream\x12\x1d.percutaneous.WsStreamRequest\x1a\x1d.percutaneous.WsStreamRequest(\x010\x01B\x10Z\x0e./percutaneousb\x06proto3" + "\tUserItems\x12\x1e.percutaneous.UserItemsRequest\x1a\x16.percutaneous.Response\"*\x82\xd3\xe4\x93\x02$:\x01*\x12\x1f/api/v3/percutaneous/user/items\x12S\n" + + "\x0fWsAddUserStream\x12\x1d.percutaneous.WsStreamRequest\x1a\x1d.percutaneous.WsStreamRequest(\x010\x01B\x10Z\x0e./percutaneousb\x06proto3" var ( file_percutaneous_percutaneous_proto_rawDescOnce sync.Once @@ -680,26 +675,16 @@ var file_percutaneous_percutaneous_proto_depIdxs = []int32{ 3, // 3: percutaneous.Percutaneous.UsedLog:input_type -> percutaneous.UsedLogRequest 5, // 4: percutaneous.Percutaneous.AddUser:input_type -> percutaneous.AddUserRequest 6, // 5: percutaneous.Percutaneous.UserItems:input_type -> percutaneous.UserItemsRequest - 9, // 6: percutaneous.Percutaneous.Weather:input_type -> percutaneous.EmptyRequest - 7, // 7: percutaneous.Percutaneous.LoginQR:input_type -> percutaneous.LoginQRRequest - 8, // 8: percutaneous.Percutaneous.AddUserQR:input_type -> percutaneous.AddUserQRRequest - 10, // 9: percutaneous.Percutaneous.WsLoginStream:input_type -> percutaneous.WsStreamRequest - 10, // 10: percutaneous.Percutaneous.WsAddUserStream:input_type -> percutaneous.WsStreamRequest - 10, // 11: percutaneous.Percutaneous.WsUsedStream:input_type -> percutaneous.WsStreamRequest - 4, // 12: percutaneous.Percutaneous.WifiCheck:output_type -> percutaneous.Response - 4, // 13: percutaneous.Percutaneous.WifiSet:output_type -> percutaneous.Response - 4, // 14: percutaneous.Percutaneous.Use:output_type -> percutaneous.Response - 4, // 15: percutaneous.Percutaneous.UsedLog:output_type -> percutaneous.Response - 4, // 16: percutaneous.Percutaneous.AddUser:output_type -> percutaneous.Response - 4, // 17: percutaneous.Percutaneous.UserItems:output_type -> percutaneous.Response - 4, // 18: percutaneous.Percutaneous.Weather:output_type -> percutaneous.Response - 4, // 19: percutaneous.Percutaneous.LoginQR:output_type -> percutaneous.Response - 4, // 20: percutaneous.Percutaneous.AddUserQR:output_type -> percutaneous.Response - 10, // 21: percutaneous.Percutaneous.WsLoginStream:output_type -> percutaneous.WsStreamRequest - 10, // 22: percutaneous.Percutaneous.WsAddUserStream:output_type -> percutaneous.WsStreamRequest - 10, // 23: percutaneous.Percutaneous.WsUsedStream:output_type -> percutaneous.WsStreamRequest - 12, // [12:24] is the sub-list for method output_type - 0, // [0:12] is the sub-list for method input_type + 10, // 6: percutaneous.Percutaneous.WsAddUserStream:input_type -> percutaneous.WsStreamRequest + 4, // 7: percutaneous.Percutaneous.WifiCheck:output_type -> percutaneous.Response + 4, // 8: percutaneous.Percutaneous.WifiSet:output_type -> percutaneous.Response + 4, // 9: percutaneous.Percutaneous.Use:output_type -> percutaneous.Response + 4, // 10: percutaneous.Percutaneous.UsedLog:output_type -> percutaneous.Response + 4, // 11: percutaneous.Percutaneous.AddUser:output_type -> percutaneous.Response + 4, // 12: percutaneous.Percutaneous.UserItems:output_type -> percutaneous.Response + 10, // 13: percutaneous.Percutaneous.WsAddUserStream:output_type -> percutaneous.WsStreamRequest + 7, // [7:14] is the sub-list for method output_type + 0, // [0:7] is the sub-list for method input_type 0, // [0:0] is the sub-list for extension type_name 0, // [0:0] is the sub-list for extension extendee 0, // [0:0] is the sub-list for field type_name diff --git a/rpc/percutaneous/pb/percutaneous_grpc.pb.go b/rpc/percutaneous/pb/percutaneous_grpc.pb.go index 061d380..fff8242 100644 --- a/rpc/percutaneous/pb/percutaneous_grpc.pb.go +++ b/rpc/percutaneous/pb/percutaneous_grpc.pb.go @@ -25,12 +25,7 @@ const ( Percutaneous_UsedLog_FullMethodName = "/percutaneous.Percutaneous/UsedLog" Percutaneous_AddUser_FullMethodName = "/percutaneous.Percutaneous/AddUser" Percutaneous_UserItems_FullMethodName = "/percutaneous.Percutaneous/UserItems" - Percutaneous_Weather_FullMethodName = "/percutaneous.Percutaneous/Weather" - Percutaneous_LoginQR_FullMethodName = "/percutaneous.Percutaneous/LoginQR" - Percutaneous_AddUserQR_FullMethodName = "/percutaneous.Percutaneous/AddUserQR" - Percutaneous_WsLoginStream_FullMethodName = "/percutaneous.Percutaneous/WsLoginStream" Percutaneous_WsAddUserStream_FullMethodName = "/percutaneous.Percutaneous/WsAddUserStream" - Percutaneous_WsUsedStream_FullMethodName = "/percutaneous.Percutaneous/WsUsedStream" ) // PercutaneousClient is the client API for Percutaneous service. @@ -43,17 +38,8 @@ type PercutaneousClient interface { UsedLog(ctx context.Context, in *UsedLogRequest, opts ...grpc.CallOption) (*Response, error) AddUser(ctx context.Context, in *AddUserRequest, opts ...grpc.CallOption) (*Response, error) UserItems(ctx context.Context, in *UserItemsRequest, opts ...grpc.CallOption) (*Response, error) - Weather(ctx context.Context, in *EmptyRequest, opts ...grpc.CallOption) (*Response, error) - // 登录二维码 - LoginQR(ctx context.Context, in *LoginQRRequest, opts ...grpc.CallOption) (*Response, error) - // 添加用户 - AddUserQR(ctx context.Context, in *AddUserQRRequest, opts ...grpc.CallOption) (*Response, error) - // WebSocket登录 - WsLoginStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) // 添加用户 WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) - // 使用 - WsUsedStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) } type percutaneousClient struct { @@ -124,52 +110,9 @@ func (c *percutaneousClient) UserItems(ctx context.Context, in *UserItemsRequest return out, nil } -func (c *percutaneousClient) Weather(ctx context.Context, in *EmptyRequest, opts ...grpc.CallOption) (*Response, error) { - cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(Response) - err := c.cc.Invoke(ctx, Percutaneous_Weather_FullMethodName, in, out, cOpts...) - if err != nil { - return nil, err - } - return out, nil -} - -func (c *percutaneousClient) LoginQR(ctx context.Context, in *LoginQRRequest, opts ...grpc.CallOption) (*Response, error) { - cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(Response) - err := c.cc.Invoke(ctx, Percutaneous_LoginQR_FullMethodName, in, out, cOpts...) - if err != nil { - return nil, err - } - return out, nil -} - -func (c *percutaneousClient) AddUserQR(ctx context.Context, in *AddUserQRRequest, opts ...grpc.CallOption) (*Response, error) { - cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(Response) - err := c.cc.Invoke(ctx, Percutaneous_AddUserQR_FullMethodName, in, out, cOpts...) - if err != nil { - return nil, err - } - return out, nil -} - -func (c *percutaneousClient) WsLoginStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) { - cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Percutaneous_ServiceDesc.Streams[0], Percutaneous_WsLoginStream_FullMethodName, cOpts...) - if err != nil { - return nil, err - } - x := &grpc.GenericClientStream[WsStreamRequest, WsStreamRequest]{ClientStream: stream} - return x, nil -} - -// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type Percutaneous_WsLoginStreamClient = grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest] - func (c *percutaneousClient) WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Percutaneous_ServiceDesc.Streams[1], Percutaneous_WsAddUserStream_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &Percutaneous_ServiceDesc.Streams[0], Percutaneous_WsAddUserStream_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -180,19 +123,6 @@ func (c *percutaneousClient) WsAddUserStream(ctx context.Context, opts ...grpc.C // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Percutaneous_WsAddUserStreamClient = grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest] -func (c *percutaneousClient) WsUsedStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) { - cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Percutaneous_ServiceDesc.Streams[2], Percutaneous_WsUsedStream_FullMethodName, cOpts...) - if err != nil { - return nil, err - } - x := &grpc.GenericClientStream[WsStreamRequest, WsStreamRequest]{ClientStream: stream} - return x, nil -} - -// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type Percutaneous_WsUsedStreamClient = grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest] - // PercutaneousServer is the server API for Percutaneous service. // All implementations must embed UnimplementedPercutaneousServer // for forward compatibility. @@ -203,17 +133,8 @@ type PercutaneousServer interface { UsedLog(context.Context, *UsedLogRequest) (*Response, error) AddUser(context.Context, *AddUserRequest) (*Response, error) UserItems(context.Context, *UserItemsRequest) (*Response, error) - Weather(context.Context, *EmptyRequest) (*Response, error) - // 登录二维码 - LoginQR(context.Context, *LoginQRRequest) (*Response, error) - // 添加用户 - AddUserQR(context.Context, *AddUserQRRequest) (*Response, error) - // WebSocket登录 - WsLoginStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error // 添加用户 WsAddUserStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error - // 使用 - WsUsedStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error mustEmbedUnimplementedPercutaneousServer() } @@ -242,24 +163,9 @@ func (UnimplementedPercutaneousServer) AddUser(context.Context, *AddUserRequest) func (UnimplementedPercutaneousServer) UserItems(context.Context, *UserItemsRequest) (*Response, error) { return nil, status.Error(codes.Unimplemented, "method UserItems not implemented") } -func (UnimplementedPercutaneousServer) Weather(context.Context, *EmptyRequest) (*Response, error) { - return nil, status.Error(codes.Unimplemented, "method Weather not implemented") -} -func (UnimplementedPercutaneousServer) LoginQR(context.Context, *LoginQRRequest) (*Response, error) { - return nil, status.Error(codes.Unimplemented, "method LoginQR not implemented") -} -func (UnimplementedPercutaneousServer) AddUserQR(context.Context, *AddUserQRRequest) (*Response, error) { - return nil, status.Error(codes.Unimplemented, "method AddUserQR not implemented") -} -func (UnimplementedPercutaneousServer) WsLoginStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error { - return status.Error(codes.Unimplemented, "method WsLoginStream not implemented") -} func (UnimplementedPercutaneousServer) WsAddUserStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error { return status.Error(codes.Unimplemented, "method WsAddUserStream not implemented") } -func (UnimplementedPercutaneousServer) WsUsedStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error { - return status.Error(codes.Unimplemented, "method WsUsedStream not implemented") -} func (UnimplementedPercutaneousServer) mustEmbedUnimplementedPercutaneousServer() {} func (UnimplementedPercutaneousServer) testEmbeddedByValue() {} @@ -389,67 +295,6 @@ func _Percutaneous_UserItems_Handler(srv interface{}, ctx context.Context, dec f return interceptor(ctx, in, info, handler) } -func _Percutaneous_Weather_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(EmptyRequest) - if err := dec(in); err != nil { - return nil, err - } - if interceptor == nil { - return srv.(PercutaneousServer).Weather(ctx, in) - } - info := &grpc.UnaryServerInfo{ - Server: srv, - FullMethod: Percutaneous_Weather_FullMethodName, - } - handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(PercutaneousServer).Weather(ctx, req.(*EmptyRequest)) - } - return interceptor(ctx, in, info, handler) -} - -func _Percutaneous_LoginQR_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(LoginQRRequest) - if err := dec(in); err != nil { - return nil, err - } - if interceptor == nil { - return srv.(PercutaneousServer).LoginQR(ctx, in) - } - info := &grpc.UnaryServerInfo{ - Server: srv, - FullMethod: Percutaneous_LoginQR_FullMethodName, - } - handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(PercutaneousServer).LoginQR(ctx, req.(*LoginQRRequest)) - } - return interceptor(ctx, in, info, handler) -} - -func _Percutaneous_AddUserQR_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(AddUserQRRequest) - if err := dec(in); err != nil { - return nil, err - } - if interceptor == nil { - return srv.(PercutaneousServer).AddUserQR(ctx, in) - } - info := &grpc.UnaryServerInfo{ - Server: srv, - FullMethod: Percutaneous_AddUserQR_FullMethodName, - } - handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(PercutaneousServer).AddUserQR(ctx, req.(*AddUserQRRequest)) - } - return interceptor(ctx, in, info, handler) -} - -func _Percutaneous_WsLoginStream_Handler(srv interface{}, stream grpc.ServerStream) error { - return srv.(PercutaneousServer).WsLoginStream(&grpc.GenericServerStream[WsStreamRequest, WsStreamRequest]{ServerStream: stream}) -} - -// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type Percutaneous_WsLoginStreamServer = grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest] - func _Percutaneous_WsAddUserStream_Handler(srv interface{}, stream grpc.ServerStream) error { return srv.(PercutaneousServer).WsAddUserStream(&grpc.GenericServerStream[WsStreamRequest, WsStreamRequest]{ServerStream: stream}) } @@ -457,13 +302,6 @@ func _Percutaneous_WsAddUserStream_Handler(srv interface{}, stream grpc.ServerSt // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Percutaneous_WsAddUserStreamServer = grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest] -func _Percutaneous_WsUsedStream_Handler(srv interface{}, stream grpc.ServerStream) error { - return srv.(PercutaneousServer).WsUsedStream(&grpc.GenericServerStream[WsStreamRequest, WsStreamRequest]{ServerStream: stream}) -} - -// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type Percutaneous_WsUsedStreamServer = grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest] - // Percutaneous_ServiceDesc is the grpc.ServiceDesc for Percutaneous service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -495,38 +333,14 @@ var Percutaneous_ServiceDesc = grpc.ServiceDesc{ MethodName: "UserItems", Handler: _Percutaneous_UserItems_Handler, }, - { - MethodName: "Weather", - Handler: _Percutaneous_Weather_Handler, - }, - { - MethodName: "LoginQR", - Handler: _Percutaneous_LoginQR_Handler, - }, - { - MethodName: "AddUserQR", - Handler: _Percutaneous_AddUserQR_Handler, - }, }, Streams: []grpc.StreamDesc{ - { - StreamName: "WsLoginStream", - Handler: _Percutaneous_WsLoginStream_Handler, - ServerStreams: true, - ClientStreams: true, - }, { StreamName: "WsAddUserStream", Handler: _Percutaneous_WsAddUserStream_Handler, ServerStreams: true, ClientStreams: true, }, - { - StreamName: "WsUsedStream", - Handler: _Percutaneous_WsUsedStream_Handler, - ServerStreams: true, - ClientStreams: true, - }, }, Metadata: "percutaneous/percutaneous.proto", } diff --git a/rpc/percutaneous/percutaneous.pb b/rpc/percutaneous/percutaneous.pb index 1bfe5d1e2a248bd947688f5b6597d3c4eb60e230..758dd316df37da3ad3cf9eedea2e4cbe24395b6a 100644 GIT binary patch delta 40 wcmZp!m{_vmu`$ycp3P5F0i+gmSsqee%;Y^8$l}WKcCjg~8f& zz}nS>6hYceb3hvO3yYHTQ&K@@lyGr6rlf=xrxpQCkwY~FE&(=2A8d}MkP6ToSy3b- h0=am@i@`QdUT^6!d68u(mrwu~Pk1rV)D)ml5&*}=RPF!( diff --git a/services/percutaneous/Dockerfile b/services/percutaneous/Dockerfile new file mode 100644 index 0000000..23f1fbd --- /dev/null +++ b/services/percutaneous/Dockerfile @@ -0,0 +1,37 @@ +FROM golang:1.26.5-alpine AS builder + +WORKDIR /src + +ENV GOPROXY=https://goproxy.cn,direct \ + CGO_ENABLED=0 \ + GOOS=linux \ + GOARCH=amd64 + +COPY pkg ./pkg +COPY go.mod go.sum ./ + +RUN --mount=type=cache,target=/go/pkg/mod \ + go mod download + +COPY rpc/ ./rpc +COPY services/percutaneous/ ./services/percutaneous/ + +WORKDIR /src/services/percutaneous + +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + go build -ldflags="-s -w" -o /out/main ./ + +FROM alpine:3.22 + +WORKDIR /app + +ENV TZ=Asia/Shanghai + +RUN apk add --no-cache tzdata \ + && cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ + && echo "Asia/Shanghai" > /etc/timezone + +COPY --from=builder /out/main . + +CMD ["./main", "-f", "etc/percutaneous.yaml"] diff --git a/services/percutaneous/internal/logic/addUserLogic.go b/services/percutaneous/internal/logic/addUserLogic.go index 2e5ed1a..a92365d 100644 --- a/services/percutaneous/internal/logic/addUserLogic.go +++ b/services/percutaneous/internal/logic/addUserLogic.go @@ -41,9 +41,10 @@ func (l *AddUserLogic) AddUser(in *percutaneous.AddUserRequest) (*percutaneous.R } w := modelbase.Params{Eq: map[string]string{ "host_code": in.HostCode, - "host_user_id": strconv.FormatInt(adminInfo.ID, utils.NumberTen), - "user_id": strconv.FormatInt(in.UserId, utils.NumberTen), + "host_user_id": strconv.FormatInt(in.UserId, utils.NumberTen), + "user_id": strconv.FormatInt(adminInfo.ID, utils.NumberTen), }} + var info dao.Group err := model.GroupModel{}.Init().GetOne(w, &info) if err != nil { @@ -61,5 +62,13 @@ func (l *AddUserLogic) AddUser(in *percutaneous.AddUserRequest) (*percutaneous.R } } + // 取出等待中的ws会话 + sess := l.svcCtx.SessionMgr.Pop(in.HostCode) + if sess == nil { + l.Logger.Error("no waiting ws session code is = ", in.HostCode) + return l.ok(utils.NumberOne) + } + + sess.MsgCh <- []byte(strconv.FormatInt(adminInfo.ID, utils.NumberTen)) return l.ok(utils.NumberOne) } diff --git a/services/percutaneous/internal/logic/wsAddUserStreamLogic.go b/services/percutaneous/internal/logic/wsAddUserStreamLogic.go index 5057c33..fccf268 100644 --- a/services/percutaneous/internal/logic/wsAddUserStreamLogic.go +++ b/services/percutaneous/internal/logic/wsAddUserStreamLogic.go @@ -3,6 +3,8 @@ package logic import ( "context" "lone-services/pkg/utils" + user "lone-services/rpc/user/pb" + "strconv" "time" "lone-services/rpc/percutaneous/pb" @@ -13,12 +15,29 @@ import ( "google.golang.org/grpc/status" ) +type grpcStreamWrapper struct { + stream percutaneous.Percutaneous_WsAddUserStreamServer +} + +func (w *grpcStreamWrapper) Send(payload []byte) error { + frame := &percutaneous.WsStreamRequest{ + RawBody: payload, + } + return w.stream.Send(frame) +} + type WsAddUserStreamLogic struct { ctx context.Context svcCtx *svc.ServiceContext logx.Logger } +type userInfo struct { + Id int64 `json:"id"` + Name string `json:"name"` + Avatar string `json:"avatar"` +} + func NewWsAddUserStreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *WsAddUserStreamLogic { return &WsAddUserStreamLogic{ ctx: ctx, @@ -29,50 +48,85 @@ func NewWsAddUserStreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *W // 添加用户 func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error { - l.Logger.Error("in coming ws tunnel stream") + l.Logger.Info("in coming ws tunnel stream") ctx := stream.Context() - // 扫码超时30s - timeout := 30 * time.Second + timeout := time.Duration(utils.GetConfigInt("ws.out_time")) * time.Second ctx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() + defer func() { + cancel() + }() - for { - // 先判断上下文是否已经结束 - if err := ctx.Err(); err != nil { - if err == context.DeadlineExceeded { - return status.Errorf(codes.DeadlineExceeded, "扫码超时") - } + // BFF自动发来第一条消息，读取host_code + recvFrame, err := stream.Recv() + if err != nil { + l.Logger.Errorf("recv host_code failed: %v", err) + return err + } + var hostCode struct { + Code string `json:"host_code"` + } + l.Logger.Errorf(" host_code raw msg is: %v", string(recvFrame.RawBody)) + // 保留你确认的 !err 判断 + if ok := utils.JsonStringToStruct(string(recvFrame.RawBody), &hostCode); !ok { + return status.Errorf(codes.InvalidArgument, "device code parse error") + } + l.Logger.Infof("device connected, code=%s", hostCode.Code) + + msgCh := make(chan []byte, 1) + wrapper := &grpcStreamWrapper{stream: stream} + sess := &utils.WaitingSession{ + Stream: wrapper, + Ctx: ctx, + Cancel: cancel, + MsgCh: msgCh, + } + l.svcCtx.SessionMgr.Register(hostCode.Code, sess) + defer l.svcCtx.SessionMgr.Remove(hostCode.Code) + l.Logger.Info("session register success, waiting for scan event") + + select { + case <-ctx.Done(): + if ctx.Err() == context.DeadlineExceeded { + return status.Errorf(codes.DeadlineExceeded, "扫码等待超时") + } + return ctx.Err() + case payload := <-msgCh: + id, _ := strconv.Atoi(string(payload)) + info := l.getUserInfo(int64(id)) + jsonData := utils.StructToJson(info) + if err := sess.Stream.Send([]byte(jsonData)); err != nil { + l.Logger.Errorf("send to device err: %v", err) return err } - - // 阻塞等待客户端上报扫码消息 - recvFrame, err := stream.Recv() - if err != nil { - // 客户端主动关闭流 - return err - } - - // 解析扫码payload - type res struct { - Id int64 `json:"id"` - Name string `json:"name"` - } - var resData res - if err := utils.JsonStringToStruct(string(recvFrame.RawBody), &resData); !err { - l.Logger.Errorf("parse scan payload failed: %v", err) - return status.Errorf(codes.InvalidArgument, "解析扫码数据失败") - } - l.Logger.Errorf("receive scan success data: %+v", resData) - - // 下发结果给ws客户端（同步设备） - sendFrame := &percutaneous.WsStreamRequest{ - RawBody: []byte(utils.StructToJson(resData)), - } - if err := stream.Send(sendFrame); err != nil { - l.Logger.Errorf("send result frame failed: %v", err) - return err - } - // 收发完成，直接关闭grpc双向流（ws连接随之断开） return nil } } + +func (l *WsAddUserStreamLogic) getUserInfo(id int64) *userInfo { + if id == utils.NumberZero { + return nil + } + logx.Errorf("get user id: %v", id) + cli, err := svc.GetRpcClient(l.svcCtx.UserName) + if err != nil { + logx.Errorf("get user err: %v", err) + return nil + } + + client := user.NewUserClient(cli.Conn()) + + data := user.InternalGetUserInfoReq{ + UserId: id, + } + res, err := client.InternalGetUserInfo(context.Background(), &data) + if err != nil { + logx.Errorf("get user err: %v", err) + return nil + } + + return &userInfo{ + Id: res.Id, + Name: res.Name, + Avatar: res.Avatar, + } +} diff --git a/services/percutaneous/internal/server/percutaneousServer.go b/services/percutaneous/internal/server/percutaneousServer.go index 681326a..d62d2cc 100644 --- a/services/percutaneous/internal/server/percutaneousServer.go +++ b/services/percutaneous/internal/server/percutaneousServer.go @@ -53,37 +53,8 @@ func (s *PercutaneousServer) UserItems(ctx context.Context, in *percutaneous.Use return l.UserItems(in) } -func (s *PercutaneousServer) Weather(ctx context.Context, in *percutaneous.EmptyRequest) (*percutaneous.Response, error) { - l := logic.NewWeatherLogic(ctx, s.svcCtx) - return l.Weather(in) -} - -// 登录二维码 -func (s *PercutaneousServer) LoginQR(ctx context.Context, in *percutaneous.LoginQRRequest) (*percutaneous.Response, error) { - l := logic.NewLoginQRLogic(ctx, s.svcCtx) - return l.LoginQR(in) -} - -// 添加用户 -func (s *PercutaneousServer) AddUserQR(ctx context.Context, in *percutaneous.AddUserQRRequest) (*percutaneous.Response, error) { - l := logic.NewAddUserQRLogic(ctx, s.svcCtx) - return l.AddUserQR(in) -} - -// WebSocket登录 -func (s *PercutaneousServer) WsLoginStream(stream percutaneous.Percutaneous_WsLoginStreamServer) error { - l := logic.NewWsLoginStreamLogic(stream.Context(), s.svcCtx) - return l.WsLoginStream(stream) -} - // 添加用户 func (s *PercutaneousServer) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error { l := logic.NewWsAddUserStreamLogic(stream.Context(), s.svcCtx) return l.WsAddUserStream(stream) } - -// 使用 -func (s *PercutaneousServer) WsUsedStream(stream percutaneous.Percutaneous_WsUsedStreamServer) error { - l := logic.NewWsUsedStreamLogic(stream.Context(), s.svcCtx) - return l.WsUsedStream(stream) -} diff --git a/services/percutaneous/internal/svc/servicecontext.go b/services/percutaneous/internal/svc/servicecontext.go index b5d262f..32b220e 100644 --- a/services/percutaneous/internal/svc/servicecontext.go +++ b/services/percutaneous/internal/svc/servicecontext.go @@ -72,10 +72,11 @@ func GetRpcClient(serviceName string) (zrpc.Client, error) { } type ServiceContext struct { - Config config.Config - DB *gorm.DB - Prefix string - UserName string + Config config.Config + DB *gorm.DB + Prefix string + UserName string + SessionMgr *utils.WsSessionManager } func NewServiceContext(c config.Config, db *gorm.DB) *ServiceContext { @@ -86,9 +87,10 @@ func NewServiceContext(c config.Config, db *gorm.DB) *ServiceContext { } return &ServiceContext{ - Config: c, - DB: db, - Prefix: utils.GetConfigString("mysql.prefix"), - UserName: UserSvc, + Config: c, + DB: db, + Prefix: utils.GetConfigString("mysql.prefix"), + UserName: UserSvc, + SessionMgr: utils.NewWsSessionManager(), } } diff --git a/services/percutaneous/percutaneousclient/percutaneous.go b/services/percutaneous/percutaneousclient/percutaneous.go index 17f616b..fbdc901 100644 --- a/services/percutaneous/percutaneousclient/percutaneous.go +++ b/services/percutaneous/percutaneousclient/percutaneous.go @@ -33,17 +33,8 @@ type ( UsedLog(ctx context.Context, in *UsedLogRequest, opts ...grpc.CallOption) (*Response, error) AddUser(ctx context.Context, in *AddUserRequest, opts ...grpc.CallOption) (*Response, error) UserItems(ctx context.Context, in *UserItemsRequest, opts ...grpc.CallOption) (*Response, error) - Weather(ctx context.Context, in *EmptyRequest, opts ...grpc.CallOption) (*Response, error) - // 登录二维码 - LoginQR(ctx context.Context, in *LoginQRRequest, opts ...grpc.CallOption) (*Response, error) - // 添加用户 - AddUserQR(ctx context.Context, in *AddUserQRRequest, opts ...grpc.CallOption) (*Response, error) - // WebSocket登录 - WsLoginStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsLoginStreamClient, error) // 添加用户 WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsAddUserStreamClient, error) - // 使用 - WsUsedStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsUsedStreamClient, error) } defaultPercutaneous struct { @@ -87,37 +78,8 @@ func (m *defaultPercutaneous) UserItems(ctx context.Context, in *UserItemsReques return client.UserItems(ctx, in, opts...) } -func (m *defaultPercutaneous) Weather(ctx context.Context, in *EmptyRequest, opts ...grpc.CallOption) (*Response, error) { - client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.Weather(ctx, in, opts...) -} - -// 登录二维码 -func (m *defaultPercutaneous) LoginQR(ctx context.Context, in *LoginQRRequest, opts ...grpc.CallOption) (*Response, error) { - client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.LoginQR(ctx, in, opts...) -} - -// 添加用户 -func (m *defaultPercutaneous) AddUserQR(ctx context.Context, in *AddUserQRRequest, opts ...grpc.CallOption) (*Response, error) { - client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.AddUserQR(ctx, in, opts...) -} - -// WebSocket登录 -func (m *defaultPercutaneous) WsLoginStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsLoginStreamClient, error) { - client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.WsLoginStream(ctx, opts...) -} - // 添加用户 func (m *defaultPercutaneous) WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsAddUserStreamClient, error) { client := percutaneous.NewPercutaneousClient(m.cli.Conn()) return client.WsAddUserStream(ctx, opts...) } - -// 使用 -func (m *defaultPercutaneous) WsUsedStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsUsedStreamClient, error) { - client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.WsUsedStream(ctx, opts...) -}