msggateway

package
v3.3.0-rc.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 7, 2023 License: Apache-2.0 Imports: 39 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// MessageText is for UTF-8 encoded text messages like JSON.
	MessageText = iota + 1
	// MessageBinary is for binary messages like protobufs.
	MessageBinary
	// CloseMessage denotes a close control message. The optional message
	// payload contains a numeric code and text. Use the FormatCloseMessage
	// function to format a close message payload.
	CloseMessage = 8

	// PingMessage denotes a ping control message. The optional message payload
	// is UTF-8 encoded text.
	PingMessage = 9

	// PongMessage denotes a pong control message. The optional message payload
	// is UTF-8 encoded text.
	PongMessage = 10
)
View Source 
const (
	WsUserID                = "sendID"
	CommonUserID            = "userID"
	PlatformID              = "platformID"
	ConnID                  = "connID"
	Token                   = "token"
	OperationID             = "operationID"
	Compression             = "compression"
	GzipCompressionProtocol = "gzip"
	BackgroundStatus        = "isBackground"
)
View Source 
const (
	// Websocket Protocol.
	WSGetNewestSeq        = 1001
	WSPullMsgBySeqList    = 1002
	WSSendMsg             = 1003
	WSSendSignalMsg       = 1004
	WSPushMsg             = 2001
	WSKickOnlineMsg       = 2002
	WsLogoutMsg           = 2003
	WsSetBackgroundStatus = 2004
	WSDataError           = 3001
)
View Source 
const (
	WebSocket = iota + 1
)

Variables

View Source 
var (
	ErrConnClosed                = errors.New("conn has closed")
	ErrNotSupportMessageProtocol = errors.New("not support message protocol")
	ErrClientClosed              = errors.New("client actively close the connection")
	ErrPanic                     = errors.New("panic error")
)

Functions

func CallbackUserKickOff 

func CallbackUserKickOff(ctx context.Context, userID string, platformID int) error

func CallbackUserOffline 

func CallbackUserOffline(ctx context.Context, userID string, platformID int, connID string) error

func CallbackUserOnline 

func CallbackUserOnline(ctx context.Context, userID string, platformID int, isAppBackground bool, connID string) error

func RunWsAndServer 

func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error

Types

type Client 

type Client struct {
	PlatformID   int    `json:"platformID"`
	IsCompress   bool   `json:"isCompress"`
	UserID       string `json:"userID"`
	IsBackground bool   `json:"isBackground"`
	// contains filtered or unexported fields
}

func (*Client) KickOnlineMessage 

func (c *Client) KickOnlineMessage() error

func (*Client) PushMessage 

func (c *Client) PushMessage(ctx context.Context, msgData *sdkws.MsgData) error

func (*Client) ResetClient 

func (c *Client) ResetClient(
	ctx *UserConnContext,
	conn LongConn,
	isBackground, isCompress bool,
	longConnServer LongConnServer,
	token string,
)

type Compressor 

type Compressor interface {
	Compress(rawData []byte) ([]byte, error)
	DeCompress(compressedData []byte) ([]byte, error)
}

type Encoder 

type Encoder interface {
	Encode(data interface{}) ([]byte, error)
	Decode(encodeData []byte, decodeData interface{}) error
}

type GWebSocket 

type GWebSocket struct {
	// contains filtered or unexported fields
}

func (*GWebSocket) Close 

func (d *GWebSocket) Close() error

func (*GWebSocket) Dial 

func (d *GWebSocket) Dial(urlStr string, requestHeader http.Header) (*http.Response, error)

func (*GWebSocket) GenerateLongConn 

func (d *GWebSocket) GenerateLongConn(w http.ResponseWriter, r *http.Request) error

func (*GWebSocket) IsNil 

func (d *GWebSocket) IsNil() bool

func (*GWebSocket) ReadMessage 

func (d *GWebSocket) ReadMessage() (int, []byte, error)

func (*GWebSocket) SetConnNil 

