ws

Documentation

Index

• Constants
• type Channel
  • func NewChannel(id string) Channel
  • func (c *Channel) Write(msg Message) (err error)
• type Client
  • func NewClient(uid string, conn *websocket.Conn) *Client
  • func (c *Client) JoinChannel(ch *Channel)
  • func (c *Client) LeaveChannel(ch *Channel)
  • func (c *Client) Read() (Message, error)
  • func (c *Client) Write(msg Message) error
• type ClientToServerMap
  • func (m *ClientToServerMap) Del(key string)
  • func (m *ClientToServerMap) Get(key string) (string, error)
  • func (m *ClientToServerMap) Set(key, val string) error
• type Message
  • func NewMessage(t MessageType, data []byte, from, to string) Message
• type MessageQueue
  • func NewMessageQueue(id string, ch *amqp.Channel) MessageQueue
  • func (mq *MessageQueue) Consume() (msgs chan Message, err error)
  • func (mq *MessageQueue) Publish(wid string, msg Message)
• type MessageType
• type WebSocketManager
  • func NewWebSocketManager(id string, cap int, rdb *redis.Client, mq MessageQueue) *WebSocketManager
  • func (m *WebSocketManager) Broadcast(msg Message) (err error)
  • func (m *WebSocketManager) HandleMessage()
  • func (m *WebSocketManager) IsManaged(uid string) bool
  • func (m *WebSocketManager) Logout(uid string)
  • func (m *WebSocketManager) PushMessage(msg Message)
  • func (m *WebSocketManager) ReceiveMessage()
  • func (m *WebSocketManager) Register(client *Client)
  • func (m *WebSocketManager) SendMessage(msg Message) error
• type WebSocketManagerError
  • func (e WebSocketManagerError) Error() string

Constants

const (
	ExchangeName = "websocket-messages-router"
	ExchangeType = "direct"

	QueueName = "websocket-messages-queue"
)

const WebSocketClientToServerMap = "websocket-client-to-server-map"

Types

type Channel

type Channel struct {
	ID      string             // 群组ID
	Members map[string]*Client // 群组成员 UID => WebSocket客户端
}

Channel websocket频道

func NewChannel

func NewChannel(id string) Channel

NewChannel 实例化频道

func (*Channel) Write

func (c *Channel) Write(msg Message) (err error)

Write 向频道中写入消息

type Client

type Client struct {
	UID  string          // 用户ID
	Conn *websocket.Conn // websocket连接
}

Client websocket客户端

func NewClient

func NewClient(uid string, conn *websocket.Conn) *Client

NewClient 实例化一个 WebSocket 客户端

func (*Client) JoinChannel

func (c *Client) JoinChannel(ch *Channel)

JoinChannel 加入频道

func (*Client) LeaveChannel

func (c *Client) LeaveChannel(ch *Channel)

LeaveChannel 离开频道

func (*Client) Read

func (c *Client) Read() (Message, error)

Read 接收消息

func (*Client) Write

func (c *Client) Write(msg Message) error

Write 向客户端中写入信息

type ClientToServerMap

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

func (*ClientToServerMap) Del

func (m *ClientToServerMap) Del(key string)

func (*ClientToServerMap) Get

func (m *ClientToServerMap) Get(key string) (string, error)

func (*ClientToServerMap) Set

func (m *ClientToServerMap) Set(key, val string) error

type Message

type Message struct {
	Type      MessageType `json:"type"`      // 消息类型
	Content   []byte      `json:"content"`   // 消息内容
	From      string      `json:"from"`      // 来源
	To        string      `json:"to"`        // 去向
	Timestamp int64       `json:"timestamp"` // 时间戳
}

Message websocket 消息

func NewMessage

func NewMessage(t MessageType, data []byte, from, to string) Message

NewMessage 实例化一条消息

type MessageQueue

type MessageQueue struct {
	Queue   *amqp.Queue
	Channel *amqp.Channel
}

func NewMessageQueue

func NewMessageQueue(id string, ch *amqp.Channel) MessageQueue

NewMessageQueue 实例化 websocket 消息队列

func (*MessageQueue) Consume

func (mq *MessageQueue) Consume() (msgs chan Message, err error)

Consume 消费消息

func (*MessageQueue) Publish

func (mq *MessageQueue) Publish(wid string, msg Message)

Publish 发送消息 wid => websocket id;

type MessageType

type MessageType uint8

const (
	MessageTypeRegister  MessageType = iota + 1 // 注册
	MessageTypeLogout                           // 注销
	MessageTypeHeartbeat                        // 心跳
	MessageTypeOneOnOne                         // 一对一消息
	MessageTypeGroup                            // 群信息
	MessageTypeChannel                          // 频道消息
	MessageTypeBroadcast                        // 广播消息
	MessageTypeEcho                             // 广播消息
)

type WebSocketManager

type WebSocketManager struct {
	ID                string
	Clients           map[string]*Client
	Channels          map[string]*Channel
	Messages          chan Message      // 消息处理通道
	MessageQueue      MessageQueue      // 当前服务器独占的消息队列 用于接收来自其他服务器发送的 websocket 消息
	ClientToServerMap ClientToServerMap // 客户端 到 管理器的映射 服务于消息队列
	Lock              sync.RWMutex
}

WebSocketManager websocket管理器

func NewWebSocketManager

func NewWebSocketManager(id string, cap int, rdb *redis.Client, mq MessageQueue) *WebSocketManager

NewWebSocketManager 实例化websocket管理器 cap: 消息通道容量

func (*WebSocketManager) Broadcast

func (m *WebSocketManager) Broadcast(msg Message) (err error)

Broadcast 广播消息

func (*WebSocketManager) HandleMessage

func (m *WebSocketManager) HandleMessage()

HandleMessage 处理消息

func (*WebSocketManager) IsManaged

func (m *WebSocketManager) IsManaged(uid string) bool

IsManaged 判断用户是否被 websocket 管理器 所管理

func (*WebSocketManager) Logout

func (m *WebSocketManager) Logout(uid string)

Logout 取消 WebSocketManager 对 client 的管理 & 从所有频道中移除该客户端

func (*WebSocketManager) PushMessage

func (m *WebSocketManager) PushMessage(msg Message)

func (*WebSocketManager) ReceiveMessage

func (m *WebSocketManager) ReceiveMessage()

ReceiveMessage 接收消息

func (*WebSocketManager) Register

func (m *WebSocketManager) Register(client *Client)

Register 使用 WebSocketManager 对 client 进行管理 & 接收客户端发送过来的所有消息

func (*WebSocketManager) SendMessage

func (m *WebSocketManager) SendMessage(msg Message) error

SendMessage 发送消息

type WebSocketManagerError

type WebSocketManagerError string

const ManagerNotExist WebSocketManagerError = "manager not exist"

func (WebSocketManagerError) Error

func (e WebSocketManagerError) Error() string