feat 初次提交

internal/grpc_server/server/server.go

@@ -0,0 +1,74 @@
+package server
+
+import (
+	"common/log"
+	"common/proto/sc/sc_pb"
+	"common/proto/ss/grpc_pb"
+	"context"
+	"google.golang.org/protobuf/proto"
+	instance2 "scene/internal/instance"
+	"sync"
+)
+
+func (s *Server) Enter(ctx context.Context, req *grpc_pb.EnterReq) (*grpc_pb.EnterResp, error) {
+	var i *instance2.Instance
+	if len(instance2.Mgr.GetAll()) == 0 {
+		i = instance2.NewScene(s.SID, req.InstanceID)
+		i.Start(s.EtcdTTL)
+	} else {
+		for _, v := range instance2.Mgr.GetAll() {
+			i = v
+			break
+		}
+	}
+	i.EventIn <- req
+
+	payload, _ := proto.Marshal(&sc_pb.S2C_EnterInstance{
+		Info: &sc_pb.PositionInfo{
+			USN: req.USN,
+			X:   1,
+			Y:   1,
+		},
+	})
+	return &grpc_pb.EnterResp{
+		SceneSID:  s.SID,
+		UniqueNo:  i.UniqueNo,
+		MessageID: int32(sc_pb.MessageID_MESSAGE_ID_ENTER_INSTANCE),
+		Payload:   payload,
+	}, nil
+}
+
+func (s *Server) Leave(ctx context.Context, req *grpc_pb.LeaveReq) (*grpc_pb.LeaveResp, error) {
+	if i := instance2.Mgr.GetByUniqueNo(req.UniqueNo); i != nil {
+		i.EventIn <- req
+	}
+	return &grpc_pb.LeaveResp{}, nil
+}
+
+func (s *Server) Action(server grpc_pb.Scene_ActionServer) error {
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		defer func() {
+			if err := recover(); err != nil {
+				log.Errorf("Action panic: %v", err)
+			}
+		}()
+		for {
+			if args, err := server.Recv(); err != nil {
+				return
+			} else {
+				if ins := instance2.Mgr.GetByUniqueNo(args.UniqueNo); ins != nil {
+					select {
+					case ins.EventIn <- args:
+					default:
+						log.Warnf("instance event in full: %v, %v", ins.InstanceID, ins.UniqueNo)
+					}
+				}
+			}
+		}
+	}()
+	wg.Wait()
+	return server.SendAndClose(&grpc_pb.ActionResp{})
+}

internal/grpc_server/server/server_init.go

@@ -0,0 +1,32 @@
+package server
+
+import (
+	"common/discover/common"
+	"common/net/grpc/service"
+	"common/proto/ss/grpc_pb"
+	"google.golang.org/grpc"
+)
+
+type Server struct {
+	grpc_pb.UnimplementedSceneServer
+	service.Base
+}
+
+func NewServer(ttl int64) *Server {
+	s := &Server{
+		Base: service.Base{
+			Target:  common.KeyDiscoverScene,
+			EtcdTTL: ttl,
+		},
+	}
+	s.Base.OnInit = s.OnInit
+	s.Base.OnClose = s.OnClose
+	return s
+}
+
+func (s *Server) OnInit(serve *grpc.Server) {
+	grpc_pb.RegisterSceneServer(serve, s)
+}
+
+func (s *Server) OnClose() {
+}

internal/grpc_server/stream_client/gateway.go

@@ -0,0 +1,65 @@
+package stream_client
+
+import (
+	"common/log"
+	"common/net/grpc/service"
+	"context"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+)
+
+var gatewayServerM map[int64]map[GatewayFun]grpc.ClientStream // map[sid]map[方法名]流连接
+
+type GatewayFun int
+
+const (
+	FunToClient GatewayFun = iota
+)
+
+func init() {
+	gatewayServerM = make(map[int64]map[GatewayFun]grpc.ClientStream)
+}
+
+func FindGatewayBySID(sid int64, fun GatewayFun) (grpc.ClientStream, error) {
+	g := gatewayServerM[sid]
+	if g == nil {
+		g = make(map[GatewayFun]grpc.ClientStream)
+		gatewayServerM[sid] = g
+	}
+	gatewayLink := g[fun]
+	if gatewayLink == nil {
+		gatewayClient, err := service.GatewayNewClient(sid)
+		if err != nil {
+			log.Errorf("cannot find gatewayClient: %v", err)
+			return nil, err
+		}
+		var link grpc.ClientStream
+		switch fun {
+		case FunToClient:
+			link, err = gatewayClient.ToClient(context.Background())
+		}
+		if err != nil {
+			log.Errorf("FindGatewayBySID %v err: %v, sid: %v", fun, err, sid)
+			return nil, err
+		}
+		g[fun] = link
+		gatewayLink = link
+	}
+	return gatewayLink, nil
+}
+
+func SendMessageToGateway(sid int64, fun GatewayFun, msg proto.Message, re ...bool) error {
+	stream, err := FindGatewayBySID(sid, fun)
+	if err != nil {
+		return err
+	}
+	if err = stream.SendMsg(msg); err != nil {
+		if re == nil || !re[0] {
+			_ = stream.CloseSend()
+			delete(gatewayServerM[sid], fun)
+			return SendMessageToGateway(sid, fun, msg, true)
+		}
+		return err
+	}
+	return nil
+}