lone-services/bff/internal/wsClient/client.go

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) {
		cli, ready := rpcMgr.GetClient()
		if !ready {
			logx.Error("BFF: rpc client not ready, return 503")
			http.Error(w, "rpc service unavailable", http.StatusServiceUnavailable)
			return
		}

		hostCode := r.URL.Query().Get("host_code")
		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) {
				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
				}

				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
				}
				return stream, nil
			},
			func() protoreflect.ProtoMessage {
				return &percutaneous.WsStreamRequest{}
			},
		)
		proxyHandler(w, r)
	})
}