From f0c7acd903c3bc0dc2243d2865dd9bfbef1b6435 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sybren=20A=2E=20St=C3=BCvel?= <sybren@blender.org>
Date: Sun, 4 Feb 2024 16:11:38 +0100
Subject: [PATCH] Manager: fix web interface not showing last-rendered images
 on job view

Fix SocketIO subscriptions so that the client also subscribes to
job-specific last-rendered images whenever subscribing to job-specific
events. These are sent to another event topic, and thus need some extra
care. Before the introduction of the generic event bus, both message types
were sent to the same topic, but that's not supported by MQTT, and so things
had to change.
---
 internal/manager/eventbus/socketio.go | 46 ++++++++++++++++-----------
 1 file changed, 28 insertions(+), 18 deletions(-)

diff --git a/internal/manager/eventbus/socketio.go b/internal/manager/eventbus/socketio.go
index 234972e1..2997b73b 100644
--- a/internal/manager/eventbus/socketio.go
+++ b/internal/manager/eventbus/socketio.go
@@ -3,6 +3,7 @@ package eventbus
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 
@@ -122,44 +123,37 @@ func (s *SocketIOForwarder) handleRoomSubscription(c *gosocketio.Channel, subs a
 		return "invalid UUID, ignoring request"
 	}
 
-	var sioRoom EventTopic
+	var err error
 	switch subs.Type {
 	case api.SocketIOSubscriptionTypeAllJobs:
-		sioRoom = TopicJobUpdate
+		err = s.subUnsub(c, TopicJobUpdate, subs.Op)
 	case api.SocketIOSubscriptionTypeAllWorkers:
-		sioRoom = TopicWorkerUpdate
+		err = s.subUnsub(c, TopicWorkerUpdate, subs.Op)
 	case api.SocketIOSubscriptionTypeAllLastRendered:
-		sioRoom = TopicLastRenderedImage
+		err = s.subUnsub(c, TopicLastRenderedImage, subs.Op)
 	case api.SocketIOSubscriptionTypeAllWorkerTags:
-		sioRoom = TopicWorkerTagUpdate
+		err = s.subUnsub(c, TopicWorkerTagUpdate, subs.Op)
 	case api.SocketIOSubscriptionTypeJob:
 		if subs.Uuid == nil {
 			logger.Warn().Msg("socketIO: trying to (un)subscribe to job without UUID")
 			return "operation on job requires a UUID"
 		}
-		sioRoom = topicForJob(*subs.Uuid)
+		logger.Trace().Msg("socketio: sub subscription, also going to do last-rendered for that job")
+		err = s.subUnsub(c, topicForJob(*subs.Uuid), subs.Op)
+		if err == nil {
+			err = s.subUnsub(c, topicForJobLastRendered(*subs.Uuid), subs.Op)
+		}
 	case api.SocketIOSubscriptionTypeTasklog:
 		if subs.Uuid == nil {
 			logger.Warn().Msg("socketIO: trying to (un)subscribe to task without UUID")
 			return "operation on task requires a UUID"
 		}
-		sioRoom = topicForTaskLog(*subs.Uuid)
+		err = s.subUnsub(c, topicForJob(*subs.Uuid), subs.Op)
 	default:
 		logger.Warn().Msg("socketIO: unknown subscription type, ignoring")
 		return "unknown subscription type, ignoring request"
 	}
 
-	var err error
-	switch subs.Op {
-	case api.SocketIOSubscriptionOperationSubscribe:
-		err = c.Join(string(sioRoom))
-	case api.SocketIOSubscriptionOperationUnsubscribe:
-		err = c.Leave(string(sioRoom))
-	default:
-		logger.Warn().Msg("socketIO: invalid subscription operation, ignoring")
-		return "invalid subscription operation, ignoring request"
-	}
-
 	if err != nil {
 		logger.Warn().Err(err).Msg("socketIO: performing subscription operation")
 		return fmt.Sprintf("unable to perform subscription operation: %v", err)
@@ -168,3 +162,19 @@ func (s *SocketIOForwarder) handleRoomSubscription(c *gosocketio.Channel, subs a
 	logger.Debug().Msg("socketIO: subscription")
 	return "ok"
 }
+
+func (s *SocketIOForwarder) subUnsub(
+	c *gosocketio.Channel,
+	topic EventTopic,
+	operation api.SocketIOSubscriptionOperation,
+) error {
+	room := string(topic)
+	switch operation {
+	case api.SocketIOSubscriptionOperationSubscribe:
+		return c.Join(room)
+	case api.SocketIOSubscriptionOperationUnsubscribe:
+		return c.Leave(room)
+	default:
+		return errors.New("invalid subscription operation, ignoring request")
+	}
+}