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

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
}