script

internal/msggateway/hub_server.go

@@ -1,66 +1,31 @@
 package msggateway
 
 import (
-	"OpenIM/internal/common/network"
+	"OpenIM/internal/startrpc"
 	"OpenIM/pkg/common/config"
 	"OpenIM/pkg/common/constant"
-	"OpenIM/pkg/common/mw"
 	"OpenIM/pkg/common/prome"
 	"OpenIM/pkg/common/tokenverify"
+	"OpenIM/pkg/discoveryregistry"
 	"OpenIM/pkg/errs"
 	"OpenIM/pkg/proto/msggateway"
 	"OpenIM/pkg/utils"
 	"context"
-	"fmt"
-	"github.com/OpenIMSDK/openKeeper"
-	grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"google.golang.org/grpc"
-	"net"
 )
 
-func (s *Server) Start() error {
-	zkClient, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema, 10, "", "")
-	if err != nil {
-		return err
-	}
-	defer zkClient.Close()
-	registerIP, err := network.GetRpcRegisterIP(config.Config.RpcRegisterIP)
-	if err != nil {
-		return err
-	}
-	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, s.rpcPort))
-	if err != nil {
-		panic("listening err:" + err.Error())
-	}
-	defer listener.Close()
-	var options []grpc.ServerOption
-	options = append(options, mw.GrpcServer()) // ctx 中间件
-	if config.Config.Prometheus.Enable {
-		prome.NewGrpcRequestCounter()
-		prome.NewGrpcRequestFailedCounter()
-		prome.NewGrpcRequestSuccessCounter()
-		options = append(options, []grpc.ServerOption{
-			//grpc.UnaryInterceptor(prome.UnaryServerInterceptorPrometheus),
-			grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
-			grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
-		}...)
-	}
-	srv := grpc.NewServer(options...)
-	defer srv.GracefulStop()
-	msggateway.RegisterMsgGatewayServer(srv, s)
-	err = zkClient.Register("", registerIP, s.rpcPort)
-	if err != nil {
-		return err
-	}
-	err = srv.Serve(listener)
-	if err != nil {
-		return err
-	}
+func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
+	msggateway.RegisterMsgGatewayServer(server, &Server{})
 	return nil
 }
 
+func (s *Server) Start() error {
+	return startrpc.Start(s.rpcPort, config.Config.RpcRegisterName.OpenImMessageGatewayName, s.prometheusPort, Start)
+}
+
 type Server struct {
 	rpcPort        int
+	prometheusPort int
 	LongConnServer LongConnServer
 	pushTerminal   []int
 	//rpcServer      *RpcServer

internal/msggateway/init.go

@@ -0,0 +1,30 @@
+package msggateway
+
+import (
+	"OpenIM/pkg/common/config"
+	"OpenIM/pkg/common/constant"
+	"OpenIM/pkg/common/log"
+	"fmt"
+	"sync"
+	"time"
+)
+
+func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	log.NewPrivateLog(constant.LogFileName)
+	fmt.Println("start rpc/msg_gateway server, port: ", rpcPort, wsPort, prometheusPort, ", OpenIM version: ", config.Version)
+	longServer, err := NewWsServer(
+		WithPort(wsPort),
+		WithMaxConnNum(int64(config.Config.LongConnSvr.WebsocketMaxConnNum)),
+		WithHandshakeTimeout(time.Duration(config.Config.LongConnSvr.WebsocketTimeOut)*time.Second),
+		WithMessageMaxMsgLength(config.Config.LongConnSvr.WebsocketMaxMsgLen))
+	if err != nil {
+		return err
+	}
+	hubServer := NewServer(rpcPort, longServer)
+	go hubServer.Start()
+	go hubServer.LongConnServer.Run()
+	wg.Wait()
+	return nil
+}

internal/rpc/auth/auth.go

@@ -9,7 +9,7 @@ import (
 	"OpenIM/pkg/common/log"
 	"OpenIM/pkg/common/tokenverify"
 	"OpenIM/pkg/common/tracelog"
-	discoveryRegistry "OpenIM/pkg/discoveryregistry"
+	"OpenIM/pkg/discoveryregistry"
 	"OpenIM/pkg/errs"
 	pbAuth "OpenIM/pkg/proto/auth"
 	"OpenIM/pkg/proto/msggateway"
@@ -21,10 +21,10 @@ import (
 type authServer struct {
 	authDatabase   controller.AuthDatabase
 	userCheck      *check.UserCheck
-	RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
+	RegisterCenter discoveryregistry.SvcDiscoveryRegistry
 }
 
-func Start(client discoveryRegistry.SvcDiscoveryRegistry, server *grpc.Server) error {
+func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
 	rdb, err := cache.NewRedis()
 	if err != nil {
 		return err

internal/startrpc/start.go

@@ -16,13 +16,11 @@ import (
 	"net"
 )
 
-func start(rpcPort int, rpcRegisterName string, prometheusPorts int, rpcFn func(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error, options []grpc.ServerOption) error {
-	flagRpcPort := flag.Int("port", rpcPort, "get RpcGroupPort from cmd,default 16000 as port")
-	flagPrometheusPort := flag.Int("prometheus_port", prometheusPorts, "groupPrometheusPort default listen port")
+func start(rpcPort int, rpcRegisterName string, prometheusPort int, rpcFn func(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error, options []grpc.ServerOption) error {
 	flag.Parse()
-	fmt.Println("start group rpc server, port: ", *flagRpcPort, ", OpenIM version: ", config.Version)
+	fmt.Println("start group rpc server, port: ", rpcPort, ", OpenIM version: ", config.Version)
 	log.NewPrivateLog(constant.LogFileName)
-	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, *flagRpcPort))
+	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, rpcPort))
 	if err != nil {
 		return err
 	}
@@ -49,12 +47,12 @@ func start(rpcPort int, rpcRegisterName string, prometheusPorts int, rpcFn func(
 	}
 	srv := grpc.NewServer(options...)
 	defer srv.GracefulStop()
-	err = zkClient.Register(rpcRegisterName, registerIP, *flagRpcPort)
+	err = zkClient.Register(rpcRegisterName, registerIP, rpcPort)
 	if err != nil {
 		return err
 	}
-	if config.Config.Prometheus.Enable {
-		err := prome.StartPrometheusSrv(*flagPrometheusPort)
+	if config.Config.Prometheus.Enable && prometheusPort != 0 {
+		err := prome.StartPrometheusSrv(prometheusPort)
 		if err != nil {
 			return err
 		}