Merge remote-tracking branch 'upstream/main' into main-new

internal/msggateway/constant.go

@@ -26,6 +26,7 @@ const (
 	Compression             = "compression"
 	GzipCompressionProtocol = "gzip"
 	BackgroundStatus        = "isBackground"
+	MsgResp                 = "isMsgResp"
 )
 
 const (

internal/msggateway/n_ws_server.go

@@ -16,7 +16,10 @@ package msggateway
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
+	"github.com/OpenIMSDK/tools/apiresp"
 	"net/http"
 	"os"
 	"os/signal"
@@ -422,84 +425,102 @@ func (ws *WsServer) unregisterClient(client *Client) {
 	)
 }
 
-func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
+func (ws *WsServer) ParseWSArgs(r *http.Request) (args *WSArgs, err error) {
-	connContext := newContext(w, r)
+	var v WSArgs
+	defer func() {
+		args = &v
+	}()
+	query := r.URL.Query()
+	v.MsgResp, _ = strconv.ParseBool(query.Get(MsgResp))
 	if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum {
-		httpError(connContext, errs.ErrConnOverMaxNumLimit)
+		return nil, errs.ErrConnOverMaxNumLimit.Wrap("over max conn num limit")
-		return
 	}
-	var (
+	if v.Token = query.Get(Token); v.Token == "" {
-		token         string
+		return nil, errs.ErrConnArgsErr.Wrap("token is empty")
-		userID        string
-		platformIDStr string
-		exists        bool
-		compression   bool
-	)
-
-	token, exists = connContext.Query(Token)
-	if !exists {
-		httpError(connContext, errs.ErrConnArgsErr)
-		return
 	}
-	userID, exists = connContext.Query(WsUserID)
+	if v.UserID = query.Get(WsUserID); v.UserID == "" {
-	if !exists {
+		return nil, errs.ErrConnArgsErr.Wrap("sendID is empty")
-		httpError(connContext, errs.ErrConnArgsErr)
-		return
 	}
-	platformIDStr, exists = connContext.Query(PlatformID)
+	platformIDStr := query.Get(PlatformID)
-	if !exists {
+	if platformIDStr == "" {
-		httpError(connContext, errs.ErrConnArgsErr)
+		return nil, errs.ErrConnArgsErr.Wrap("platformID is empty")
-		return
 	}
 	platformID, err := strconv.Atoi(platformIDStr)
 	if err != nil {
-		httpError(connContext, errs.ErrConnArgsErr)
+		return nil, errs.ErrConnArgsErr.Wrap("platformID is not int")
-		return
+	}
+	v.PlatformID = platformID
+	if err = authverify.WsVerifyToken(v.Token, v.UserID, platformID); err != nil {
+		return nil, err
+	}
+	if query.Get(Compression) == GzipCompressionProtocol {
+		v.Compression = true
 	}
-	if err = authverify.WsVerifyToken(token, userID, platformID); err != nil {
+	if r.Header.Get(Compression) == GzipCompressionProtocol {
-		httpError(connContext, err)
+		v.Compression = true
-		return
 	}
-	m, err := ws.cache.GetTokensWithoutError(context.Background(), userID, platformID)
+	m, err := ws.cache.GetTokensWithoutError(context.Background(), v.UserID, platformID)
 	if err != nil {
-		httpError(connContext, err)
+		return nil, err
-		return
 	}
-	if v, ok := m[token]; ok {
+	if v, ok := m[v.Token]; ok {
 		switch v {
 		case constant.NormalToken:
 		case constant.KickedToken:
-			httpError(connContext, errs.ErrTokenKicked.Wrap())
+			return nil, errs.ErrTokenKicked.Wrap()
-			return
 		default:
-			httpError(connContext, errs.ErrTokenUnknown.Wrap())
+			return nil, errs.ErrTokenUnknown.Wrap(fmt.Sprintf("token status is %d", v))
-			return
 		}
 	} else {
-		httpError(connContext, errs.ErrTokenNotExist.Wrap())
+		return nil, errs.ErrTokenNotExist.Wrap()
-		return
 	}
+	return &v, nil
+}
 
-	wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
+type WSArgs struct {
-	err = wsLongConn.GenerateLongConn(w, r)
+	Token       string
-	if err != nil {
+	UserID      string
-		httpError(connContext, err)
+	PlatformID  int
-		return
+	Compression bool
-	}
+	MsgResp     bool
-	compressProtoc, exists := connContext.Query(Compression)
+}
-	if exists {
+
-		if compressProtoc == GzipCompressionProtocol {
+func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
-			compression = true
+	connContext := newContext(w, r)
+	args, pErr := ws.ParseWSArgs(r)
+	var wsLongConn *GWebSocket
+	if args.MsgResp {
+		wsLongConn = newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
+		if err := wsLongConn.GenerateLongConn(w, r); err != nil {
+			httpError(connContext, err)
+			return
 		}
-	}
+		data, err := json.Marshal(apiresp.ParseError(pErr))
-	compressProtoc, exists = connContext.GetHeader(Compression)
+		if err != nil {
-	if exists {
+			_ = wsLongConn.Close()
-		if compressProtoc == GzipCompressionProtocol {
+			return
-			compression = true
+		}
+		if err := wsLongConn.WriteMessage(MessageText, data); err != nil {
+			_ = wsLongConn.Close()
+			return
+		}
+		if pErr != nil {
+			_ = wsLongConn.Close()
+			return
+		}
+	} else {
+		if pErr != nil {
+			httpError(connContext, pErr)
+			return
+		}
+		wsLongConn = newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
+		if err := wsLongConn.GenerateLongConn(w, r); err != nil {
+			httpError(connContext, err)
+			return
 		}
 	}
 	client := ws.clientPool.Get().(*Client)
-	client.ResetClient(connContext, wsLongConn, connContext.GetBackground(), compression, ws, token)
+	client.ResetClient(connContext, wsLongConn, connContext.GetBackground(), args.Compression, ws, args.Token)
 	ws.registerChan <- client
 	go client.readMessage()
 }

scripts/install-im-server.sh

@@ -45,8 +45,19 @@ pushd "${OPENIM_ROOT}"
 ${DOCKER_COMPOSE_COMMAND} stop
 curl https://gitee.com/openimsdk/openim-docker/raw/main/example/full-openim-server-and-chat.yml -o docker-compose.yml
 ${DOCKER_COMPOSE_COMMAND} up -d
-sleep 60
+
+# Wait for a short period to allow containers to initialize
+sleep 10
+
+# Check the status of the containers
+if ! ${DOCKER_COMPOSE_COMMAND} ps | grep -q 'Up'; then
+    echo "Error: One or more docker containers failed to start."
+    ${DOCKER_COMPOSE_COMMAND} logs
+    exit 1
+fi
+
+sleep 50  # Keep the original 60-second wait, adjusted for the 10-second check above
 ${DOCKER_COMPOSE_COMMAND} logs openim-server
 ${DOCKER_COMPOSE_COMMAND} ps
 
-popd
+popd