blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

rpc register ip

Browse Source
This commit is contained in:
skiffer-git 2022-05-07 17:05:05 +08:00
parent fed369ad5d
commit dca16e83d3
19 changed files with 307 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/open_im_api/main.go
View File

@ -178,7 +178,7 @@ func main() {
} }
go apiThird.MinioInit() go apiThird.MinioInit()
ginPort := flag.Int("port", 10002, "get ginServerPort from cmd,default 10000 as port") ginPort := flag.Int("port", 10002, "get ginServerPort from cmd,default 10002 as port")
flag.Parse() flag.Parse()
fmt.Println("start api server, port: ", *ginPort) fmt.Println("start api server, port: ", *ginPort)
err := r.Run(":" + strconv.Itoa(*ginPort)) err := r.Run(":" + strconv.Itoa(*ginPort))

5
config/config.yaml
View File

@ -59,6 +59,11 @@ kafka:
#如果是单机模式，用0.0.0.0或者不填，默认即可 #如果是单机模式，用0.0.0.0或者不填，默认即可
serverip: 0.0.0.0 serverip: 0.0.0.0
rpcRegisterIP: 127.0.0.1
listenIP: 0.0.0.0
# endpoints 内部组件间访问的端点host名称，访问时，可以内部直接访问 host:port 来访问 # endpoints 内部组件间访问的端点host名称，访问时，可以内部直接访问 host:port 来访问
endpoints: endpoints:
api: openim_api api: openim_api

29
internal/msg_gateway/gate/rpc_server.go
View File

@ -11,9 +11,9 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/gob" "encoding/gob"
"fmt"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"net" "net"
"strconv"
"strings" "strings"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
@ -34,24 +34,37 @@ func (r *RPCServer) onInit(rpcPort int) {
r.etcdAddr = config.Config.Etcd.EtcdAddr r.etcdAddr = config.Config.Etcd.EtcdAddr
} }
func (r *RPCServer) run() { func (r *RPCServer) run() {
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + utils.IntToString(r.rpcPort) if config.Config.ListenIP == "" {
listener, err := net.Listen("tcp", registerAddress) listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(r.rpcPort)
listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.ErrorByArgs(fmt.Sprintf("fail to listening consumer, err:%v\n", err)) log.Error("", "fail to listening consumer failed ", err.Error(), address)
return return
} }
defer listener.Close() defer listener.Close()
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
pbRelay.RegisterOnlineMessageRelayServiceServer(srv, r) pbRelay.RegisterOnlineMessageRelayServiceServer(srv, r)
err = getcdv3.RegisterEtcd4Unique(r.etcdSchema, strings.Join(r.etcdAddr, ","), ip, r.rpcPort, r.rpcRegisterName, 10)
rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil { if err != nil {
log.ErrorByKv("register push message rpc to etcd err", "", "err", err.Error()) log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd4Unique(r.etcdSchema, strings.Join(r.etcdAddr, ","), rpcRegisterIP, r.rpcPort, r.rpcRegisterName, 10)
if err != nil {
log.Error("", "register push message rpc to etcd err", "", "err", err.Error(), r.etcdSchema, strings.Join(r.etcdAddr, ","), rpcRegisterIP, r.rpcPort, r.rpcRegisterName)
} }
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.ErrorByKv("push message rpc listening err", "", "err", err.Error()) log.Error("", "push message rpc listening err", "", "err", err.Error())
return return
} }
} }

29
internal/push/logic/push_rpc_server.go
View File

@ -9,6 +9,7 @@ import (
"context" "context"
"google.golang.org/grpc" "google.golang.org/grpc"
"net" "net"
"strconv"
"strings" "strings"
) )
@ -26,24 +27,38 @@ func (r *RPCServer) Init(rpcPort int) {
r.etcdAddr = config.Config.Etcd.EtcdAddr r.etcdAddr = config.Config.Etcd.EtcdAddr
} }
func (r *RPCServer) run() { func (r *RPCServer) run() {
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + utils.IntToString(r.rpcPort) if config.Config.ListenIP == "" {
listener, err := net.Listen("tcp", registerAddress) listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(r.rpcPort)
listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.ErrorByKv("push module rpc listening port err", "", "err", err.Error()) log.Error("", "push module rpc listening port err", err.Error(), address)
return return
} }
defer listener.Close() defer listener.Close()
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
pbPush.RegisterPushMsgServiceServer(srv, r) pbPush.RegisterPushMsgServiceServer(srv, r)
err = getcdv3.RegisterEtcd(r.etcdSchema, strings.Join(r.etcdAddr, ","), ip, r.rpcPort, r.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil { if err != nil {
log.ErrorByKv("register push module rpc to etcd err", "", "err", err.Error()) log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(r.etcdSchema, strings.Join(r.etcdAddr, ","), rpcRegisterIP, r.rpcPort, r.rpcRegisterName, 10)
if err != nil {
log.Error("", "register push module rpc to etcd err", err.Error(), r.etcdSchema, strings.Join(r.etcdAddr, ","), rpcRegisterIP, r.rpcPort, r.rpcRegisterName)
} }
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.ErrorByKv("push module rpc start err", "", "err", err.Error()) log.Error("", "push module rpc start err", err.Error())
return return
} }
} }

