connect/server/hub.go

package server

import (
	"sync"

	"sikey.com/websocket/stackexchange/redis"
	"sikey.com/websocket/utils/zlog"
)

type HubConfig struct {
	ConnectSize    int
	DisconnectSize int
	MessageSize    int

	// StackExchange *stackexchange.RedisStackExchange
}

type Hub struct {
	remotelyClients map[string]*Client
	remotelyMutex   sync.RWMutex
	clients         map[string]*Client
	mutex           sync.RWMutex

	Connect    chan *Client
	Disconnect chan *Client

	Message chan *Message

	StackExchange *redis.StackExchange
}

func NewHub(cfg HubConfig) *Hub {
	hub := &Hub{
		remotelyClients: make(map[string]*Client),
		remotelyMutex:   sync.RWMutex{},
		clients:         make(map[string]*Client),
		mutex:           sync.RWMutex{},

		Connect:    make(chan *Client, cfg.ConnectSize),
		Disconnect: make(chan *Client, cfg.DisconnectSize),
		Message:    make(chan *Message, cfg.MessageSize),
	}

	hub.StackExchange = redis.NewStackExchange()

	go hub.run()
	go hub.remotely() // 远程消息
	return hub
}

func (h *Hub) run() {
	for {
		select {
		case client := <-h.Connect:
			if client.isRemotely {
				h.remotelyClients[client.UserId] = client
			} else {
				h.clients[client.UserId] = client
			}
		case client := <-h.Disconnect:
			close(client.Send)
			if client.isRemotely {
				delete(h.remotelyClients, client.UserId)
			} else {
				delete(h.clients, client.UserId)
			}
		case message := <-h.Message:
			if client, ok := h.clients[message.receiver]; ok {
				client.Send <- message
			} else {
				// 不在同一台服务器, 这里将消息发送至拓展应用
				if client != nil {
					client.RemotelyMessage <- message
				}
			}
		}
	}
}

func (h *Hub) remotely() {
	for {

		zlog.Info("Remotely client actions monitoring.")
		// km, err := h.exchange.ReadMessage()
		// if err != nil {
		// 	zlog.Error(err)
		// }

		// // Create new a remote client
		// remotelyClient := &Client{
		// 	isRemotely: true, // mark remotely client
		// 	UserId:     string(km.Value),
		// 	Send:       make(chan *Message, 256),
		// 	hub:        h,
		// }
		// switch {
		// case bytes.Equal(km.Key, stackexchange.StackExchangeOnline):
		// 	h.Connect <- remotelyClient
		// 	zlog.Info("Remotely client online: ", remotelyClient.UserId)
		// case bytes.Equal(km.Key, stackexchange.StackExchangeOffline):
		// 	h.Disconnect <- remotelyClient
		// 	zlog.Info("Remotely client offline: ", remotelyClient.UserId)
		// case bytes.Equal(km.Key, stackexchange.StackExchangeMessaging):
		// 	h.Message <- deserializeMessage(km.Value)
		// }
	}
}