@ -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. U serID, platformID )
if err != nil {
if err != nil {
httpError ( connContext , err )
return nil , err
return
}
}
if v , ok := m [ t oken] ; ok {
if v , ok := m [ v. T oken] ; 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 . T oken)
ws . registerChan <- client
ws . registerChan <- client
go client . readMessage ( )
go client . readMessage ( )
}
}