func (d *GWebSocket) SetConnNil()

func (*GWebSocket) SetPongHandler 

func (d *GWebSocket) SetPongHandler(handler PongHandler)

func (*GWebSocket) SetReadDeadline 

func (d *GWebSocket) SetReadDeadline(timeout time.Duration) error

func (*GWebSocket) SetReadLimit 

func (d *GWebSocket) SetReadLimit(limit int64)

func (*GWebSocket) SetWriteDeadline 

func (d *GWebSocket) SetWriteDeadline(timeout time.Duration) error

func (*GWebSocket) WriteMessage 

func (d *GWebSocket) WriteMessage(messageType int, message []byte) error

type GobEncoder 

type GobEncoder struct{}

func NewGobEncoder 

func NewGobEncoder() *GobEncoder

func (*GobEncoder) Decode 

func (g *GobEncoder) Decode(encodeData []byte, decodeData interface{}) error

func (*GobEncoder) Encode 

func (g *GobEncoder) Encode(data interface{}) ([]byte, error)

type GrpcHandler 

type GrpcHandler struct {
	// contains filtered or unexported fields
}

func NewGrpcHandler 

func NewGrpcHandler(validate *validator.Validate, client discoveryregistry.SvcDiscoveryRegistry) *GrpcHandler

func (GrpcHandler) GetSeq 

func (g GrpcHandler) GetSeq(context context.Context, data Req) ([]byte, error)

func (GrpcHandler) PullMessageBySeqList 

func (g GrpcHandler) PullMessageBySeqList(context context.Context, data Req) ([]byte, error)

func (GrpcHandler) SendMessage 

func (g GrpcHandler) SendMessage(context context.Context, data Req) ([]byte, error)

func (GrpcHandler) SendSignalMessage 

func (g GrpcHandler) SendSignalMessage(context context.Context, data Req) ([]byte, error)

func (GrpcHandler) SetUserDeviceBackground 

func (g GrpcHandler) SetUserDeviceBackground(_ context.Context, data Req) ([]byte, bool, error)

func (GrpcHandler) UserLogout 

func (g GrpcHandler) UserLogout(context context.Context, data Req) ([]byte, error)

type GzipCompressor 

type GzipCompressor struct {
	// contains filtered or unexported fields
}

func NewGzipCompressor 

func NewGzipCompressor() *GzipCompressor

func (*GzipCompressor) Compress 

func (g *GzipCompressor) Compress(rawData []byte) ([]byte, error)

func (*GzipCompressor) DeCompress 

func (g *GzipCompressor) DeCompress(compressedData []byte) ([]byte, error)

type LongConn 

type LongConn interface {
	// Close this connection
	Close() error
	// WriteMessage Write message to connection,messageType means data type,can be set binary(2) and text(1).
	WriteMessage(messageType int, message []byte) error
	// ReadMessage Read message from connection.
	ReadMessage() (int, []byte, error)
	// SetReadDeadline sets the read deadline on the underlying network connection,
	// after a read has timed out, will return an error.
	SetReadDeadline(timeout time.Duration) error
	// SetWriteDeadline sets to write deadline when send message,when read has timed out,will return error.
	SetWriteDeadline(timeout time.Duration) error
	// Dial Try to dial a connection,url must set auth args,header can control compress data
	Dial(urlStr string, requestHeader http.Header) (*http.Response, error)
	// IsNil Whether the connection of the current long connection is nil
	IsNil() bool
	// SetConnNil Set the connection of the current long connection to nil
	SetConnNil()
	// SetReadLimit sets the maximum size for a message read from the peer.bytes
	SetReadLimit(limit int64)
	SetPongHandler(handler PongHandler)
	// GenerateLongConn Check the connection of the current and when it was sent are the same
	GenerateLongConn(w http.ResponseWriter, r *http.Request) error
}

type LongConnServer 

