From 6b8500ac5e9f06f21d4e519a849fc5d962eb6f1c Mon Sep 17 00:00:00 2001 From: gjs Date: Thu, 8 Oct 2026 14:14:39 +0800 Subject: [PATCH] ws --- bff/bff.go | 46 +++- bff/internal/config/config.go | 5 +- bff/internal/request/user.go | 37 +++ bff/internal/wsClient/client.go | 80 +++++++ bff/internal/wsproxy/proxy.go | 123 ++++++++++ deploy/apisix/apisix.yaml | 27 ++- go.mod | 2 +- rpc/percutaneous/pb/percutaneous.pb.go | 221 +++++++++++++++--- rpc/percutaneous/pb/percutaneous_grpc.pb.go | 199 +++++++++++++++- rpc/percutaneous/percutaneous.pb | Bin 14712 -> 15363 bytes rpc/percutaneous/percutaneous.proto | 24 +- .../internal/logic/weatherLogic.go | 30 --- .../internal/logic/wsAddUserStreamLogic.go | 78 +++++++ .../internal/server/percutaneousServer.go | 30 +++ .../percutaneousclient/percutaneous.go | 43 ++++ services/sale/.gitignore | 3 +- 16 files changed, 862 insertions(+), 86 deletions(-) create mode 100644 bff/internal/wsClient/client.go create mode 100644 bff/internal/wsproxy/proxy.go delete mode 100644 services/percutaneous/internal/logic/weatherLogic.go create mode 100644 services/percutaneous/internal/logic/wsAddUserStreamLogic.go diff --git a/bff/bff.go b/bff/bff.go index 6165a89..9c110c7 100644 --- a/bff/bff.go +++ b/bff/bff.go @@ -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,50 @@ 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/login", + Handler: wsClient.PercutaneousLogin(rpcMgr), + }, + { + Method: http.MethodGet, + Path: "/ws/percutaneous/used", + Handler: wsClient.PercutaneousUsed(rpcMgr), + }, + { + 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() } diff --git a/bff/internal/config/config.go b/bff/internal/config/config.go index df019d3..04b5957 100644 --- a/bff/internal/config/config.go +++ b/bff/internal/config/config.go @@ -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 { diff --git a/bff/internal/request/user.go b/bff/internal/request/user.go index 125bed6..b69a977 100644 --- a/bff/internal/request/user.go +++ b/bff/internal/request/user.go @@ -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...) +} diff --git a/bff/internal/wsClient/client.go b/bff/internal/wsClient/client.go new file mode 100644 index 0000000..075e4f0 --- /dev/null +++ b/bff/internal/wsClient/client.go @@ -0,0 +1,80 @@ +package wsClient + +import ( + "context" + "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" +) + +// 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) { + 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.WsAddUserStream(ctx) + }, + 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{} }, + ) + proxyHandler(w, r) + }) + + return wsHandler +} diff --git a/bff/internal/wsproxy/proxy.go b/bff/internal/wsproxy/proxy.go new file mode 100644 index 0000000..575912f --- /dev/null +++ b/bff/internal/wsproxy/proxy.go @@ -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 +} diff --git a/deploy/apisix/apisix.yaml b/deploy/apisix/apisix.yaml index 9facdaa..669455e 100644 --- a/deploy/apisix/apisix.yaml +++ b/deploy/apisix/apisix.yaml @@ -9,8 +9,8 @@ plugin_configs: local mod = f() return mod.access routes: - - id: bff - name: bff + - id: bff-api + name: bff-api priority: 10 uris: - /admin/* @@ -25,10 +25,29 @@ routes: cors: allow_origins: "*" allow_methods: "GET,POST,PUT,OPTIONS" - allow_headers: "Authorization,Authorization-Type,Authorization-Auth,Refresh,X-Refresh,X-Service-Code" + allow_headers: "Authorization,Authorization-Type,Authorization-Auth,Refresh,X-Refresh,X-Service-Code,Sec-WebSocket-Key,Sec-WebSocket-Version,Sec-WebSocket-Accept,Connection,Upgrade" max_age: 86400 upstream: type: roundrobin nodes: "bff:10000": 1 -#END \ No newline at end of file + + - id: bff-ws + name: bff-ws + priority: 20 + uris: + - /ws/* + methods: + - GET + enable_websocket: true # 放到route顶层，不在plugins内部！ + plugins: + cors: + allow_origins: "*" + allow_methods: "GET" + allow_headers: "Authorization,Authorization-Type,Authorization-Auth,Refresh,X-Refresh,X-Service-Code,Sec-WebSocket-Key,Sec-WebSocket-Version,Sec-WebSocket-Accept,Connection,Upgrade" + max_age: 86400 + upstream: + type: roundrobin + nodes: + "bff:10001": 1 +#END diff --git a/go.mod b/go.mod index c2c7969..b852c63 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/alibabacloud-go/darabonba-openapi v0.2.1 github.com/alibabacloud-go/darabonba-openapi/v2 v2.2.4 github.com/alibabacloud-go/dysmsapi-20170525/v2 v2.0.18 + github.com/alibabacloud-go/openapi-util v0.1.0 github.com/alibabacloud-go/sts-20150401/v2 v2.1.0 github.com/alibabacloud-go/tea v1.5.3 github.com/alibabacloud-go/tea-utils/v2 v2.0.9 @@ -40,7 +41,6 @@ require ( github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 // indirect github.com/alibabacloud-go/debug v1.0.1 // indirect github.com/alibabacloud-go/endpoint-util v1.1.0 // indirect - github.com/alibabacloud-go/openapi-util v0.1.0 // indirect github.com/alibabacloud-go/tea-utils v1.4.5 // indirect github.com/alibabacloud-go/tea-xml v1.1.2 // indirect github.com/aliyun/credentials-go v1.4.13 // indirect diff --git a/rpc/percutaneous/pb/percutaneous.pb.go b/rpc/percutaneous/pb/percutaneous.pb.go index 6b4517e..d21dacf 100644 --- a/rpc/percutaneous/pb/percutaneous.pb.go +++ b/rpc/percutaneous/pb/percutaneous.pb.go @@ -418,6 +418,102 @@ func (x *UserItemsRequest) GetSize() int32 { return 0 } +type LoginQRRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + HostCode string `protobuf:"bytes,1,opt,name=host_code,json=hostCode,proto3" json:"host_code,omitempty"` + Bluetooth string `protobuf:"bytes,2,opt,name=bluetooth,proto3" json:"bluetooth,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LoginQRRequest) Reset() { + *x = LoginQRRequest{} + mi := &file_percutaneous_percutaneous_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LoginQRRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LoginQRRequest) ProtoMessage() {} + +func (x *LoginQRRequest) ProtoReflect() protoreflect.Message { + mi := &file_percutaneous_percutaneous_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LoginQRRequest.ProtoReflect.Descriptor instead. +func (*LoginQRRequest) Descriptor() ([]byte, []int) { + return file_percutaneous_percutaneous_proto_rawDescGZIP(), []int{7} +} + +func (x *LoginQRRequest) GetHostCode() string { + if x != nil { + return x.HostCode + } + return "" +} + +func (x *LoginQRRequest) GetBluetooth() string { + if x != nil { + return x.Bluetooth + } + return "" +} + +type AddUserQRRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + HostCode string `protobuf:"bytes,1,opt,name=host_code,json=hostCode,proto3" json:"host_code,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AddUserQRRequest) Reset() { + *x = AddUserQRRequest{} + mi := &file_percutaneous_percutaneous_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AddUserQRRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AddUserQRRequest) ProtoMessage() {} + +func (x *AddUserQRRequest) ProtoReflect() protoreflect.Message { + mi := &file_percutaneous_percutaneous_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AddUserQRRequest.ProtoReflect.Descriptor instead. +func (*AddUserQRRequest) Descriptor() ([]byte, []int) { + return file_percutaneous_percutaneous_proto_rawDescGZIP(), []int{8} +} + +func (x *AddUserQRRequest) GetHostCode() string { + if x != nil { + return x.HostCode + } + return "" +} + type EmptyRequest struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -426,7 +522,7 @@ type EmptyRequest struct { func (x *EmptyRequest) Reset() { *x = EmptyRequest{} - mi := &file_percutaneous_percutaneous_proto_msgTypes[7] + mi := &file_percutaneous_percutaneous_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -438,7 +534,7 @@ func (x *EmptyRequest) String() string { func (*EmptyRequest) ProtoMessage() {} func (x *EmptyRequest) ProtoReflect() protoreflect.Message { - mi := &file_percutaneous_percutaneous_proto_msgTypes[7] + mi := &file_percutaneous_percutaneous_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -451,7 +547,51 @@ func (x *EmptyRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use EmptyRequest.ProtoReflect.Descriptor instead. func (*EmptyRequest) Descriptor() ([]byte, []int) { - return file_percutaneous_percutaneous_proto_rawDescGZIP(), []int{7} + return file_percutaneous_percutaneous_proto_rawDescGZIP(), []int{9} +} + +type WsStreamRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + RawBody []byte `protobuf:"bytes,1,opt,name=raw_body,json=rawBody,proto3" json:"raw_body,omitempty"` // websocket原始业务报文，不做解析 + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *WsStreamRequest) Reset() { + *x = WsStreamRequest{} + mi := &file_percutaneous_percutaneous_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *WsStreamRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*WsStreamRequest) ProtoMessage() {} + +func (x *WsStreamRequest) ProtoReflect() protoreflect.Message { + mi := &file_percutaneous_percutaneous_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use WsStreamRequest.ProtoReflect.Descriptor instead. +func (*WsStreamRequest) Descriptor() ([]byte, []int) { + return file_percutaneous_percutaneous_proto_rawDescGZIP(), []int{10} +} + +func (x *WsStreamRequest) GetRawBody() []byte { + if x != nil { + return x.RawBody + } + return nil } var File_percutaneous_percutaneous_proto protoreflect.FileDescriptor @@ -484,8 +624,15 @@ const file_percutaneous_percutaneous_proto_rawDesc = "" + "\x10UserItemsRequest\x12\x1b\n" + "\thost_code\x18\x01 \x01(\tR\bhostCode\x12\x12\n" + "\x04page\x18\x02 \x01(\x05R\x04page\x12\x12\n" + - "\x04size\x18\x03 \x01(\x05R\x04size\"\x0e\n" + - "\fEmptyRequest2\xf7\x05\n" + + "\x04size\x18\x03 \x01(\x05R\x04size\"K\n" + + "\x0eLoginQRRequest\x12\x1b\n" + + "\thost_code\x18\x01 \x01(\tR\bhostCode\x12\x1c\n" + + "\tbluetooth\x18\x02 \x01(\tR\tbluetooth\"/\n" + + "\x10AddUserQRRequest\x12\x1b\n" + + "\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" + "\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" + @@ -493,7 +640,12 @@ const file_percutaneous_percutaneous_proto_rawDesc = "" + "\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/weatherB\x10Z\x0e./percutaneousb\x06proto3" + "\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" var ( file_percutaneous_percutaneous_proto_rawDescOnce sync.Once @@ -507,7 +659,7 @@ func file_percutaneous_percutaneous_proto_rawDescGZIP() []byte { return file_percutaneous_percutaneous_proto_rawDescData } -var file_percutaneous_percutaneous_proto_msgTypes = make([]protoimpl.MessageInfo, 8) +var file_percutaneous_percutaneous_proto_msgTypes = make([]protoimpl.MessageInfo, 11) var file_percutaneous_percutaneous_proto_goTypes = []any{ (*WifiCheckRequest)(nil), // 0: percutaneous.WifiCheckRequest (*WifiSetRequest)(nil), // 1: percutaneous.WifiSetRequest @@ -516,28 +668,41 @@ var file_percutaneous_percutaneous_proto_goTypes = []any{ (*Response)(nil), // 4: percutaneous.Response (*AddUserRequest)(nil), // 5: percutaneous.AddUserRequest (*UserItemsRequest)(nil), // 6: percutaneous.UserItemsRequest - (*EmptyRequest)(nil), // 7: percutaneous.EmptyRequest + (*LoginQRRequest)(nil), // 7: percutaneous.LoginQRRequest + (*AddUserQRRequest)(nil), // 8: percutaneous.AddUserQRRequest + (*EmptyRequest)(nil), // 9: percutaneous.EmptyRequest + (*WsStreamRequest)(nil), // 10: percutaneous.WsStreamRequest } var file_percutaneous_percutaneous_proto_depIdxs = []int32{ - 0, // 0: percutaneous.Percutaneous.WifiCheck:input_type -> percutaneous.WifiCheckRequest - 1, // 1: percutaneous.Percutaneous.WifiSet:input_type -> percutaneous.WifiSetRequest - 2, // 2: percutaneous.Percutaneous.Use:input_type -> percutaneous.UseRequest - 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 - 7, // 6: percutaneous.Percutaneous.Weather:input_type -> percutaneous.EmptyRequest - 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 - 4, // 13: percutaneous.Percutaneous.Weather:output_type -> percutaneous.Response - 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 + 0, // 0: percutaneous.Percutaneous.WifiCheck:input_type -> percutaneous.WifiCheckRequest + 1, // 1: percutaneous.Percutaneous.WifiSet:input_type -> percutaneous.WifiSetRequest + 2, // 2: percutaneous.Percutaneous.Use:input_type -> percutaneous.UseRequest + 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 + 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 } func init() { file_percutaneous_percutaneous_proto_init() } @@ -551,7 +716,7 @@ func file_percutaneous_percutaneous_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_percutaneous_percutaneous_proto_rawDesc), len(file_percutaneous_percutaneous_proto_rawDesc)), NumEnums: 0, - NumMessages: 8, + NumMessages: 11, NumExtensions: 0, NumServices: 1, }, diff --git a/rpc/percutaneous/pb/percutaneous_grpc.pb.go b/rpc/percutaneous/pb/percutaneous_grpc.pb.go index b420a66..061d380 100644 --- a/rpc/percutaneous/pb/percutaneous_grpc.pb.go +++ b/rpc/percutaneous/pb/percutaneous_grpc.pb.go @@ -19,13 +19,18 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - Percutaneous_WifiCheck_FullMethodName = "/percutaneous.Percutaneous/WifiCheck" - Percutaneous_WifiSet_FullMethodName = "/percutaneous.Percutaneous/WifiSet" - Percutaneous_Use_FullMethodName = "/percutaneous.Percutaneous/Use" - 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_WifiCheck_FullMethodName = "/percutaneous.Percutaneous/WifiCheck" + Percutaneous_WifiSet_FullMethodName = "/percutaneous.Percutaneous/WifiSet" + Percutaneous_Use_FullMethodName = "/percutaneous.Percutaneous/Use" + 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. @@ -39,6 +44,16 @@ type PercutaneousClient interface { 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 { @@ -119,6 +134,65 @@ func (c *percutaneousClient) Weather(ctx context.Context, in *EmptyRequest, opts 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...) + 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_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. @@ -130,6 +204,16 @@ type PercutaneousServer interface { 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() } @@ -161,6 +245,21 @@ func (UnimplementedPercutaneousServer) UserItems(context.Context, *UserItemsRequ 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() {} @@ -308,6 +407,63 @@ func _Percutaneous_Weather_Handler(srv interface{}, ctx context.Context, dec fun 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}) +} + +// 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) @@ -343,7 +499,34 @@ var Percutaneous_ServiceDesc = grpc.ServiceDesc{ 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, + }, }, - Streams: []grpc.StreamDesc{}, Metadata: "percutaneous/percutaneous.proto", } diff --git a/rpc/percutaneous/percutaneous.pb b/rpc/percutaneous/percutaneous.pb index d074a09d6fbf3ceb6ae1c4dfa0f805527da44617..1bfe5d1e2a248bd947688f5b6597d3c4eb60e230 100644 GIT binary patch delta 548 zcmexS)LgOQu`$y;fz3~i3z)dPx%hnY(=+n|gC^UXsAvnxaB(K(l%|&C=a*ziFexx< za0Vd+l=Qg-98*$4i&Klh>Qx>1xOiN13rZ@3QVUB{i%XPrxcI}1gG-816LX`+$-$>08g dm>R&v6J89AjuaFlNivZS5mg|EPL8vb0sv+tyD$I% delta 66 zcmZp!_))asu`$zR?#)k)3z&rXxOiN13rZ@3QVUB{i%X2Yvu+kP|IQ@kBoM`?r(cj- Wlw4Ypn3tMgTAajIP?TSiZwvrbS{Qi% diff --git a/rpc/percutaneous/percutaneous.proto b/rpc/percutaneous/percutaneous.proto index 7c474f5..9d1c4f3 100644 --- a/rpc/percutaneous/percutaneous.proto +++ b/rpc/percutaneous/percutaneous.proto @@ -44,13 +44,22 @@ message UserItemsRequest { int32 size = 3; } +message LoginQRRequest { + string host_code = 1; + string bluetooth = 2; +} + +message AddUserQRRequest { + string host_code = 1; +} + message EmptyRequest { } -//message WsStreamFrame { -// bytes data = 1; -//} +message WsStreamRequest { + bytes raw_body = 1; // websocket原始业务报文，不做解析 +} service Percutaneous { rpc WifiCheck(WifiCheckRequest) returns (Response) { @@ -92,11 +101,6 @@ service Percutaneous { }; } - rpc Weather(EmptyRequest) returns (Response) { - option (google.api.http) = { - get: "/api/v3/percutaneous/weather" - body: "*" - }; - } -// rpc WsTunnelStream(stream WsStreamFrame) returns (stream Response); + // 添加用户 + rpc WsAddUserStream(stream WsStreamRequest) returns (stream WsStreamRequest); } diff --git a/services/percutaneous/internal/logic/weatherLogic.go b/services/percutaneous/internal/logic/weatherLogic.go deleted file mode 100644 index 0e3a4e0..0000000 --- a/services/percutaneous/internal/logic/weatherLogic.go +++ /dev/null @@ -1,30 +0,0 @@ -package logic - -import ( - "context" - - "lone-services/rpc/percutaneous/pb" - "lone-services/services/percutaneous/internal/svc" - - "github.com/zeromicro/go-zero/core/logx" -) - -type WeatherLogic struct { - ctx context.Context - svcCtx *svc.ServiceContext - logx.Logger -} - -func NewWeatherLogic(ctx context.Context, svcCtx *svc.ServiceContext) *WeatherLogic { - return &WeatherLogic{ - ctx: ctx, - svcCtx: svcCtx, - Logger: logx.WithContext(ctx), - } -} - -func (l *WeatherLogic) Weather(in *percutaneous.EmptyRequest) (*percutaneous.Response, error) { - // todo: add your logic here and delete this line - - return &percutaneous.Response{}, nil -} diff --git a/services/percutaneous/internal/logic/wsAddUserStreamLogic.go b/services/percutaneous/internal/logic/wsAddUserStreamLogic.go new file mode 100644 index 0000000..5057c33 --- /dev/null +++ b/services/percutaneous/internal/logic/wsAddUserStreamLogic.go @@ -0,0 +1,78 @@ +package logic + +import ( + "context" + "lone-services/pkg/utils" + "time" + + "lone-services/rpc/percutaneous/pb" + "lone-services/services/percutaneous/internal/svc" + + "github.com/zeromicro/go-zero/core/logx" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type WsAddUserStreamLogic struct { + ctx context.Context + svcCtx *svc.ServiceContext + logx.Logger +} + +func NewWsAddUserStreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *WsAddUserStreamLogic { + return &WsAddUserStreamLogic{ + ctx: ctx, + svcCtx: svcCtx, + Logger: logx.WithContext(ctx), + } +} + +// 添加用户 +func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error { + l.Logger.Error("in coming ws tunnel stream") + ctx := stream.Context() + // 扫码超时30s + timeout := 30 * time.Second + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + for { + // 先判断上下文是否已经结束 + if err := ctx.Err(); err != nil { + if err == context.DeadlineExceeded { + return status.Errorf(codes.DeadlineExceeded, "扫码超时") + } + 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 + } +} diff --git a/services/percutaneous/internal/server/percutaneousServer.go b/services/percutaneous/internal/server/percutaneousServer.go index 877b437..681326a 100644 --- a/services/percutaneous/internal/server/percutaneousServer.go +++ b/services/percutaneous/internal/server/percutaneousServer.go @@ -57,3 +57,33 @@ func (s *PercutaneousServer) Weather(ctx context.Context, in *percutaneous.Empty 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/percutaneousclient/percutaneous.go b/services/percutaneous/percutaneousclient/percutaneous.go index 8e91fb5..17f616b 100644 --- a/services/percutaneous/percutaneousclient/percutaneous.go +++ b/services/percutaneous/percutaneousclient/percutaneous.go @@ -14,14 +14,17 @@ import ( ) type ( + AddUserQRRequest = percutaneous.AddUserQRRequest AddUserRequest = percutaneous.AddUserRequest EmptyRequest = percutaneous.EmptyRequest + LoginQRRequest = percutaneous.LoginQRRequest Response = percutaneous.Response UseRequest = percutaneous.UseRequest UsedLogRequest = percutaneous.UsedLogRequest UserItemsRequest = percutaneous.UserItemsRequest WifiCheckRequest = percutaneous.WifiCheckRequest WifiSetRequest = percutaneous.WifiSetRequest + WsStreamRequest = percutaneous.WsStreamRequest Percutaneous interface { WifiCheck(ctx context.Context, in *WifiCheckRequest, opts ...grpc.CallOption) (*Response, error) @@ -31,6 +34,16 @@ type ( 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 { @@ -78,3 +91,33 @@ func (m *defaultPercutaneous) Weather(ctx context.Context, in *EmptyRequest, opt 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...) +} diff --git a/services/sale/.gitignore b/services/sale/.gitignore index 1004e40..d03386d 100644 --- a/services/sale/.gitignore +++ b/services/sale/.gitignore @@ -1,3 +1,4 @@ services/sale/logs/ services/sale/run.toml -tmp \ No newline at end of file +tmp +run.toml \ No newline at end of file