init: 初始化 web_hgdjb_server 项目

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

pub/common/comm/comm.go

@@ -0,0 +1,138 @@
+package comm
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+	"videoweb/pub/common/godef"
+	"videoweb/pub/common/lock"
+	"videoweb/pub/common/log"
+
+	"github.com/go-redis/redis"
+	"go.uber.org/zap"
+)
+
+type MessageType int64
+
+const (
+	MessageTypeError MessageType = iota // 心跳消息
+	MessageTypeAlive
+	MessageTypeMediaAdd // 添加一个媒体
+	MessageTypeMediaDel // 删除一个媒体
+	MessageTypePostAdd  // 添加一个帖子
+	MessageTypePostDel  // 删除一个帖子
+)
+
+type message struct {
+	Type    MessageType
+	Payload []byte
+}
+
+type OnReceive func(data []byte)
+
+type Communicate struct {
+	sendChannel string        // 发送消息的频道
+	recvChannel string        // 接收消息的频道
+	client      *redis.Client // Redis的客户端
+	lock        sync.RWMutex
+	callbacks   map[MessageType][]OnReceive
+	publish     *lock.Queue[*message]
+}
+
+func NewCommunicate(client *redis.Client, sendChannel string, recvChannel string) *Communicate {
+	return &Communicate{
+		sendChannel: sendChannel,
+		recvChannel: recvChannel,
+		client:      client,
+		callbacks:   make(map[MessageType][]OnReceive),
+		publish:     lock.NewQueue[*message](),
+	}
+}
+
+func (c *Communicate) RegisterReceiver(mType MessageType, receive OnReceive) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.callbacks[mType] = append(c.callbacks[mType], receive)
+}
+
+func (c *Communicate) findEvent(mType MessageType) []OnReceive {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	return c.callbacks[mType]
+}
+
+func (c *Communicate) dispatchMessage(mType MessageType, payload []byte) {
+	events := c.findEvent(mType)
+	for _, event := range events {
+		event(payload)
+	}
+}
+
+func (c *Communicate) handleRecv(data *redis.Message) {
+	if data.Channel == "error" {
+		log.Error("recv from channel %s failed", zap.Any("error", data.Payload))
+		return
+	}
+	if data.Channel == c.recvChannel {
+		msg := &message{}
+		err := json.Unmarshal([]byte(data.Payload), msg)
+		if err != nil {
+			log.Error("recv from channel %s failed", zap.Any("error", err))
+		}
+		c.dispatchMessage(msg.Type, msg.Payload)
+	}
+}
+
+func (c *Communicate) Send(mType MessageType, payload []byte) error {
+	msg := &message{
+		Type:    mType,
+		Payload: payload,
+	}
+	c.publish.Enqueue(msg)
+	return nil
+}
+
+func (c *Communicate) Start() {
+	// 启动接收线程
+	godef.Go(func() {
+		sub := c.client.Subscribe(c.recvChannel)
+		defer sub.Close()
+		// 循环接收消息
+		for {
+			msg, err := sub.ReceiveMessage()
+			if err != nil {
+				msg = &redis.Message{
+					Channel: "error",
+					Payload: err.Error(),
+				}
+				c.handleRecv(msg)
+				time.Sleep(time.Second)
+			} else {
+				c.handleRecv(msg)
+			}
+		}
+	})
+	// 启动发送线程
+	godef.Go(func() {
+		for {
+			time.Sleep(time.Second)
+			for !c.publish.IsEmpty() {
+				msg, ok := c.publish.Dequeue()
+				if !ok {
+					break
+				}
+				data, err := json.Marshal(msg)
+				if err != nil {
+					log.Error("Communicate msg marshal failed", zap.Any("error", err))
+					break
+				}
+				err = c.client.Publish(c.sendChannel, data).Err()
+				if err != nil {
+					c.publish.Enqueue(msg) // 再写回队列
+					log.Error("Communicate send msg failed", zap.Any("error", err))
+					break
+				}
+			}
+		}
+	})
+}