redis 拓展 websocket

config/redis.go

@@ -0,0 +1,8 @@
+package config
+
+type redis struct {
+	Addr     string
+	Channel  string
+	Db       int
+	Password string
+}

server/hub.go

@@ -48,8 +48,6 @@ func NewHub(cfg HubConfig) *Hub {
 	go hub.run()
 
 	go hub.remotelyEvent() // 远程事件
-	//go hub.remotelyMessageReader()
-	//go hub.remotelyMessageWriter()
 	return hub
 }
 
@@ -58,9 +56,11 @@ func (h *Hub) run() {
 		select {
 		case client := <-h.Connect:
 			h.clients[client.UserId] = client
+			h.OnPublishConnect(context.Background(), client)
 		case client := <-h.Disconnect:
 			close(client.Send)
 			delete(h.clients, client.UserId)
+			h.OnPublishDisconnect(context.Background(), client)
 		case message := <-h.Message:
 			if client, ok := h.clients[message.receiver]; ok {
 				if client.isRemotely {
@@ -115,3 +115,11 @@ func (h *Hub) remotelyEvent() {
 		fmt.Println(rMsg)
 	}
 }
+
+func (h *Hub) OnPublishConnect(ctx context.Context, client *Client) error {
+	return h.rdb.Publish(ctx, connectChannelEvent, client.UserId).Err()
+}
+
+func (h *Hub) OnPublishDisconnect(ctx context.Context, client *Client) error {
+	return h.rdb.Publish(ctx, disconnectChannelEvent, client.UserId).Err()
+}

server/redis_exchange.go

@@ -1,10 +1,12 @@
 package server
 
-type exchange struct {
-	eventChannel   string
-	messageChannel string
+type Exchange struct {
 }
 
-func NewExchange() *exchange {
-	return &exchange{}
+func NewExchange() *Exchange {
+	return &Exchange{}
+}
+
+func (exc *Exchange) SendMessage(message *Message) {
+
 }

server/server.go

@@ -62,8 +62,7 @@ func WebsocketHandler(ctx *gin.Context, srv *Server) {
 
 		repos: srv.Repositories,
 	}
-	client.hub.Connect <- client
-	// client.hub.exchange.OnlineNotify(client.UserId)
+	srv.Hub.Connect <- client
 	zlog.Info("client online: ", client.UserId)
 
 	go client.reader()