type LongConnServer interface {
	Run() error

	GetUserAllCons(userID string) ([]*Client, bool)
	GetUserPlatformCons(userID string, platform int) ([]*Client, bool, bool)
	Validate(s interface{}) error
	SetCacheHandler(cache cache.MsgModel)
	SetDiscoveryRegistry(client discoveryregistry.SvcDiscoveryRegistry)
	KickUserConn(client *Client) error
	UnRegister(c *Client)
	SetKickHandlerInfo(i *kickHandler)
	Compressor
	Encoder
	MessageHandler
	// contains filtered or unexported methods
}

type MessageHandler 

type MessageHandler interface {
	GetSeq(context context.Context, data Req) ([]byte, error)
	SendMessage(context context.Context, data Req) ([]byte, error)
	SendSignalMessage(context context.Context, data Req) ([]byte, error)
	PullMessageBySeqList(context context.Context, data Req) ([]byte, error)
	UserLogout(context context.Context, data Req) ([]byte, error)
	SetUserDeviceBackground(context context.Context, data Req) ([]byte, bool, error)
}

type Option 

type Option func(opt *configs)

func WithHandshakeTimeout 

func WithHandshakeTimeout(t time.Duration) Option

func WithMaxConnNum 

func WithMaxConnNum(num int64) Option

func WithMessageMaxMsgLength 

func WithMessageMaxMsgLength(length int) Option

func WithPort 

func WithPort(port int) Option

type PongHandler 

type PongHandler func(string) error

type Req 

type Req struct {
	ReqIdentifier int32  `json:"reqIdentifier" validate:"required"`
	Token         string `json:"token"`
	SendID        string `json:"sendID"        validate:"required"`
	OperationID   string `json:"operationID"   validate:"required"`
	MsgIncr       string `json:"msgIncr"       validate:"required"`
	Data          []byte `json:"data"`
}

func (*Req) String 

func (r *Req) String() string

type Resp 

type Resp struct {
	ReqIdentifier int32  `json:"reqIdentifier"`
	MsgIncr       string `json:"msgIncr"`
	OperationID   string `json:"operationID"`
	ErrCode       int    `json:"errCode"`
	ErrMsg        string `json:"errMsg"`
	Data          []byte `json:"data"`
}

func (*Resp) String 

func (r *Resp) String() string

type Server 

type Server struct {
	LongConnServer LongConnServer
	// contains filtered or unexported fields
}

func NewServer 

func NewServer(rpcPort int, longConnServer LongConnServer) *Server

func (*Server) GetUsersOnlineStatus 

func (s *Server) GetUsersOnlineStatus(
	ctx context.Context,
	req *msggateway.GetUsersOnlineStatusReq,
) (*msggateway.GetUsersOnlineStatusResp, error)

func (*Server) InitServer 

