fix: fix the conflect

pull/1663/head
luhaoling 2 years ago
commit a1c4be3c84

@ -3,12 +3,36 @@
before: before:
hooks: hooks:
- make clean
# You may remove this if you don't use go modules. # You may remove this if you don't use go modules.
- make tidy - make tidy
- make copyright.add - make copyright.add
# you may remove this if you don't need go generate # you may remove this if you don't need go generate
- 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: snapshot:
name_template: "{{ incpatch .Version }}-next" name_template: "{{ incpatch .Version }}-next"
@ -495,4 +519,4 @@ checksum:
algorithm: sha256 algorithm: sha256
release: release:
prerelease: auto prerelease: auto

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

@ -16,7 +16,10 @@ package msggateway
import ( import (
"context" "context"
"encoding/json"
"errors" "errors"
"fmt"
"github.com/OpenIMSDK/tools/apiresp"
"net/http" "net/http"
"os" "os"
"os/signal" "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 { 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) platformID, err := strconv.Atoi(platformIDStr)
if err != nil { 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 { 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 { switch v {
case constant.NormalToken: case constant.NormalToken:
case constant.KickedToken: case constant.KickedToken:
httpError(connContext, errs.ErrTokenKicked.Wrap()) return nil, errs.ErrTokenKicked.Wrap()
return
default: default:
httpError(connContext, errs.ErrTokenUnknown.Wrap()) return nil, errs.ErrTokenUnknown.Wrap(fmt.Sprintf("token status is %d", v))
return
} }
} else { } 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 := 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 ws.registerChan <- client
go client.readMessage() go client.readMessage()
} }

@ -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 { func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, deleteMember bool) error {
return g.ctxTx.Transaction(ctx, func(ctx context.Context) error { return g.ctxTx.Transaction(ctx, func(ctx context.Context) error {
c := g.cache.NewCache() 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 return err
} }
if deleteMember { if deleteMember {

@ -49,8 +49,8 @@ func (g *GroupMgo) Create(ctx context.Context, groups []*relation.GroupModel) (e
return mgoutil.InsertMany(ctx, g.coll, groups) return mgoutil.InsertMany(ctx, g.coll, groups)
} }
func (g *GroupMgo) UpdateState(ctx context.Context, groupID string, state int32) (err error) { func (g *GroupMgo) UpdateStatus(ctx context.Context, groupID string, status int32) (err error) {
return g.UpdateMap(ctx, groupID, map[string]any{"state": state}) 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) { func (g *GroupMgo) UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error) {

@ -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) { 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 { 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) { func (g *GroupMemberMgo) SearchMember(ctx context.Context, keyword string, groupID string, pagination pagination.Pagination) (total int64, groupList []*relation.GroupMemberModel, err error) {
//TODO implement me filter := bson.M{"group_id": groupID, "nickname": bson.M{"$regex": keyword}}
panic("implement me") return mgoutil.FindPage[*relation.GroupMemberModel](ctx, g.coll, filter, pagination)
} }
func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) { func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) {

@ -42,7 +42,7 @@ type GroupModel struct {
type GroupModelInterface interface { type GroupModelInterface interface {
Create(ctx context.Context, groups []*GroupModel) (err error) Create(ctx context.Context, groups []*GroupModel) (err error)
UpdateMap(ctx context.Context, groupID string, args map[string]any) (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) Find(ctx context.Context, groupIDs []string) (groups []*GroupModel, err error)
Take(ctx context.Context, groupID string) (group *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) Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*GroupModel, err error)

@ -45,8 +45,19 @@ pushd "${OPENIM_ROOT}"
${DOCKER_COMPOSE_COMMAND} stop ${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://gitee.com/openimsdk/openim-docker/raw/main/example/full-openim-server-and-chat.yml -o docker-compose.yml
${DOCKER_COMPOSE_COMMAND} up -d ${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} logs openim-server
${DOCKER_COMPOSE_COMMAND} ps ${DOCKER_COMPOSE_COMMAND} ps
popd popd

Loading…
Cancel
Save