mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-04-29 06:49:19 +08:00
Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e99c8ab065 | |||
| 2aab7034b0 | |||
| 7a9c336a96 | |||
| 02025278b3 | |||
| 90596b1a02 |
+6
-1
@@ -63,7 +63,12 @@ services:
|
||||
restart: always
|
||||
sysctls:
|
||||
net.core.somaxconn: 1024
|
||||
command: redis-server /usr/local/redis/config/redis.conf --requirepass openIM123 --appendonly yes
|
||||
command: >
|
||||
redis-server
|
||||
--requirepass openIM123
|
||||
--appendonly yes
|
||||
--aof-use-rdb-preamble yes
|
||||
--save ""
|
||||
networks:
|
||||
- openim
|
||||
|
||||
|
||||
@@ -244,6 +244,7 @@ func (s *Server) MultiTerminalLoginCheck(ctx context.Context, req *msggateway.Mu
|
||||
tempUserCtx.SetOperationID(mcontext.GetOperationID(ctx))
|
||||
client := &Client{}
|
||||
client.ctx = tempUserCtx
|
||||
client.token = req.Token
|
||||
client.UserID = req.UserID
|
||||
client.PlatformID = int(req.PlatformID)
|
||||
i := &kickHandler{
|
||||
|
||||
@@ -337,17 +337,51 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
|
||||
}
|
||||
}
|
||||
|
||||
// If reconnect: When multiple msgGateway instances are deployed, a client may disconnect from instance A and reconnect to instance B.
|
||||
// During this process, instance A might still be executing, resulting in two clients with the same token existing simultaneously.
|
||||
// This situation needs to be filtered to prevent duplicate clients.
|
||||
checkSameTokenFunc := func(oldClients []*Client) []*Client {
|
||||
var clientsNeedToKick []*Client
|
||||
|
||||
for _, c := range oldClients {
|
||||
if c.token == newClient.token {
|
||||
log.ZDebug(newClient.ctx, "token is same, not kick",
|
||||
"userID", newClient.UserID,
|
||||
"platformID", newClient.PlatformID,
|
||||
"token", newClient.token)
|
||||
continue
|
||||
}
|
||||
|
||||
clientsNeedToKick = append(clientsNeedToKick, c)
|
||||
}
|
||||
|
||||
return clientsNeedToKick
|
||||
}
|
||||
|
||||
switch ws.msgGatewayConfig.Share.MultiLogin.Policy {
|
||||
case constant.DefalutNotKick:
|
||||
case constant.PCAndOther:
|
||||
if constant.PlatformIDToClass(newClient.PlatformID) == constant.TerminalPC {
|
||||
return
|
||||
}
|
||||
clients, ok := ws.clients.GetAll(newClient.UserID)
|
||||
clientOK = ok
|
||||
oldClients = make([]*Client, 0, len(clients))
|
||||
for _, c := range clients {
|
||||
if constant.PlatformIDToClass(c.PlatformID) == constant.TerminalPC {
|
||||
continue
|
||||
}
|
||||
oldClients = append(oldClients, c)
|
||||
}
|
||||
|
||||
fallthrough
|
||||
case constant.AllLoginButSameTermKick:
|
||||
if !clientOK {
|
||||
return
|
||||
}
|
||||
|
||||
oldClients = checkSameTokenFunc(oldClients)
|
||||
|
||||
ws.clients.DeleteClients(newClient.UserID, oldClients)
|
||||
for _, c := range oldClients {
|
||||
err := c.KickOnlineMessage()
|
||||
@@ -373,14 +407,15 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var (
|
||||
kickClients []*Client
|
||||
)
|
||||
|
||||
var kickClients []*Client
|
||||
for _, client := range clients {
|
||||
if constant.PlatformIDToClass(client.PlatformID) == constant.PlatformIDToClass(newClient.PlatformID) {
|
||||
kickClients = append(kickClients, client)
|
||||
}
|
||||
}
|
||||
kickClients = checkSameTokenFunc(kickClients)
|
||||
|
||||
kickTokenFunc(kickClients)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -675,12 +675,19 @@ func (db *commonMsgDatabase) SearchMessage(ctx context.Context, req *pbmsg.Searc
|
||||
func (db *commonMsgDatabase) FindOneByDocIDs(ctx context.Context, conversationIDs []string, seqs map[string]int64) (map[string]*sdkws.MsgData, error) {
|
||||
totalMsgs := make(map[string]*sdkws.MsgData)
|
||||
for _, conversationID := range conversationIDs {
|
||||
seq := seqs[conversationID]
|
||||
seq, ok := seqs[conversationID]
|
||||
if !ok {
|
||||
log.ZWarn(ctx, "seq not found for conversationID", errs.New("seq not found for conversation"), "conversationID", conversationID)
|
||||
continue
|
||||
}
|
||||
|
||||
docID := db.msgTable.GetDocID(conversationID, seq)
|
||||
msgs, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
log.ZWarn(ctx, "FindOneByDocID failed", err, "conversationID", conversationID, "docID", docID, "seq", seq)
|
||||
continue
|
||||
}
|
||||
|
||||
index := db.msgTable.GetMsgIndex(seq)
|
||||
totalMsgs[conversationID] = convert.MsgDB2Pb(msgs.Msg[index].Msg)
|
||||
}
|
||||
|
||||
@@ -5,6 +5,11 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
@@ -14,10 +19,6 @@ import (
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/openimsdk/tools/utils/jsonutil"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
)
|
||||
|
||||
func NewMsgMongo(db *mongo.Database) (database.Msg, error) {
|
||||
@@ -1149,7 +1150,7 @@ func (m *MsgMgo) findBeforeDocSendTime(ctx context.Context, docID string, limit
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
for i := len(res) - 1; i > 0; i-- {
|
||||
for i := len(res) - 1; i >= 0; i-- {
|
||||
v := res[i]
|
||||
if v.Msgs != nil && v.Msgs.Msg != nil && v.Msgs.Msg.SendTime > 0 {
|
||||
return v.Msgs.Msg.Seq, v.Msgs.Msg.SendTime, nil
|
||||
@@ -1164,7 +1165,7 @@ func (m *MsgMgo) findBeforeSendTime(ctx context.Context, conversationID string,
|
||||
limit := int64(-1)
|
||||
if first {
|
||||
first = false
|
||||
limit = m.model.GetMsgIndex(seq)
|
||||
limit = m.model.GetLimitForSingleDoc(seq)
|
||||
}
|
||||
docID := m.model.BuildDocIDByIndex(conversationID, i)
|
||||
msgSeq, msgSendTime, err := m.findBeforeDocSendTime(ctx, docID, limit)
|
||||
|
||||
@@ -120,15 +120,11 @@ func (m *MsgDocModel) GetDocID(conversationID string, seq int64) string {
|
||||
|
||||
func (m *MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
|
||||
t := make(map[string][]int64)
|
||||
for i := 0; i < len(seqs); i++ {
|
||||
docID := m.GetDocID(conversationID, seqs[i])
|
||||
if value, ok := t[docID]; !ok {
|
||||
var temp []int64
|
||||
t[docID] = append(temp, seqs[i])
|
||||
} else {
|
||||
t[docID] = append(value, seqs[i])
|
||||
}
|
||||
for _, seq := range seqs {
|
||||
docID := m.GetDocID(conversationID, seq)
|
||||
t[docID] = append(t[docID], seq)
|
||||
}
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
@@ -136,6 +132,10 @@ func (*MsgDocModel) GetMsgIndex(seq int64) int64 {
|
||||
return (seq - 1) % singleGocMsgNum
|
||||
}
|
||||
|
||||
func (*MsgDocModel) GetLimitForSingleDoc(seq int64) int64 {
|
||||
return seq % singleGocMsgNum
|
||||
}
|
||||
|
||||
func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
|
||||
return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
|
||||
}
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
v3.8.3
|
||||
v3.8.3-patch.9
|
||||
Reference in New Issue
Block a user