test: add function of create group and send messages

internal/conversation_msg/sync.go

@@ -131,7 +131,7 @@ func (c *Conversation) SyncConversationHashReadSeqs(ctx context.Context) error {
 	log.ZDebug(ctx, "update conversations", "conversations", conversations)
 	if len(conversationIDs) > 0 {
 		common.TriggerCmdUpdateConversation(ctx, common.UpdateConNode{Action: constant.ConChange, Args: conversationIDs}, c.GetCh())
-		common.TriggerCmdUpdateConversation(ctx, common.UpdateConNode{Action: constant.TotalUnreadMessageChanged, Args: conversationIDs}, c.GetCh())
+		common.TriggerCmdUpdateConversation(ctx, common.UpdateConNode{Action: constant.TotalUnreadMessageChanged}, c.GetCh())
 	}
 	return nil
 }

internal/interaction/long_conn_mgr.go

@@ -242,7 +242,8 @@ func (c *LongConnMgr) writePump(ctx context.Context) {
 				c.closedErr = ErrChanClosed
 				return
 			}
-			log.ZDebug(c.ctx, "writePump recv message", "message", message.Message)
+			log.ZDebug(c.ctx, "writePump recv message", "reqIdentifier", message.Message.ReqIdentifier,
+				"operationID", message.Message.OperationID, "sendID", message.Message.SendID)
 			resp, err := c.sendAndWaitResp(&message.Message)
 			if err != nil {
 				resp = &GeneralWsResp{
@@ -452,7 +453,8 @@ func (c *LongConnMgr) handleMessage(message []byte) error {
 		fallthrough
 	case constant.SetBackgroundStatus:
 		if err := c.Syncer.NotifyResp(ctx, wsResp); err != nil {
-			log.ZError(ctx, "notifyResp failed", err, "wsResp", wsResp)
+			log.ZError(ctx, "notifyResp failed", err, "reqIdentifier", wsResp.ReqIdentifier, "errCode",
+				wsResp.ErrCode, "errMsg", wsResp.ErrMsg, "msgIncr", wsResp.MsgIncr, "operationID", wsResp.OperationID)
 		}
 	default:
 		// log.Error(wsResp.OperationID, "type failed, ", wsResp.ReqIdentifier)

internal/util/post.go

@@ -49,11 +49,10 @@ func ApiPost(ctx context.Context, api string, req, resp any) (err error) {
 		return err
 	}
 	defer func(start time.Time) {
-		end := time.Now()
 		if err == nil {
-			log.ZDebug(ctx, "CallApi", "api", api, "use", "state", "success", time.Duration(end.UnixNano()-start.UnixNano()))
+			log.ZDebug(ctx, "CallApi", "api", api, "state", "success", "cost time", time.Since(start).Milliseconds())
 		} else {
-			log.ZError(ctx, "CallApi", err, "api", api, "use", "state", "failed", time.Duration(end.UnixNano()-start.UnixNano()))
+			log.ZError(ctx, "CallApi", err, "api", api, "state", "failed", "cost time", time.Since(start).Milliseconds())
 		}
 	}(time.Now())
 	reqBody, err := json.Marshal(req)

test/login.go

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"open_im_sdk/open_im_sdk"
+	"open_im_sdk/pkg/constant"
 	"open_im_sdk/pkg/log"
 	"open_im_sdk/pkg/utils"
 	"open_im_sdk/sdk_struct"
@@ -48,13 +49,13 @@ func (b *BaseSuccessFailed) OnSuccess(data string) {
 func InOutDoTest(uid, tk, ws, api string) {
 	var cf sdk_struct.IMConfig
 	cf.ApiAddr = api
-	cf.PlatformID = 1
+	cf.PlatformID = constant.WindowsPlatformID
 	cf.WsAddr = ws
 	cf.DataDir = "./"
 	cf.LogLevel = LogLevel
 	cf.IsExternalExtensions = true
-	cf.IsLogStandardOutput = true
-	cf.LogFilePath = ""
+	cf.IsLogStandardOutput = false
+	cf.LogFilePath = "./"
 
 	b, _ := json.Marshal(cf)
 	s := string(b)

testv3new/pressure_test.go

@@ -2,6 +2,8 @@ package testv3new
 
 import (
 	"fmt"
+	"github.com/OpenIMSDK/tools/mcontext"
+	"open_im_sdk/pkg/utils"
 	"testing"
 	"time"
 
@@ -79,3 +81,58 @@ func TestPressureTester_PressureSendMsgs2(t *testing.T) {
 		time.Sleep(time.Second)
 	}
 }
+func Test_CreateConversationsAndSendMessages(t *testing.T) {
+	if err := log.InitFromConfig("sdk.log", "sdk", 6, true, false, "", 2, 24); err != nil {
+		panic(err)
+	}
+	recvID := "6959062403"
+	conversationNum := 3
+	onePeopleMessageNum := 100
+	pressureTester := NewPressureTester(APIADDR, WSADDR, SECRET, Admin)
+	ctx := pressureTester.NewAdminCtx()
+	ctx = mcontext.SetOperationID(ctx, utils.OperationIDGenerator())
+	fixedUserIDs := []string{"register_test_1", "register_test_2", "register_test_3"}
+	pressureTester.CreateConversationsAndBatchSendMsg(ctx, conversationNum, onePeopleMessageNum, recvID, fixedUserIDs)
+	time.Sleep(time.Minute * 10)
+}
+func Test_CreateConversationsAndSendGroupMessages(t *testing.T) {
+	if err := log.InitFromConfig("sdk.log", "sdk", 6, true, false, "./", 2, 24); err != nil {
+		panic(err)
+	}
+	groupID := "227809258"
+	conversationNum := 3
+	onePeopleMessageNum := 1000
+	pressureTester := NewPressureTester(APIADDR, WSADDR, SECRET, Admin)
+	ctx := pressureTester.NewAdminCtx()
+	ctx = mcontext.SetOperationID(ctx, utils.OperationIDGenerator())
+	fixedUserIDs := []string{"register_test_1", "register_test_2", "register_test_3"}
+	pressureTester.CreateConversationsAndBatchSendGroupMsg(ctx, conversationNum, onePeopleMessageNum, groupID, fixedUserIDs)
+	time.Sleep(time.Minute * 10)
+}
+func Test_CreateGroup(t *testing.T) {
+	count := 9
+	ownerUserID := "6959062403"
+	p := NewPressureTester(APIADDR, WSADDR, SECRET, Admin)
+	ctx := p.NewAdminCtx()
+	ctx = mcontext.SetOperationID(ctx, utils.OperationIDGenerator())
+	token, err := p.testUserMananger.GetToken(ctx, ownerUserID, p.platformID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var userIDs []string
+
+	for i := 1; i <= count; i++ {
+		userIDs = append(userIDs, fmt.Sprintf("register_test_%v", i))
+	}
+	//if err := p.testUserMananger.RegisterUsers(ctx, userIDs...); err != nil {
+	//	t.Fatal(err)
+	//}
+	ctx = p.NewCtx(ownerUserID, token)
+	ctx = mcontext.SetOperationID(ctx, utils.OperationIDGenerator())
+	err = p.testUserMananger.CreateGroup(ctx, "", ownerUserID, userIDs, fmt.Sprintf("group_test_%v", "Gordon"))
+	if err != nil {
+		t.Fatal(err)
+	} else {
+		t.Log("create group success")
+	}
+}

testv3new/pressure_tester.go

@@ -165,6 +165,73 @@ func (p *PressureTester) CreateConversations(ctx context.Context, conversationNu
 	wg.Wait()
 	return nil
 }
+func (p *PressureTester) CreateConversationsAndBatchSendMsg(ctx context.Context, conversationNum int, onePeopleMessageNum int,
+	recvUserID string, fixedUserIDs []string) error {
+	var userIDs []string
+	if len(fixedUserIDs) <= 0 {
+		userIDs = p.testUserMananger.GenUserIDs(conversationNum)
+		if err := p.testUserMananger.RegisterUsers(ctx, userIDs...); err != nil {
+			return err
+		}
+	} else {
+		userIDs = fixedUserIDs
+	}
+
+	var wg sync.WaitGroup
+	for _, userID := range userIDs {
+		go func() {
+			time.Sleep(time.Millisecond * 100)
+			token, _ := p.testUserMananger.GetToken(ctx, userID, p.platformID)
+			ctx2 := NewUserCtx(userID, token)
+			baseCore := testcore.NewBaseCore(ctx2, userID, p.platformID)
+			ctx2 = mcontext.SetOperationID(ctx2, utils.OperationIDGenerator())
+			for i := 0; i < onePeopleMessageNum; i++ {
+				if err := baseCore.BatchSendSingleMsg(ctx2, recvUserID, i); err != nil {
+					log.ZError(ctx2, "send msg error", err, "sendUserID", userID)
+				}
+				time.Sleep(time.Millisecond * 5000)
+			}
+		}()
+
+	}
+	wg.Add(1)
+	wg.Wait()
+	return nil
+}
+func (p *PressureTester) CreateConversationsAndBatchSendGroupMsg(ctx context.Context, conversationNum int, onePeopleMessageNum int,
+	groupID string, fixedUserIDs []string) error {
+	var userIDs []string
+	if len(fixedUserIDs) <= 0 {
+		userIDs = p.testUserMananger.GenUserIDs(conversationNum)
+		if err := p.testUserMananger.RegisterUsers(ctx, userIDs...); err != nil {
+			return err
+		}
+	} else {
+		userIDs = fixedUserIDs
+	}
+
+	var wg sync.WaitGroup
+	for _, userID := range userIDs {
+		go func(u string) {
+			log.ZDebug(ctx, "start send msg", "userID", u)
+			time.Sleep(time.Millisecond * 100)
+			token, _ := p.testUserMananger.GetToken(ctx, u, p.platformID)
+			ctx2 := NewUserCtx(u, token)
+			baseCore := testcore.NewBaseCore(ctx2, u, p.platformID)
+			ctx2 = mcontext.SetOperationID(ctx2, utils.OperationIDGenerator())
+			for i := 0; i < onePeopleMessageNum; i++ {
+				if err := baseCore.BatchSendGroupMsg(ctx2, groupID, i); err != nil {
+					log.ZError(ctx2, "send msg error", err, "sendUserID", u)
+				}
+				time.Sleep(time.Millisecond * 500)
+			}
+		}(userID)
+
+	}
+	wg.Add(1)
+	wg.Wait()
+	return nil
+}
 
 // PressureSendMsgs2 user single chat send msg pressure test
 func (p *PressureTester) PressureSendMsgs2(ctx context.Context, sendUserIDs []string, recvUserIDs []string, msgNum int, duration time.Duration) {

testv3new/test_user_manager.go

@@ -84,6 +84,7 @@ func (p *TestUserManager) CreateGroup(ctx context.Context, groupID string, owner
 		GroupInfo: &sdkws.GroupInfo{
 			GroupID:   groupID,
 			GroupName: groupName,
+			GroupType: constant.WorkingGroup,
 		},
 	}
 	resp := &group.CreateGroupResp{}

testv3new/testcore/lightweight.go

@@ -21,6 +21,7 @@ type BaseCore struct {
 	platformID          int32
 	pushMsgAndMaxSeqCh  chan common.Cmd2Value
 	recvPushMsgCallback func(msg *sdkws.MsgData)
+	failedMessageMap    map[string]error
 }
 
 func (b BaseCore) GetUserID() string {
@@ -41,6 +42,7 @@ func NewBaseCore(ctx context.Context, userID string, platformID int32, opts ...f
 		longConnMgr:        longConnMgr,
 		userID:             userID,
 		platformID:         platformID,
+		failedMessageMap:   make(map[string]error),
 	}
 	for _, opt := range opts {
 		opt(core)
@@ -57,10 +59,28 @@ func (b *BaseCore) Close(ctx context.Context) {
 func (b *BaseCore) SendSingleMsg(ctx context.Context, userID string, index int) error {
 	return b.sendMsg(ctx, userID, "", index, constant.SingleChatType, fmt.Sprintf("this is test msg user %s to user %s, index: %d", b.userID, userID, index))
 }
+func (b *BaseCore) BatchSendSingleMsg(ctx context.Context, userID string, index int) error {
+	content := fmt.Sprintf("this is test msg user %s to user %s, index: %d", b.userID, userID, index)
+	err := b.sendMsg(ctx, userID, "", index, constant.SingleChatType, content)
+	if err != nil {
+		log.ZError(ctx, "send msg failed", err, "userID", userID, "index", index, "content", content)
+		b.failedMessageMap[content] = err
+	}
+	return nil
+}
 
 func (b *BaseCore) SendGroupMsg(ctx context.Context, groupID string, index int) error {
 	return b.sendMsg(ctx, "", groupID, index, constant.SuperGroupChatType, fmt.Sprintf("this is test msg user %s to group %s, index: %d", b.userID, groupID, index))
 }
+func (b *BaseCore) BatchSendGroupMsg(ctx context.Context, groupID string, index int) error {
+	content := fmt.Sprintf("this is test msg user %s to group %s, index: %d", b.userID, groupID, index)
+	err := b.sendMsg(ctx, "", groupID, index, constant.SuperGroupChatType, content)
+	if err != nil {
+		log.ZError(ctx, "send msg failed", err, "groupID", groupID, "index", index, "content", content)
+		b.failedMessageMap[content] = err
+	}
+	return nil
+}
 
 func (b *BaseCore) sendMsg(ctx context.Context, userID, groupID string, index int, sesstionType int32, content string) error {
 	var resp sdkws.UserSendMsgResp