保存进度

Signed-off-by: allan716 <[email protected]>

internal/backend/backend.go

@@ -41,7 +41,6 @@ func StartBackEnd(httpPort int, cronHelper *cron_helper.CronHelper) {
 	hub := ws_helper.NewHub()
 	go hub.Run()
 
-	// middle.CheckAuth(),
 	engine.GET("/ws", func(context *gin.Context) {
 		ws_helper.ServeWs(hub, context.Writer, context.Request)
 	})

internal/backend/ws_helper/client.go

@@ -29,11 +29,6 @@ const (
 	upGraderWriteBufferSize = 5 * 1024
 )
 
-var (
-	newline = []byte{'\n'}
-	space   = []byte{' '}
-)
-
 var upGrader = websocket.Upgrader{
 	ReadBufferSize:  upGraderReadBufferSize,
 	WriteBufferSize: upGraderWriteBufferSize,
@@ -43,20 +38,22 @@ var upGrader = websocket.Upgrader{
 }
 
 type Client struct {
-	hub *Hub
-
-	conn *websocket.Conn // 与服务器连接实例
-
-	sendLogLineIndex int // 日志发送到那个位置了
-
-	send chan []byte
+	hub              *Hub
+	conn             *websocket.Conn // 与服务器连接实例
+	sendLogLineIndex int             // 日志发送到那个位置了
+	authed           bool            // 是否已经通过认证
+	send             chan []byte     // 发送给 client 的内容 bytes
 }
 
+// 接收 Client 发送来的消息
 func (c *Client) readPump() {
+
 	defer func() {
+		// 触发移除 client 的逻辑
 		c.hub.unregister <- c
 		_ = c.conn.Close()
 	}()
+
 	c.conn.SetReadLimit(maxMessageSize)
 	err := c.conn.SetReadDeadline(time.Now().Add(pongWait))
 	if err != nil {
@@ -66,6 +63,7 @@ func (c *Client) readPump() {
 	c.conn.SetPongHandler(func(string) error {
 		return c.conn.SetReadDeadline(time.Now().Add(pongWait))
 	})
+	// 收取 client 发送过来的消息
 	for {
 		_, message, err := c.conn.ReadMessage()
 		if err != nil {
@@ -74,12 +72,15 @@ func (c *Client) readPump() {
 			}
 			break
 		}
-		message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
+		message = bytes.TrimSpace(bytes.Replace(message, []byte{}, []byte{}, -1))
 		c.hub.broadcast <- message
 	}
 }
 
+// 向 Client 发送消息的队列
 func (c *Client) writePump() {
+
+	// 心跳计时器
 	ticker := time.NewTicker(pingPeriod)
 	defer func() {
 		ticker.Stop()
@@ -89,6 +90,8 @@ func (c *Client) writePump() {
 	for {
 		select {
 		case message, ok := <-c.send:
+			// 这里是需要发送给 client 的消息
+			// 当然首先还是得先把当前消息的发送超时，给确定下来
 			err := c.conn.SetWriteDeadline(time.Now().Add(writeWait))
 			if err != nil {
 				log_helper.GetLogger().Errorln("writePump.SetWriteDeadline", err)
@@ -105,6 +108,7 @@ func (c *Client) writePump() {
 
 			w, err := c.conn.NextWriter(websocket.TextMessage)
 			if err != nil {
+				log_helper.GetLogger().Errorln("writePump.NextWriter", err)
 				return
 			}
 			_, err = w.Write(message)
@@ -115,18 +119,25 @@ func (c *Client) writePump() {
 			// Add queued chat messages to the current websocket message.
 			n := len(c.send)
 			for i := 0; i < n; i++ {
-				w.Write(newline)
-				w.Write(<-c.send)
+
+				_, err = w.Write(<-c.send)
+				if err != nil {
+					log_helper.GetLogger().Errorln("writePump.Write", err)
+					return
+				}
 			}
 
 			if err := w.Close(); err != nil {
+				log_helper.GetLogger().Errorln("writePump.Close", err)
 				return
 			}
 		case <-ticker.C:
+			// 心跳相关，这里是定时器到了触发的间隔，设置发送下一条心跳的超时时间
 			if err := c.conn.SetWriteDeadline(time.Now().Add(writeWait)); err != nil {
 				log_helper.GetLogger().Errorln("writePump.ticker.C.SetWriteDeadline", err)
 				return
 			}
+			// 然后发送心跳
 			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
 				log_helper.GetLogger().Errorln("writePump.ticker.C.WriteMessage", err)
 				return
@@ -135,17 +146,21 @@ func (c *Client) writePump() {
 	}
 }
 
+// ServeWs 每个 Client 连接 ws 上线时触发
 func ServeWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
+
 	conn, err := upGrader.Upgrade(w, r, nil)
 	if err != nil {
-		log_helper.GetLogger().Errorln("ServeWs", err)
+		log_helper.GetLogger().Errorln("ServeWs.Upgrade", err)
 		return
 	}
 
 	client := &Client{
-		hub:  hub,
-		conn: conn,
-		send: make(chan []byte, bufSize),
+		hub:              hub,
+		conn:             conn,
+		sendLogLineIndex: 0,
+		authed:           false,
+		send:             make(chan []byte, bufSize),
 	}
 	client.hub.register <- client
 

internal/backend/ws_helper/hub.go

@@ -29,8 +29,10 @@ func (h *Hub) Run() {
 	for {
 		select {
 		case client := <-h.register:
+			// 上线后先注册，然后等待 client 发送认证的 token 来确认有效，才进行后续的通信
 			h.clients[client] = true
 		case client := <-h.unregister:
+			// 下线后删除实例
 			if _, ok := h.clients[client]; ok {
 				delete(h.clients, client)
 				close(client.send)

internal/types/backend/ws/sub_download_jobs_info.go

@@ -0,0 +1,13 @@
+package ws
+
+type SubDownloadJobInfo struct {
+	Type              WSType `json:"type"`
+	Status            string `json:"status"`              // "running", "waiting"，不是运行中，就是等待中
+	StartedTime       string `json:"started_time"`        // 任务开始的时间
+	WorkingUnitIndex  int    `json:"working_unit_index"`  // 正在处理到第几部电影或者连续剧
+	UnitCount         int    `json:"unit_count"`          // 一共有多少部电影或者连续剧
+	WorkingUnitName   string `json:"working_unit_name"`   // 电影名称，或者连续剧的名称
+	WorkingVideoIndex int    `json:"working_video_index"` // 正在处理到第几个视频
+	VideoCount        int    `json:"video_count"`         // 一共有几个视频
+	WorkingVideoName  string `json:"working_video_name"`  // 电影名称，或者是连续剧中某一季的某一集的名称
+}

internal/types/backend/ws/ws_type.go

@@ -3,10 +3,10 @@ package ws
 type WSType int
 
 const (
-	CommonReply   WSType = iota // 通用回复
-	Auth                        // Client 发送登录握手消息
-	GetRunningLog               // Client 获取第一次的 Log 信息
-	RunningLog                  // Server 回复 Log 的信息
+	CommonReply           WSType = iota // 通用回复
+	Auth                                // Client 发送登录握手消息
+	RunningLog                          // Server 回复 Log 的信息
+	SubDownloadJobsStatus               // Server 主动发送的字幕下载任务状态信息
 )
 
 func (c WSType) String() string {
@@ -15,10 +15,10 @@ func (c WSType) String() string {
 		return "common_reply"
 	case Auth:
 		return "auth"
-	case GetRunningLog:
-		return "get_running_log"
 	case RunningLog:
 		return "running_log"
+	case SubDownloadJobsStatus:
+		return "sub_download_jobs_status"
 	}
 	return "N/A"
 }