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 6165a89..2a6364a 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,40 @@ func main() { } }() + // Gateway 仅处理普通接口 /api /admin，端口 c.Port=10000 gw := gateway.MustNewServer(c.GatewayConf, gateway.WithDialer(dialer.Nacos), - gateway.WithMiddleware(request.FormUploadMiddleware), - // 权鉴+穿透 gateway.WithMiddleware(request.UserReadMiddleware), gateway.WithMiddleware(response.Wrap), ) defer gw.Stop() - fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port) + rpcMgr := wsproxy.NewRpcClientManager(c.WsPercutaneousRpc, + zrpc.WithUnaryClientInterceptor(request.UserClientInterceptor), + zrpc.WithStreamClientInterceptor(request.UserStreamInterceptor), + ) + + wsRest := rest.MustNewServer(rest.RestConf{ + Host: c.Host, + Port: c.WsPort, + }) + defer wsRest.Stop() + + // 注册ws路由到独立rest服务 + wsRest.AddRoutes([]rest.Route{ + { + Method: http.MethodGet, + Path: "/ws/percutaneous/add/user", + Handler: wsClient.PercutaneousAddUser(rpcMgr), + }, + }) + + go func() { + logx.Info("WS rest server start on :", c.WsPort) + wsRest.Start() + }() + + fmt.Printf("Gateway server at %s:%d\n", c.Host, c.Port) gw.Start() } 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..d829025 --- /dev/null +++ b/bff/internal/wsClient/client.go @@ -0,0 +1,68 @@ +package wsClient + +import ( + "context" + "fmt" + "lone-services/bff/internal/request" + "lone-services/bff/internal/wsproxy" + percutaneous "lone-services/rpc/percutaneous/pb" + "net/http" + + "github.com/zeromicro/go-zero/core/logx" + "google.golang.org/grpc" + "google.golang.org/protobuf/reflect/protoreflect" +) + +func PercutaneousAddUser(rpcMgr *wsproxy.RpcClientManager) http.HandlerFunc { + return request.UserReadMiddleware(func(w http.ResponseWriter, r *http.Request) { + logx.Infof("BFF ws handler start, raw url=%s", r.URL.String()) + cli, ready := rpcMgr.GetClient() + if !ready { + logx.Error("BFF: rpc client not ready, return 503") + http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable) + return + } + logx.Info("BFF: rpc client ready") + + hostCode := r.URL.Query().Get("host_code") + logx.Infof("BFF: get host_code from query, val=%s", hostCode) + if hostCode == "" { + logx.Error("BFF: host_code is empty") + http.Error(w, "missing host_code query param", http.StatusBadRequest) + return + } + + proxyHandler := wsproxy.NewWsProxyHandler( + func(parentCtx context.Context) (grpc.ClientStream, error) { + logx.Info("BFF: enter grpc stream create callback") + streamCli := percutaneous.NewPercutaneousClient(cli.Conn()) + stream, err := streamCli.WsAddUserStream(parentCtx) + if err != nil { + logx.Errorf("BFF: create WsAddUserStream failed, err=%v", err) + return nil, err + } + logx.Info("BFF: grpc stream create SUCCESS") + + // =========核心新增：BFF自动发送第一条host_code消息给percutaneous========= + jsonStr := fmt.Sprintf(`{"host_code":"%s"}`, hostCode) + reqMsg := &percutaneous.WsStreamRequest{ + RawBody: []byte(jsonStr), + } + err = stream.SendMsg(reqMsg) + if err != nil { + logx.Errorf("BFF: auto send host_code msg failed, err=%v, json=%s", err, jsonStr) + return nil, err + } + logx.Infof("BFF: auto send host_code message success, content=%s", jsonStr) + // ===================================================================== + return stream, nil + }, + func() protoreflect.ProtoMessage { + return &percutaneous.WsStreamRequest{} + }, + ) + logx.Info("BFF: start run proxyHandler") + proxyHandler(w, r) + logx.Info("BFF: proxyHandler finished, ws stream closed") + }) +} 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/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 6b4517e..0231d65 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,16 +624,23 @@ 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\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/weatherB\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 @@ -507,7 +654,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 +663,31 @@ 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 + 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 } func init() { file_percutaneous_percutaneous_proto_init() } @@ -551,7 +701,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..fff8242 100644 --- a/rpc/percutaneous/pb/percutaneous_grpc.pb.go +++ b/rpc/percutaneous/pb/percutaneous_grpc.pb.go @@ -19,13 +19,13 @@ 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_WsAddUserStream_FullMethodName = "/percutaneous.Percutaneous/WsAddUserStream" ) // PercutaneousClient is the client API for Percutaneous service. @@ -38,7 +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) + // 添加用户 + WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) } type percutaneousClient struct { @@ -109,16 +110,19 @@ 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) { +func (c *percutaneousClient) WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[WsStreamRequest, WsStreamRequest], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(Response) - err := c.cc.Invoke(ctx, Percutaneous_Weather_FullMethodName, in, out, cOpts...) + stream, err := c.cc.NewStream(ctx, &Percutaneous_ServiceDesc.Streams[0], Percutaneous_WsAddUserStream_FullMethodName, cOpts...) if err != nil { return nil, err } - return out, nil + 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] + // PercutaneousServer is the server API for Percutaneous service. // All implementations must embed UnimplementedPercutaneousServer // for forward compatibility. @@ -129,7 +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) + // 添加用户 + WsAddUserStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error mustEmbedUnimplementedPercutaneousServer() } @@ -158,8 +163,8 @@ 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) WsAddUserStream(grpc.BidiStreamingServer[WsStreamRequest, WsStreamRequest]) error { + return status.Error(codes.Unimplemented, "method WsAddUserStream not implemented") } func (UnimplementedPercutaneousServer) mustEmbedUnimplementedPercutaneousServer() {} func (UnimplementedPercutaneousServer) testEmbeddedByValue() {} @@ -290,24 +295,13 @@ 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_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] + // 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) @@ -339,11 +333,14 @@ var Percutaneous_ServiceDesc = grpc.ServiceDesc{ MethodName: "UserItems", Handler: _Percutaneous_UserItems_Handler, }, + }, + Streams: []grpc.StreamDesc{ { - MethodName: "Weather", - Handler: _Percutaneous_Weather_Handler, + StreamName: "WsAddUserStream", + Handler: _Percutaneous_WsAddUserStream_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 d074a09..758dd31 100644 Binary files a/rpc/percutaneous/percutaneous.pb and b/rpc/percutaneous/percutaneous.pb differ 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/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/etc/percutaneous.yaml b/services/percutaneous/etc/percutaneous.yaml deleted file mode 100644 index 2c526ea..0000000 --- a/services/percutaneous/etc/percutaneous.yaml +++ /dev/null @@ -1,7 +0,0 @@ -Nacos: - Hosts: - - host.docker.internal:8848 - NamespaceId: test - Group: LONE_SERVICES - RegisterIP: percutaneous - ConfigID: percutaneous \ No newline at end of file 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/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..fccf268 --- /dev/null +++ b/services/percutaneous/internal/logic/wsAddUserStreamLogic.go @@ -0,0 +1,132 @@ +package logic + +import ( + "context" + "lone-services/pkg/utils" + user "lone-services/rpc/user/pb" + "strconv" + "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 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, + svcCtx: svcCtx, + Logger: logx.WithContext(ctx), + } +} + +// 添加用户 +func (l *WsAddUserStreamLogic) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error { + l.Logger.Info("in coming ws tunnel stream") + ctx := stream.Context() + timeout := time.Duration(utils.GetConfigInt("ws.out_time")) * time.Second + ctx, cancel := context.WithTimeout(ctx, timeout) + defer func() { + cancel() + }() + + // 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 + } + 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 877b437..d62d2cc 100644 --- a/services/percutaneous/internal/server/percutaneousServer.go +++ b/services/percutaneous/internal/server/percutaneousServer.go @@ -53,7 +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) WsAddUserStream(stream percutaneous.Percutaneous_WsAddUserStreamServer) error { + l := logic.NewWsAddUserStreamLogic(stream.Context(), s.svcCtx) + return l.WsAddUserStream(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 8e91fb5..fbdc901 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) @@ -30,7 +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) + // 添加用户 + WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsAddUserStreamClient, error) } defaultPercutaneous struct { @@ -74,7 +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) { +// 添加用户 +func (m *defaultPercutaneous) WsAddUserStream(ctx context.Context, opts ...grpc.CallOption) (percutaneous.Percutaneous_WsAddUserStreamClient, error) { client := percutaneous.NewPercutaneousClient(m.cli.Conn()) - return client.Weather(ctx, in, opts...) + return client.WsAddUserStream(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