func (s *Server) InitServer(disCov discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error

func (*Server) KickUserOffline 

func (s *Server) KickUserOffline(
	ctx context.Context,
	req *msggateway.KickUserOfflineReq,
) (*msggateway.KickUserOfflineResp, error)

func (*Server) MultiTerminalLoginCheck 

func (s *Server) MultiTerminalLoginCheck(
	ctx context.Context,
	req *msggateway.MultiTerminalLoginCheckReq,
) (*msggateway.MultiTerminalLoginCheckResp, error)

func (*Server) OnlineBatchPushOneMsg 

func (s *Server) OnlineBatchPushOneMsg(
	ctx context.Context,
	req *msggateway.OnlineBatchPushOneMsgReq,
) (*msggateway.OnlineBatchPushOneMsgResp, error)

func (*Server) OnlinePushMsg 

func (s *Server) OnlinePushMsg(
	context context.Context,
	req *msggateway.OnlinePushMsgReq,
) (*msggateway.OnlinePushMsgResp, error)

func (*Server) SetLongConnServer 

func (s *Server) SetLongConnServer(LongConnServer LongConnServer)

func (*Server) Start 

func (s *Server) Start() error

func (*Server) SuperGroupOnlineBatchPushOneMsg 

func (s *Server) SuperGroupOnlineBatchPushOneMsg(
	ctx context.Context,
	req *msggateway.OnlineBatchPushOneMsgReq,
) (*msggateway.OnlineBatchPushOneMsgResp, error)

type UserConnContext 

type UserConnContext struct {
	RespWriter http.ResponseWriter
	Req        *http.Request
	Path       string
	Method     string
	RemoteAddr string
	ConnID     string
}

func (*UserConnContext) Deadline 

func (c *UserConnContext) Deadline() (deadline time.Time, ok bool)

func (*UserConnContext) Done 

func (c *UserConnContext) Done() <-chan struct{}

func (*UserConnContext) Err 

func (c *UserConnContext) Err() error

func (*UserConnContext) ErrReturn 

func (c *UserConnContext) ErrReturn(error string, code int)

func (*UserConnContext) GetBackground 

func (c *UserConnContext) GetBackground() bool

func (*UserConnContext) GetConnID 

func (c *UserConnContext) GetConnID() string

func (*UserConnContext) GetHeader 

func (c *UserConnContext) GetHeader(key string) (string, bool)

func (*UserConnContext) GetOperationID 

func (c *UserConnContext) GetOperationID() string

func (*UserConnContext) GetPlatformID 

func (c *UserConnContext) GetPlatformID() string

func (*UserConnContext) GetRemoteAddr 

func (c *UserConnContext) GetRemoteAddr() string

func (*UserConnContext) GetToken 

func (c *UserConnContext) GetToken() string

func (*UserConnContext) GetUserID 

func (c *UserConnContext) GetUserID() string

func (*UserConnContext) Query 

func (c *UserConnContext) Query(key string) (string, bool)

func (*UserConnContext) SetHeader 

func (c *UserConnContext) SetHeader(key, value string)

func (*UserConnContext) SetOperationID 

func (c *UserConnContext) SetOperationID(operationID string)

func (*UserConnContext) SetToken 

func (c *UserConnContext) SetToken(token string)

func (*UserConnContext) Value 

func (c *UserConnContext) Value(key any) any

type UserMap 

type UserMap struct {
	// contains filtered or unexported fields
}

func (*UserMap) DeleteAll 

func (u *UserMap) DeleteAll(key string)

func (*UserMap) Get 

func (u *UserMap) Get(key string, platformID int) ([]*Client, bool, bool)

func (*UserMap) GetAll 

func (u *UserMap) GetAll(key string) ([]*Client, bool)

func (*UserMap) Set 

func (u *UserMap) Set(key string, v *Client)

type WsServer 

type WsServer struct {
	Compressor
	Encoder
	MessageHandler
	// contains filtered or unexported fields
}

func NewWsServer 

func NewWsServer(opts ...Option) (*WsServer, error)

func (*WsServer) GetUserAllCons 

func (ws *WsServer) GetUserAllCons(userID string) ([]*Client, bool)

func (*WsServer) GetUserPlatformCons 

func (ws *WsServer) GetUserPlatformCons(userID string, platform int) ([]*Client, bool, bool)

func (*WsServer) KickUserConn 

func (ws *WsServer) KickUserConn(client *Client) error

func (*WsServer) Run 

func (ws *WsServer) Run() error

func (*WsServer) SetCacheHandler 

func (ws *WsServer) SetCacheHandler(cache cache.MsgModel)

func (*WsServer) SetDiscoveryRegistry 

func (ws *WsServer) SetDiscoveryRegistry(disCov discoveryregistry.SvcDiscoveryRegistry)

func (*WsServer) SetKickHandlerInfo 

func (ws *WsServer) SetKickHandlerInfo(i *kickHandler)

func (*WsServer) SetUserOnlineStatus 

func (ws *WsServer) SetUserOnlineStatus(ctx context.Context, client *Client, status int32)

func (*WsServer) UnRegister 

func (ws *WsServer) UnRegister(c *Client)

func (*WsServer) Validate 

func (ws *WsServer) Validate(s interface{}) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.