Commit ef02152e authored by hujiebin's avatar hujiebin

feat:broadcast 有话

parent 71bff772
...@@ -2,8 +2,10 @@ package main ...@@ -2,8 +2,10 @@ package main
import ( import (
"context" "context"
"errors"
"flag" "flag"
"fmt" "fmt"
"github.com/golang/protobuf/proto"
"gorm.io/gorm/schema" "gorm.io/gorm/schema"
"net" "net"
"net/url" "net/url"
...@@ -196,7 +198,7 @@ func (s *server) Multicast(ctx context.Context, in *userCenter.MulticastMessage) ...@@ -196,7 +198,7 @@ func (s *server) Multicast(ctx context.Context, in *userCenter.MulticastMessage)
return &userCenter.MulticastMessageRsp{FailedUids: failed}, nil return &userCenter.MulticastMessageRsp{FailedUids: failed}, nil
} }
func (s *server) Broadcast(ctx context.Context, in *userCenter.BroadcastMessage) (*userCenter.BroadcastMessageRsp, error) { func (s *server) BroadcastOld(ctx context.Context, in *userCenter.BroadcastMessage) (*userCenter.BroadcastMessageRsp, error) {
//mylogrus.MyLog.Infof("Broadcasting msgType = %d, size = %d\n", in.MsgType, len(in.PayLoad)) //mylogrus.MyLog.Infof("Broadcasting msgType = %d, size = %d\n", in.MsgType, len(in.PayLoad))
failed := []uint64{} failed := []uint64{}
...@@ -249,6 +251,32 @@ func (s *server) Broadcast(ctx context.Context, in *userCenter.BroadcastMessage) ...@@ -249,6 +251,32 @@ func (s *server) Broadcast(ctx context.Context, in *userCenter.BroadcastMessage)
return &userCenter.BroadcastMessageRsp{FailedUids: failed}, nil return &userCenter.BroadcastMessageRsp{FailedUids: failed}, nil
} }
func (s *server) Broadcast(ctx context.Context, in *userCenter.BroadcastMessage) (*userCenter.BroadcastMessageRsp, error) {
redisKey := "service:userSocket"
ipPorts, err := rdbCluster.ZRangeByScore(context.Background(), redisKey, &redis.ZRangeBy{
Min: fmt.Sprintf("%d", time.Now().Add(-time.Second*15).Unix()), // 3倍心跳
Max: "+inf",
}).Result()
if err != nil {
failMsg := fmt.Sprintf("get service fail,svc:%v,err:%v", "userSocket", err)
mylogrus.MyLog.Errorf(failMsg)
_ = dingding.SendDingRobot(dingding.ROBOTWEBHOOK, failMsg, true)
return nil, err
}
if len(ipPorts) <= 0 {
failMsg := fmt.Sprintf("get service empty,svc:%v,err:%v", "userSocket", err)
mylogrus.MyLog.Errorf(failMsg)
_ = dingding.SendDingRobot(dingding.ROBOTWEBHOOK, failMsg, true)
return nil, errors.New(failMsg)
}
data, _ := proto.Marshal(in)
for _, ip := range ipPorts {
queue := "broadcast:" + ip
rdbCluster.RPush(context.Background(), queue, data)
}
return &userCenter.BroadcastMessageRsp{FailedUids: nil}, nil
}
func (s *server) Areacast(ctx context.Context, in *userCenter.AreaMessage) (*userCenter.AreaMessageRsp, error) { func (s *server) Areacast(ctx context.Context, in *userCenter.AreaMessage) (*userCenter.AreaMessageRsp, error) {
var failed []uint64 var failed []uint64
terminals := termManager.GetAll() terminals := termManager.GetAll()
...@@ -558,11 +586,13 @@ type HiloConfigs struct { ...@@ -558,11 +586,13 @@ type HiloConfigs struct {
Value string Value string
} }
var rdbCluster *redis.Client
func main() { func main() {
flag.Parse() flag.Parse()
// init redis cluster // init redis cluster
rdbCluster := redis.NewClient(&redis.Options{ rdbCluster = redis.NewClient(&redis.Options{
Addr: appConfig.GetConfigRedis().REDIS_CLUSTER_HOST, Addr: appConfig.GetConfigRedis().REDIS_CLUSTER_HOST,
Password: appConfig.GetConfigRedis().REDIS_CLUSTER_PASSWORD, Password: appConfig.GetConfigRedis().REDIS_CLUSTER_PASSWORD,
}) })
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment