feat: testing program that simulates various scenarios of sending messages.

internal/conversation_msg/conversation_notification.go

@@ -168,23 +168,6 @@ func (c *Conversation) doNotificationNew(c2v common.Cmd2Value) {
 
 }
 
-func (c *Conversation) doDeleteConversation(c2v common.Cmd2Value) {
-	node := c2v.Value.(common.DeleteConNode)
-	ctx := c2v.Ctx
-	// Mark messages related to this conversation for deletion
-	err := c.db.UpdateMessageStatusBySourceID(context.Background(), node.SourceID, constant.MsgStatusHasDeleted, int32(node.SessionType))
-	if err != nil {
-		log.ZError(ctx, "setMessageStatusBySourceID", err)
-		return
-	}
-	// Reset the session information, empty session
-	err = c.db.ResetConversation(ctx, node.ConversationID)
-	if err != nil {
-		log.ZError(ctx, "ResetConversation err:", err)
-	}
-	c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{"", constant.TotalUnreadMessageChanged, ""}})
-}
-
 func (c *Conversation) getConversationLatestMsgClientID(latestMsg string) string {
 	msg := &sdk_struct.MsgStruct{}
 	if err := json.Unmarshal([]byte(latestMsg), msg); err != nil {

internal/conversation_msg/message_check.go

@@ -17,6 +17,7 @@ package conversation_msg
 import (
 	"context"
 	"errors"
+
 	"github.com/openimsdk/openim-sdk-core/v3/pkg/constant"
 	"github.com/openimsdk/openim-sdk-core/v3/pkg/db/model_struct"
 	sdk "github.com/openimsdk/openim-sdk-core/v3/pkg/sdk_params_callback"

internal/conversation_msg/read_drawing.go

@@ -219,7 +219,6 @@ func (c *Conversation) doUnreadCount(ctx context.Context, conversation *model_st
 			log.ZError(ctx, "UpdateColumnsConversation err", err, "conversationID", conversation.ConversationID)
 		}
 	}
-
 	c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{ConID: conversation.ConversationID, Action: constant.ConChange, Args: []string{conversation.ConversationID}}})
 	c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{Action: constant.TotalUnreadMessageChanged}})
 

internal/conversation_msg/sdk.go

