diff --git a/cmd/main.go b/cmd/main.go index e43636398..3e1c30fcc 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -37,6 +37,8 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" + "github.com/openimsdk/open-im-server/v3/pkg/dbbuild" "github.com/openimsdk/open-im-server/v3/version" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/discovery/inprocess" @@ -458,16 +460,50 @@ func startRedisServerRegister(ctx context.Context, cfg *serverConfig, client dis if err != nil { return err } - addr := net.JoinHostPort(registerIP, strconv.Itoa(apiPort)) - inprocess.SetLocalTarget(addr) - timer := time.NewTimer(time.Second * 5) - defer timer.Stop() + const validTime = time.Second * 10 + dbb := dbbuild.NewBuilder(nil, &cfg.RedisConfig) + rdb, err := dbb.Redis(ctx) + if err != nil { + return err + } + gateway := redis.NewStandaloneGatewayRedis(rdb, validTime) + selfAddr := net.JoinHostPort(registerIP, strconv.Itoa(apiPort)) + inprocess.SetLocalTarget(selfAddr) + inprocess.SetBroadcastAddress(cfg.Share.Secret, func(ctx context.Context) ([]string, error) { + address, err := gateway.GetGatewayAddrs(ctx) + if err != nil { + return nil, err + } + notSelf := make([]string, 0, len(address)) + for _, addr := range address { + if addr != selfAddr { + notSelf = append(notSelf, addr) + } + } + return notSelf, nil + }) + register := func() { + ctx, cancel := context.WithTimeout(ctx, validTime/2) + defer cancel() + if err := gateway.RegisterGateway(ctx, selfAddr); err != nil { + log.ZWarn(ctx, "gateway register failed", err, "address", selfAddr) + } + } + timer := time.NewTimer(validTime / 2) + defer func() { + timer.Stop() + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Second) + defer cancel() + if err := gateway.UnregisterGateway(ctx, selfAddr); err != nil { + log.ZWarn(ctx, "gateway unregister failed", err, "address", selfAddr) + } + }() for { select { case <-timer.C: + register() case <-ctx.Done(): return context.Cause(ctx) } - } } diff --git a/config/share.yml b/config/share.yml index a42bdcdd7..df0b55be2 100644 --- a/config/share.yml +++ b/config/share.yml @@ -9,6 +9,14 @@ imAdminUser: # Each entry here corresponds by index to the matching entry in the userIDs list above. nicknames: [superAdmin] +# queue: choose message queue engine +# Supported values: +# - kafka (default) +# - redis +# - memory (standalone only; microservices cannot use memory) +queue: kafka + + # 1: For Android, iOS, Windows, Mac, and web platforms, only one instance can be online at a time multiLogin: policy: 1 diff --git a/pkg/common/storage/cache/redis/standalone_gateway.go b/pkg/common/storage/cache/redis/standalone_gateway.go new file mode 100644 index 000000000..bb46fb476 --- /dev/null +++ b/pkg/common/storage/cache/redis/standalone_gateway.go @@ -0,0 +1,54 @@ +package redis + +import ( + "context" + "strconv" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/openimsdk/tools/errs" +) + +const standaloneGatewayHashKey = "STANDALONE_GATEWAY_REGISTRY" + +type StandaloneGatewayRedis struct { + rdb redis.UniversalClient + validTime time.Duration +} + +func NewStandaloneGatewayRedis(rdb redis.UniversalClient, validTime time.Duration) *StandaloneGatewayRedis { + return &StandaloneGatewayRedis{rdb: rdb, validTime: validTime} +} + +func (s *StandaloneGatewayRedis) RegisterGateway(ctx context.Context, addr string) error { + pipe := s.rdb.Pipeline() + pipe.HSet(ctx, standaloneGatewayHashKey, addr, strconv.FormatInt(time.Now().UnixMilli(), 10)) + pipe.Expire(ctx, standaloneGatewayHashKey, s.validTime*2) + _, err := pipe.Exec(ctx) + return errs.Wrap(err) +} + +func (s *StandaloneGatewayRedis) UnregisterGateway(ctx context.Context, addr string) error { + return errs.Wrap(s.rdb.HDel(ctx, standaloneGatewayHashKey, addr).Err()) +} + +func (s *StandaloneGatewayRedis) GetGatewayAddrs(ctx context.Context) ([]string, error) { + gateways, err := s.rdb.HGetAll(ctx, standaloneGatewayHashKey).Result() + if err != nil { + return nil, errs.Wrap(err) + } + + now := time.Now() + addrs := make([]string, 0, len(gateways)) + for addr, registeredAt := range gateways { + registeredAtMs, err := strconv.ParseInt(registeredAt, 10, 64) + if err != nil { + return nil, errs.WrapMsg(err, "redis gateway register time is not int64", "addr", addr, "value", registeredAt) + } + if now.Sub(time.UnixMilli(registeredAtMs)) <= s.validTime { + addrs = append(addrs, addr) + } + } + return addrs, nil +} diff --git a/pkg/common/storage/cache/redis/standalone_gateway_test.go b/pkg/common/storage/cache/redis/standalone_gateway_test.go new file mode 100644 index 000000000..b4f2c025f --- /dev/null +++ b/pkg/common/storage/cache/redis/standalone_gateway_test.go @@ -0,0 +1,52 @@ +package redis + +import ( + "context" + "strconv" + "testing" + "time" + + "github.com/go-redis/redismock/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestStandaloneGatewayRedisRegisterGateway(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second*10) + + mock.Regexp().ExpectHSet(standaloneGatewayHashKey, "127.0.0.1:10001", `^[0-9]+$`).SetVal(1) + mock.ExpectExpire(standaloneGatewayHashKey, time.Second*20).SetVal(true) + + err := cache.RegisterGateway(context.Background(), "127.0.0.1:10001") + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestStandaloneGatewayRedisUnregisterGateway(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second) + + mock.ExpectHDel(standaloneGatewayHashKey, "127.0.0.1:10001").SetVal(1) + + err := cache.UnregisterGateway(context.Background(), "127.0.0.1:10001") + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestStandaloneGatewayRedisGetGatewayAddrs(t *testing.T) { + rdb, mock := redismock.NewClientMock() + cache := NewStandaloneGatewayRedis(rdb, time.Second*10) + + now := time.Now() + mock.ExpectHGetAll(standaloneGatewayHashKey).SetVal(map[string]string{ + "127.0.0.1:10001": strconv.FormatInt(now.Add(-time.Second).UnixMilli(), 10), + "127.0.0.1:10002": strconv.FormatInt(now.Add(-time.Second*20).UnixMilli(), 10), + "127.0.0.1:10003": strconv.FormatInt(now.Add(time.Second).UnixMilli(), 10), + }) + + addrs, err := cache.GetGatewayAddrs(context.Background()) + require.NoError(t, err) + assert.Equal(t, []string{"127.0.0.1:10001", "127.0.0.1:10003"}, addrs) + assert.NoError(t, mock.ExpectationsWereMet()) +}