25
internal/rpc/admin_cms/admin_cms.go
View File

@ -35,22 +35,35 @@ func NewAdminCMSServer(port int) *adminCMSServer {
func (s *adminCMSServer) Run() { func (s *adminCMSServer) Run() {
log.NewInfo("0", "AdminCMS rpc start ") log.NewInfo("0", "AdminCMS rpc start ")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbAdminCMS.RegisterAdminCMSServer(srv, s) pbAdminCMS.RegisterAdminCMSServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error()) log.NewError("0", "RegisterEtcd failed ", err.Error())
return return

22
internal/rpc/auth/auth.go
View File

@ -79,7 +79,13 @@ func (rpc *rpcAuth) Run() {
operationID := utils.OperationIDGenerator() operationID := utils.OperationIDGenerator()
log.NewInfo(operationID, "rpc auth start...") log.NewInfo(operationID, "rpc auth start...")
address := utils.ServerIP + ":" + strconv.Itoa(rpc.rpcPort) listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(rpc.rpcPort)
listener, err := net.Listen("tcp", address) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError(operationID, "listen network failed ", err.Error(), address) log.NewError(operationID, "listen network failed ", err.Error(), address)
@ -92,13 +98,21 @@ func (rpc *rpcAuth) Run() {
//service registers with etcd //service registers with etcd
pbAuth.RegisterAuthServer(srv, rpc) pbAuth.RegisterAuthServer(srv, rpc)
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError(operationID, "RegisterEtcd failed ", err.Error(), log.NewError(operationID, "RegisterEtcd failed ", err.Error(),
rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName) rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
return return
} }
log.NewInfo(operationID, "RegisterAuthServer ok ", rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName) log.NewInfo(operationID, "RegisterAuthServer ok ", rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.NewError(operationID, "Serve failed ", err.Error()) log.NewError(operationID, "Serve failed ", err.Error())

26
internal/rpc/cache/cache.go vendored
View File

@ -37,15 +37,20 @@ func NewCacheServer(port int) *cacheServer {
func (s *cacheServer) Run() { func (s *cacheServer) Run() {
log.NewInfo("0", "cacheServer rpc start ") log.NewInfo("0", "cacheServer rpc start ")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
@ -59,7 +64,16 @@ func (s *cacheServer) Run() {
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
pbCache.RegisterCacheServer(srv, s) pbCache.RegisterCacheServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10)
rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error()) log.NewError("0", "RegisterEtcd failed ", err.Error())
return return

22
internal/rpc/conversation/conversaion.go
View File

@ -138,7 +138,14 @@ func NewRpcConversationServer(port int) *rpcConversation {
func (rpc *rpcConversation) Run() { func (rpc *rpcConversation) Run() {
log.NewInfo("0", "rpc conversation start...") log.NewInfo("0", "rpc conversation start...")
address := utils.ServerIP + ":" + strconv.Itoa(rpc.rpcPort) listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(rpc.rpcPort)
listener, err := net.Listen("tcp", address) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "listen network failed ", err.Error(), address) log.NewError("0", "listen network failed ", err.Error(), address)
@ -151,13 +158,20 @@ func (rpc *rpcConversation) Run() {
//service registers with etcd //service registers with etcd
pbConversation.RegisterConversationServer(srv, rpc) pbConversation.RegisterConversationServer(srv, rpc)
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error(), log.NewError("0", "RegisterEtcd failed ", err.Error(),
rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName) rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
return return
} }
log.NewInfo("0", "RegisterConversationServer ok ", rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName) log.NewInfo("0", "RegisterConversationServer ok ", rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName)
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.NewError("0", "Serve failed ", err.Error()) log.NewError("0", "Serve failed ", err.Error())

28
internal/rpc/friend/firend.go
View File

@ -42,24 +42,38 @@ func NewFriendServer(port int) *friendServer {
func (s *friendServer) Run() { func (s *friendServer) Run() {
log.NewInfo("0", "friendServer run...") log.NewInfo("0", "friendServer run...")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen ok ", registerAddress) log.NewInfo("0", "listen ok ", address)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//User friend related services register to etcd //User friend related services register to etcd
pbFriend.RegisterFriendServer(srv, s) pbFriend.RegisterFriendServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error(), s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName) log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error(), s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName)
return return
} }
err = srv.Serve(listener) err = srv.Serve(listener)

25
internal/rpc/group/group.go
View File

@ -44,22 +44,35 @@ func NewGroupServer(port int) *groupServer {
func (s *groupServer) Run() { func (s *groupServer) Run() {
log.NewInfo("", "group rpc start ") log.NewInfo("", "group rpc start ")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("", "Listen failed ", err.Error(), registerAddress) log.NewError("", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("", "listen network success, ", registerAddress, listener) log.NewInfo("", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbGroup.RegisterGroupServer(srv, s) pbGroup.RegisterGroupServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10)
rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("", "RegisterEtcd failed ", err.Error()) log.NewError("", "RegisterEtcd failed ", err.Error())
return return

28
internal/rpc/message_cms/message_cms.go
View File

@ -6,6 +6,7 @@ import (
"Open_IM/pkg/common/db" "Open_IM/pkg/common/db"
errors "Open_IM/pkg/common/http" errors "Open_IM/pkg/common/http"
"context" "context"
"strconv"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
@ -17,7 +18,6 @@ import (
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"net" "net"
"strconv"
"strings" "strings"
"google.golang.org/grpc" "google.golang.org/grpc"
@ -42,22 +42,36 @@ func NewMessageCMSServer(port int) *messageCMSServer {
func (s *messageCMSServer) Run() { func (s *messageCMSServer) Run() {
log.NewInfo("0", "messageCMS rpc start ") log.NewInfo("0", "messageCMS rpc start ")
ip := utils.ServerIP
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbMessageCMS.RegisterMessageCMSServer(srv, s) pbMessageCMS.RegisterMessageCMSServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error()) log.NewError("0", "RegisterEtcd failed ", err.Error())
return return

33
internal/rpc/msg/rpcChat.go
View File

@ -35,33 +35,42 @@ func NewRpcChatServer(port int) *rpcChat {
} }
func (rpc *rpcChat) Run() { func (rpc *rpcChat) Run() {
log.Info("", "", "rpc get_token init...") log.Info("", "rpcChat init...")
listenIP := ""
address := utils.ServerIP + ":" + strconv.Itoa(rpc.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(rpc.rpcPort)
listener, err := net.Listen("tcp", address) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.Error("", "", "listen network failed, err = %s, address = %s", err.Error(), address) log.Error("", "listen network failed ", err.Error(), address)
return return
} }
log.Info("", "", "listen network success, address = ", address) log.Info("", "listen network success, address ", address)
//grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//service registers with etcd rpcRegisterIP := ""
pbChat.RegisterChatServer(srv, rpc) pbChat.RegisterChatServer(srv, rpc)
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName, 10) if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil { if err != nil {
log.Error("", "", "register rpc get_token to etcd failed, err = %s", err.Error()) log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), rpcRegisterIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
if err != nil {
log.Error("", "register rpcChat to etcd failed ", err.Error())
return return
} }
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.Info("", "", "rpc get_token fail, err = %s", err.Error()) log.Error("", "rpc rpcChat failed ", err.Error())
return return
} }
log.Info("", "", "rpc get_token init success") log.Info("", "rpc rpcChat init success")
} }

25
internal/rpc/office/office.go
View File

@ -39,22 +39,35 @@ func NewOfficeServer(port int) *officeServer {
func (s *officeServer) Run() { func (s *officeServer) Run() {
log.NewInfo("0", "officeServer rpc start ") log.NewInfo("0", "officeServer rpc start ")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbOffice.RegisterOfficeServiceServer(srv, s) pbOffice.RegisterOfficeServiceServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error()) log.NewError("0", "RegisterEtcd failed ", err.Error())
return return

26
internal/rpc/organization/organization.go
View File

@ -40,27 +40,39 @@ func NewServer(port int) *organizationServer {
func (s *organizationServer) Run() { func (s *organizationServer) Run() {
log.NewInfo("", "organization rpc start ") log.NewInfo("", "organization rpc start ")
ip := utils.ServerIP listenIP := ""
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("", "Listen failed ", err.Error(), registerAddress) log.NewError("", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("", "listen network success, ", registerAddress, listener) log.NewInfo("", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
rpc.RegisterOrganizationServer(srv, s) rpc.RegisterOrganizationServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("", "RegisterEtcd failed ", err.Error()) log.NewError("", "RegisterEtcd failed ", err.Error())
return return
} }
log.NewInfo("", "organization rpc RegisterEtcd success", ip, s.rpcPort, s.rpcRegisterName, 10) log.NewInfo("", "organization rpc RegisterEtcd success", rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
err = srv.Serve(listener) err = srv.Serve(listener)
if err != nil { if err != nil {
log.NewError("", "Serve failed ", err.Error()) log.NewError("", "Serve failed ", err.Error())

28
internal/rpc/statistics/statistics.go
View File

@ -4,6 +4,7 @@ import (
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"context" "context"
"strconv"
"sync" "sync"
"time" "time"
@ -21,7 +22,6 @@ import (
//"context" //"context"
errors "Open_IM/pkg/common/http" errors "Open_IM/pkg/common/http"
"net" "net"
"strconv"
"strings" "strings"
"google.golang.org/grpc" "google.golang.org/grpc"
@ -46,22 +46,36 @@ func NewStatisticsServer(port int) *statisticsServer {
func (s *statisticsServer) Run() { func (s *statisticsServer) Run() {
log.NewInfo("0", "Statistics rpc start ") log.NewInfo("0", "Statistics rpc start ")
ip := utils.ServerIP
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort) listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "Listen failed ", err.Error(), registerAddress) log.NewError("0", "Listen failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, ", registerAddress, listener) log.NewInfo("0", "listen network success, ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbStatistics.RegisterUserServer(srv, s) pbStatistics.RegisterUserServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error()) log.NewError("0", "RegisterEtcd failed ", err.Error())
return return

29
internal/rpc/user/user.go
View File

@ -42,26 +42,39 @@ func NewUserServer(port int) *userServer {
} }
func (s *userServer) Run() { func (s *userServer) Run() {
log.NewInfo("0", "", "rpc user start...") log.NewInfo("0", "rpc user start...")
listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
ip := utils.ServerIP
registerAddress := ip + ":" + strconv.Itoa(s.rpcPort)
//listener network //listener network
listener, err := net.Listen("tcp", registerAddress) listener, err := net.Listen("tcp", address)
if err != nil { if err != nil {
log.NewError("0", "listen network failed ", err.Error(), registerAddress) log.NewError("0", "listen network failed ", err.Error(), address)
return return
} }
log.NewInfo("0", "listen network success, address ", registerAddress, listener) log.NewInfo("0", "listen network success, address ", address, listener)
defer listener.Close() defer listener.Close()
//grpc server //grpc server
srv := grpc.NewServer() srv := grpc.NewServer()
defer srv.GracefulStop() defer srv.GracefulStop()
//Service registers with etcd //Service registers with etcd
pbUser.RegisterUserServer(srv, s) pbUser.RegisterUserServer(srv, s)
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName, 10) rpcRegisterIP := ""
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil { if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error(), s.etcdSchema, strings.Join(s.etcdAddr, ","), ip, s.rpcPort, s.rpcRegisterName) log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = getcdv3.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10)
if err != nil {
log.NewError("0", "RegisterEtcd failed ", err.Error(), s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName)
return return
} }
err = srv.Serve(listener) err = srv.Serve(listener)

12
internal/utils/get_server_ip_test.go
View File

@ -1,13 +1 @@
package utils package utils
import (
"Open_IM/pkg/utils"
"net"
"testing"
)
func TestServerIP(t *testing.T) {
if net.ParseIP(utils.ServerIP) == nil {
t.Fail()
}
}

4
pkg/common/config/config.go
View File

@ -25,6 +25,10 @@ type callBackConfig struct {
type config struct { type config struct {
ServerIP string `yaml:"serverip"` ServerIP string `yaml:"serverip"`
RpcRegisterIP string `yaml:"rpcRegisterIP"`
ListenIP string `yaml:"listenIP"`
ServerVersion string `yaml:"serverversion"` ServerVersion string `yaml:"serverversion"`
Api struct { Api struct {
GinPort []int `yaml:"openImApiPort"` GinPort []int `yaml:"openImApiPort"`

30
pkg/utils/get_server_ip.go
View File

@ -1,26 +1,28 @@
package utils package utils
import ( import (
"Open_IM/pkg/common/config" "errors"
"fmt"
"net" "net"
) )
var ServerIP = "" var ServerIP = ""
func init() { func GetLocalIP() (string, error) {
//fixme In the configuration file, ip takes precedence, if not, get the valid network card ip of the machine addrs, err := net.InterfaceAddrs()
if config.Config.ServerIP != "" {
ServerIP = config.Config.ServerIP
return
}
// see https://gist.github.com/jniltinho/9787946#gistcomment-3019898
conn, err := net.Dial("udp", "8.8.8.8:80")
if err != nil { if err != nil {
panic(err.Error())
return "", err
}
for _, address := range addrs {
if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
fmt.Println(ipnet.IP.String())
return ipnet.IP.String(), nil
}
}
} }
defer conn.Close() return "", errors.New("no ip")
localAddr := conn.LocalAddr().(*net.UDPAddr)
ServerIP = localAddr.IP.String()
} }