diff --git a/build/goreleaser.yaml b/build/goreleaser.yaml index ed7f7cd1b..93fe9f4c8 100644 --- a/build/goreleaser.yaml +++ b/build/goreleaser.yaml @@ -3,12 +3,36 @@ before: hooks: + - make clean # You may remove this if you don't use go modules. - make tidy - make copyright.add # you may remove this if you don't need go generate - go generate ./... +git: + # What should be used to sort tags when gathering the current and previous + # tags if there are more than one tag in the same commit. + # + # Default: '-version:refname' + tag_sort: -version:creatordate + + # What should be used to specify prerelease suffix while sorting tags when gathering + # the current and previous tags if there are more than one tag in the same commit. + # + # Since: v1.17 + prerelease_suffix: "-" + + # Tags to be ignored by GoReleaser. + # This means that GoReleaser will not pick up tags that match any of the + # provided values as either previous or current tags. + # + # Templates: allowed. + # Since: v1.21. + ignore_tags: + - nightly + # - "{{.Env.IGNORE_TAG}}" + snapshot: name_template: "{{ incpatch .Version }}-next" @@ -495,4 +519,4 @@ checksum: algorithm: sha256 release: - prerelease: auto \ No newline at end of file + prerelease: auto diff --git a/docker-compose.yml b/docker-compose.yml index 5b1e11d4e..fd71896a7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -124,7 +124,7 @@ services: ## Uncomment and configure the following services as needed # openim-admin: - # image: ${IMAGE_REGISTRY:-ghcr.io/openimsdk}/openim-admin-front:v3.4.0 + # image: ${IMAGE_REGISTRY:-ghcr.io/openimsdk}/openim-admin:toc-base-open-docker.35 # container_name: openim-admin # restart: always # ports: diff --git a/go.mod b/go.mod index 9b47aebe6..f1b3239c3 100644 --- a/go.mod +++ b/go.mod @@ -155,4 +155,4 @@ require ( go.uber.org/zap v1.24.0 // indirect golang.org/x/crypto v0.14.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect -) \ No newline at end of file +) diff --git a/go.sum b/go.sum index fc1d15242..e72d862e6 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c= github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ= -github.com/OpenIMSDK/protocol v0.0.42 h1:vIWXqZJZZ1ddleJA25fxhjZ1GyEHATpYM3wVWh4/+PY= -github.com/OpenIMSDK/protocol v0.0.42/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= +github.com/OpenIMSDK/protocol v0.0.43 h1:8B921vEyO7r0AfQfZd7kCycYja+hJ2vuIZsKge/WRhU= +github.com/OpenIMSDK/protocol v0.0.43/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= github.com/OpenIMSDK/tools v0.0.21 h1:iTapc2mIEVH/xl5Nd6jfwPub11Pgp44tVcE1rjB3a48= github.com/OpenIMSDK/tools v0.0.21/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI= github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM= diff --git a/internal/msggateway/constant.go b/internal/msggateway/constant.go index fe5f09bdc..045629b4e 100644 --- a/internal/msggateway/constant.go +++ b/internal/msggateway/constant.go @@ -26,6 +26,7 @@ const ( Compression = "compression" GzipCompressionProtocol = "gzip" BackgroundStatus = "isBackground" + MsgResp = "isMsgResp" ) const ( diff --git a/internal/msggateway/n_ws_server.go b/internal/msggateway/n_ws_server.go index 70c2f8fe0..7e8129105 100644 --- a/internal/msggateway/n_ws_server.go +++ b/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" @@ -342,11 +345,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien if !clientOK { return } - - isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients) - if isDeleteUser { - ws.onlineUserNum.Add(-1) - } + ws.clients.deleteClients(newClient.UserID, oldClients) for _, c := range oldClients { err := c.KickOnlineMessage() if err != nil { @@ -422,84 +421,102 @@ func (ws *WsServer) unregisterClient(client *Client) { ) } -func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { - connContext := newContext(w, r) +func (ws *WsServer) ParseWSArgs(r *http.Request) (args *WSArgs, err error) { + 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 + return nil, errs.ErrConnOverMaxNumLimit.Wrap("over max conn num limit") } - var ( - token string - userID string - platformIDStr string - exists bool - compression bool - ) - - token, exists = connContext.Query(Token) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + if v.Token = query.Get(Token); v.Token == "" { + return nil, errs.ErrConnArgsErr.Wrap("token is empty") } - userID, exists = connContext.Query(WsUserID) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + if v.UserID = query.Get(WsUserID); v.UserID == "" { + return nil, errs.ErrConnArgsErr.Wrap("sendID is empty") } - platformIDStr, exists = connContext.Query(PlatformID) - if !exists { - httpError(connContext, errs.ErrConnArgsErr) - return + platformIDStr := query.Get(PlatformID) + if platformIDStr == "" { + return nil, errs.ErrConnArgsErr.Wrap("platformID is empty") } platformID, err := strconv.Atoi(platformIDStr) if err != nil { - httpError(connContext, errs.ErrConnArgsErr) - return + return nil, errs.ErrConnArgsErr.Wrap("platformID is not int") } - if err = authverify.WsVerifyToken(token, userID, platformID); err != nil { - httpError(connContext, err) - return + v.PlatformID = platformID + if err = authverify.WsVerifyToken(v.Token, v.UserID, platformID); err != nil { + return nil, err } - m, err := ws.cache.GetTokensWithoutError(context.Background(), userID, platformID) + if query.Get(Compression) == GzipCompressionProtocol { + v.Compression = true + } + if r.Header.Get(Compression) == GzipCompressionProtocol { + v.Compression = true + } + m, err := ws.cache.GetTokensWithoutError(context.Background(), v.UserID, platformID) if err != nil { - httpError(connContext, err) - return + return nil, err } - 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 + return nil, errs.ErrTokenKicked.Wrap() default: - httpError(connContext, errs.ErrTokenUnknown.Wrap()) - return + return nil, errs.ErrTokenUnknown.Wrap(fmt.Sprintf("token status is %d", v)) } } else { - httpError(connContext, errs.ErrTokenNotExist.Wrap()) - return + return nil, errs.ErrTokenNotExist.Wrap() } + return &v, nil +} - wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize) - err = wsLongConn.GenerateLongConn(w, r) - if err != nil { - httpError(connContext, err) - return - } - compressProtoc, exists := connContext.Query(Compression) - if exists { - if compressProtoc == GzipCompressionProtocol { - compression = true +type WSArgs struct { + Token string + UserID string + PlatformID int + Compression bool + MsgResp bool +} + +func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) { + 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 } - } - compressProtoc, exists = connContext.GetHeader(Compression) - if exists { - if compressProtoc == GzipCompressionProtocol { - compression = true + data, err := json.Marshal(apiresp.ParseError(pErr)) + if err != nil { + _ = wsLongConn.Close() + return + } + 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() } diff --git a/pkg/common/db/controller/group.go b/pkg/common/db/controller/group.go index 4147d59c0..decd868d6 100644 --- a/pkg/common/db/controller/group.go +++ b/pkg/common/db/controller/group.go @@ -197,7 +197,7 @@ func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data ma func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, deleteMember bool) error { return g.ctxTx.Transaction(ctx, func(ctx context.Context) error { c := g.cache.NewCache() - if err := g.groupDB.UpdateState(ctx, groupID, constant.GroupStatusDismissed); err != nil { + if err := g.groupDB.UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil { return err } if deleteMember { diff --git a/pkg/common/db/mgo/group.go b/pkg/common/db/mgo/group.go index 65dbbca59..a9c6d1eb8 100644 --- a/pkg/common/db/mgo/group.go +++ b/pkg/common/db/mgo/group.go @@ -49,8 +49,8 @@ func (g *GroupMgo) Create(ctx context.Context, groups []*relation.GroupModel) (e return mgoutil.InsertMany(ctx, g.coll, groups) } -func (g *GroupMgo) UpdateState(ctx context.Context, groupID string, state int32) (err error) { - return g.UpdateMap(ctx, groupID, map[string]any{"state": state}) +func (g *GroupMgo) UpdateStatus(ctx context.Context, groupID string, status int32) (err error) { + return g.UpdateMap(ctx, groupID, map[string]any{"status": status}) } func (g *GroupMgo) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) { diff --git a/pkg/common/db/mgo/group_member.go b/pkg/common/db/mgo/group_member.go index 8c3041901..8e3dd1efa 100644 --- a/pkg/common/db/mgo/group_member.go +++ b/pkg/common/db/mgo/group_member.go @@ -51,7 +51,11 @@ func (g *GroupMemberMgo) Create(ctx context.Context, groupMembers []*relation.Gr } func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []string) (err error) { - return mgoutil.DeleteMany(ctx, g.coll, bson.M{"group_id": groupID, "user_id": bson.M{"$in": userIDs}}) + filter := bson.M{"group_id": groupID} + if len(userIDs) > 0 { + filter["user_id"] = bson.M{"$in": userIDs} + } + return mgoutil.DeleteMany(ctx, g.coll, filter) } func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) error { @@ -84,8 +88,8 @@ func (g *GroupMemberMgo) FindRoleLevelUserIDs(ctx context.Context, groupID strin } func (g *GroupMemberMgo) SearchMember(ctx context.Context, keyword string, groupID string, pagination pagination.Pagination) (total int64, groupList []*relation.GroupMemberModel, err error) { - //TODO implement me - panic("implement me") + filter := bson.M{"group_id": groupID, "nickname": bson.M{"$regex": keyword}} + return mgoutil.FindPage[*relation.GroupMemberModel](ctx, g.coll, filter, pagination) } func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) { diff --git a/pkg/common/db/table/relation/group.go b/pkg/common/db/table/relation/group.go index bb1ddd878..57d6b1d62 100644 --- a/pkg/common/db/table/relation/group.go +++ b/pkg/common/db/table/relation/group.go @@ -42,7 +42,7 @@ type GroupModel struct { type GroupModelInterface interface { Create(ctx context.Context, groups []*GroupModel) (err error) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) - UpdateState(ctx context.Context, groupID string, state int32) (err error) + UpdateStatus(ctx context.Context, groupID string, status int32) (err error) Find(ctx context.Context, groupIDs []string) (groups []*GroupModel, err error) Take(ctx context.Context, groupID string) (group *GroupModel, err error) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*GroupModel, err error) diff --git a/pkg/common/discoveryregister/kubernetes/kubernetes.go b/pkg/common/discoveryregister/kubernetes/kubernetes.go index 2ff1539e6..c10518056 100644 --- a/pkg/common/discoveryregister/kubernetes/kubernetes.go +++ b/pkg/common/discoveryregister/kubernetes/kubernetes.go @@ -55,14 +55,17 @@ func (cli *K8sDR) Register(serviceName, host string, port int, opts ...grpc.Dial return nil } + func (cli *K8sDR) UnRegister() error { return nil } + func (cli *K8sDR) CreateRpcRootNodes(serviceNames []string) error { return nil } + func (cli *K8sDR) RegisterConf2Registry(key string, conf []byte) error { return nil @@ -123,6 +126,8 @@ func getMsgGatewayHost(ctx context.Context) []string { log.ZInfo(ctx, "getMsgGatewayHost", "instance", instance, "selfPodName", selfPodName, "replicas", replicas, "ns", ns, "ret", ret) return ret } + +// GetConns returns the gRPC client connections to the specified service. func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) { if serviceName != config.Config.RpcRegisterName.OpenImMessageGatewayName { @@ -142,6 +147,7 @@ func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc return ret, nil } } + func (cli *K8sDR) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { return grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...) @@ -151,9 +157,11 @@ func (cli *K8sDR) GetSelfConnTarget() string { return cli.rpcRegisterAddr } + func (cli *K8sDR) AddOption(opts ...grpc.DialOption) { cli.options = append(cli.options, opts...) } + func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) { conn.Close() } diff --git a/scripts/docker-start-all.sh b/scripts/docker-start-all.sh index 3c83b02bb..85954a677 100755 --- a/scripts/docker-start-all.sh +++ b/scripts/docker-start-all.sh @@ -21,6 +21,25 @@ set -o pipefail #fixme This scripts is the total startup scripts #fixme The full name of the shell scripts that needs to be started is placed in the need_to_start_server_shell array +# Fixed ports inside the docker startup container +export OPENIM_WS_PORT=10001 +export API_OPENIM_PORT=10002 +export API_PROM_PORT=20100 +export USER_PROM_PORT=20110 +export FRIEND_PROM_PORT=20120 +export MESSAGE_PROM_PORT=20130 +export MSG_GATEWAY_PROM_PORT=20140 +export GROUP_PROM_PORT=20150 +export AUTH_PROM_PORT=20160 +export PUSH_PROM_PORT=20170 +export CONVERSATION_PROM_PORT=20230 +export RTC_PROM_PORT=21300 +export THIRD_PROM_PORT=21301 +export MSG_TRANSFER_PROM_PORT=21400 +export MSG_TRANSFER_PROM_PORT=21401 +export MSG_TRANSFER_PROM_PORT=21402 +export MSG_TRANSFER_PROM_PORT=21403 + OPENIM_ROOT=$(dirname "${BASH_SOURCE[0]}")/.. source "${OPENIM_ROOT}/scripts/install/common.sh" diff --git a/scripts/install-im-server.sh b/scripts/install-im-server.sh index 47db34433..a21ae134d 100755 --- a/scripts/install-im-server.sh +++ b/scripts/install-im-server.sh @@ -43,10 +43,20 @@ fi "${OPENIM_ROOT}"/scripts/init-config.sh 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 +curl https://raw.githubusercontent.com/openimsdk/openim-docker/main/docker-compose.yaml -o docker-compose.yml ${DOCKER_COMPOSE_COMMAND} up -d -sleep 60 + +# Wait for a short period to allow containers to initialize +sleep 30 + +# 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 +fi + +sleep 30 # Keep the original 60-second wait, adjusted for the 10-second check above ${DOCKER_COMPOSE_COMMAND} logs openim-server ${DOCKER_COMPOSE_COMMAND} ps -popd \ No newline at end of file +popd