@@ -388,6 +388,11 @@ func (c *Conversation) getConversationIDBySessionType(sourceID string, sessionTy
 func (c *Conversation) GetConversationIDBySessionType(_ context.Context, sourceID string, sessionType int) string {
 	return c.getConversationIDBySessionType(sourceID, sessionType)
 }
+
+//this is a test file
+/**
+his is a test file
+*/
 func (c *Conversation) SendMessage(ctx context.Context, s *sdk_struct.MsgStruct, recvID, groupID string, p *sdkws.OfflinePushInfo, isOnlineOnly bool) (*sdk_struct.MsgStruct, error) {
 	filepathExt := func(name ...string) string {
 		for _, path := range name {

internal/interaction/compressor.go

@@ -17,13 +17,22 @@ package interaction
 import (
 	"bytes"
 	"compress/gzip"
+	"errors"
 	"github.com/openimsdk/openim-sdk-core/v3/pkg/utils"
 	"io"
+	"sync"
+)
+
+var (
+	gzipWriterPool = sync.Pool{New: func() any { return gzip.NewWriter(nil) }}
+	gzipReaderPool = sync.Pool{New: func() any { return new(gzip.Reader) }}
 )
 
 type Compressor interface {
 	Compress(rawData []byte) ([]byte, error)
+	CompressWithPool(rawData []byte) ([]byte, error)
 	DeCompress(compressedData []byte) ([]byte, error)
+	DecompressWithPool(compressedData []byte) ([]byte, error)
 }
 
 type GzipCompressor struct {
@@ -46,6 +55,22 @@ func (g *GzipCompressor) Compress(rawData []byte) ([]byte, error) {
 	return gzipBuffer.Bytes(), nil
 }
 
+func (g *GzipCompressor) CompressWithPool(rawData []byte) ([]byte, error) {
+	gz := gzipWriterPool.Get().(*gzip.Writer)
+	defer gzipWriterPool.Put(gz)
+
+	gzipBuffer := bytes.Buffer{}
+	gz.Reset(&gzipBuffer)
+
+	if _, err := gz.Write(rawData); err != nil {
+		return nil, utils.Wrap(err, "")
+	}
+	if err := gz.Close(); err != nil {
+		return nil, utils.Wrap(err, "")
+	}
+	return gzipBuffer.Bytes(), nil
+}
+
 func (g *GzipCompressor) DeCompress(compressedData []byte) ([]byte, error) {
 	buff := bytes.NewBuffer(compressedData)
 	reader, err := gzip.NewReader(buff)
@@ -59,3 +84,23 @@ func (g *GzipCompressor) DeCompress(compressedData []byte) ([]byte, error) {
 	_ = reader.Close()
 	return compressedData, nil
 }
+
+func (g *GzipCompressor) DecompressWithPool(compressedData []byte) ([]byte, error) {
+	reader := gzipReaderPool.Get().(*gzip.Reader)
+	if reader == nil {
+		return nil, errors.New("NewReader failed")
+	}
+	defer gzipReaderPool.Put(reader)
+
+	err := reader.Reset(bytes.NewReader(compressedData))
+	if err != nil {
+		return nil, utils.Wrap(err, "NewReader failed")
+	}
+
+	compressedData, err = io.ReadAll(reader)
+	if err != nil {
+		return nil, utils.Wrap(err, "ReadAll failed")
+	}
+	_ = reader.Close()
+	return compressedData, nil
+}

internal/interaction/long_conn_mgr.go

@@ -124,7 +124,6 @@ func NewLongConnMgr(ctx context.Context, listener open_im_sdk_callback.OnConnLis
 	return l
 }
 func (c *LongConnMgr) Run(ctx context.Context) {
-	//fmt.Println(mcontext.GetOperationID(ctx), "login run", string(debug.Stack()))
 	go c.readPump(ctx)
 	go c.writePump(ctx)
 	go c.heartbeat(ctx)
@@ -176,7 +175,6 @@ func (c *LongConnMgr) readPump(ctx context.Context) {
 		log.ZWarn(c.ctx, "readPump closed", c.closedErr)
 	}()
 	connNum := 0
-	//c.conn.SetPongHandler(function(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
 	for {
 		ctx = ccontext.WithOperationID(ctx, utils.OperationIDGenerator())
 		needRecon, err := c.reConn(ctx, &connNum)
@@ -193,11 +191,6 @@ func (c *LongConnMgr) readPump(ctx context.Context) {
 		_ = c.conn.SetReadDeadline(pongWait)
 		messageType, message, err := c.conn.ReadMessage()
 		if err != nil {
-			//if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
-			//	log.Printf("error: %v", err)
-			//}
-			//break
-			//c.closedErr = err
 			log.ZError(c.ctx, "readMessage err", err, "goroutine ID:", getGoroutineID())
 			_ = c.close()
 			continue
@@ -212,9 +205,6 @@ func (c *LongConnMgr) readPump(ctx context.Context) {
 		case MessageText:
 			c.closedErr = ErrNotSupportMessageProtocol
 			return
-		//case PingMessage:
-		//	err := c.writePongMsg()
-		//	log.ZError(c.ctx, "writePongMsg", err)
 		case CloseMessage:
 			c.closedErr = ErrClientClosed
 			return
@@ -290,9 +280,21 @@ func (c *LongConnMgr) heartbeat(ctx context.Context) {
 			log.ZInfo(ctx, "heartbeat done sdk logout.....")
 			return
 		case <-c.heartbeatCh:
-			c.sendPingToServer(ctx)
+			c.retrieveMaxSeq(ctx)
 		case <-ticker.C:
-			c.sendPingToServer(ctx)
+			c.sendPingMessage(ctx)
+		}
+	}
+
+}
+func (c *LongConnMgr) sendPingMessage(ctx context.Context) {
+	c.connWrite.Lock()
+	defer c.connWrite.Unlock()
+	log.ZInfo(ctx, "ping message tart", "goroutine ID:", getGoroutineID())
+	if c.IsConnected() {
+		c.conn.SetWriteDeadline(writeWait)
+		if err := c.conn.WriteMessage(PingMessage, nil); err != nil {
+			return
 		}
 	}
 
@@ -307,15 +309,16 @@ func getGoroutineID() int64 {
 	}
 	return id
 }
-func (c *LongConnMgr) sendPingToServer(ctx context.Context) {
+
+func (c *LongConnMgr) retrieveMaxSeq(ctx context.Context) {
 	if c.conn == nil {
 		return
 	}
 	var m sdkws.GetMaxSeqReq
 	m.UserID = ccontext.Info(ctx).UserID()
 	opID := utils.OperationIDGenerator()
 	sCtx := ccontext.WithOperationID(c.ctx, opID)
-	log.ZInfo(sCtx, "ping and getMaxSeq start", "goroutine ID:", getGoroutineID())
+	log.ZInfo(sCtx, "retrieveMaxSeq start", "goroutine ID:", getGoroutineID())
 	data, err := proto.Marshal(&m)
 	if err != nil {
 		log.ZError(sCtx, "proto.Marshal", err)
@@ -335,7 +338,7 @@ func (c *LongConnMgr) sendPingToServer(ctx context.Context) {
 		return
 	} else {
 		if resp.ErrCode != 0 {
-			log.ZError(sCtx, "getMaxSeq failed", nil, "errCode:", resp.ErrCode, "errMsg:", resp.ErrMsg)
+			log.ZError(sCtx, "retrieveMaxSeq failed", nil, "errCode:", resp.ErrCode, "errMsg:", resp.ErrMsg)
 		}
 		var wsSeqResp sdkws.GetMaxSeqResp
 		err = proto.Unmarshal(resp.Data, &wsSeqResp)
@@ -400,7 +403,7 @@ func (c *LongConnMgr) writeBinaryMsg(req GeneralWsReq) error {
 	}
 	_ = c.conn.SetWriteDeadline(writeWait)
 	if c.IsCompression {
-		resultBuf, compressErr := c.compressor.Compress(encodeBuf)
+		resultBuf, compressErr := c.compressor.CompressWithPool(encodeBuf)
 		if compressErr != nil {
 			return compressErr
 		}
@@ -423,7 +426,7 @@ func (c *LongConnMgr) close() error {
 func (c *LongConnMgr) handleMessage(message []byte) error {
 	if c.IsCompression {
 		var decompressErr error
-		message, decompressErr = c.compressor.DeCompress(message)
+		message, decompressErr = c.compressor.DecompressWithPool(message)
 		if decompressErr != nil {
 			log.ZError(c.ctx, "DeCompress failed", decompressErr, message)
 			return sdkerrs.ErrMsgDeCompression
@@ -547,6 +550,7 @@ func (c *LongConnMgr) reConn(ctx context.Context, num *int) (needRecon bool, err
 	c.ctx = newContext(c.conn.LocalAddr())
 	c.ctx = context.WithValue(ctx, "ConnContext", c.ctx)
 	c.SetConnectionStatus(Connected)
+	c.conn.SetPongHandler(c.pongHandler)
 	*num++
 	log.ZInfo(c.ctx, "long conn establish success", "localAddr", c.conn.LocalAddr(), "connNum", *num)
 	c.reconnectStrategy.Reset()

msgtest/config.go

@@ -10,9 +10,9 @@ import (
 
 // system
 var (
-	TESTIP        = "59.36.173.89"
-	APIADDR       = fmt.Sprintf("http://%v:10002", TESTIP)
-	WSADDR        = fmt.Sprintf("ws://%v:10001", TESTIP)
+	TESTIP        = "127.0.0.1"
+	APIADDR       = fmt.Sprintf("http://%v:20002", TESTIP)
+	WSADDR        = fmt.Sprintf("ws://%v:20001", TESTIP)
 	SECRET        = "openIM123"
 	MANAGERUSERID = "openIMAdmin"