682 Additional fixes

Author: ivicac

server/libs/platform/platform-workflow/platform-workflow-test/platform-workflow-test-rest/src/main/java/com/bytechef/platform/workflow/test/web/rest/WorkflowTestApiController.java

@@ -117,11 +117,9 @@ public SseEmitter attachWorkflowTest(@PathVariable Long jobId) {
         if (future == null) {
             List<SseEmitter.SseEventBuilder> bufferedEvents = pendingEvents.getIfPresent(key);
 
-            if (bufferedEvents != null) {
+            if (bufferedEvents != null && !bufferedEvents.isEmpty()) {
                 pendingEvents.invalidate(key);
-            }
 
-            if (bufferedEvents != null && !bufferedEvents.isEmpty()) {
                 try {
                     for (SseEmitter.SseEventBuilder eventBuilder : bufferedEvents) {
                         try {
@@ -136,6 +134,10 @@ public SseEmitter attachWorkflowTest(@PathVariable Long jobId) {
                     emitter.complete();
                 }
             } else {
+                if (bufferedEvents != null) {
+                    pendingEvents.invalidate(key);
+                }
+
                 try {
                     emitter.send(createEvent("error", "Not running"));
                 } catch (Exception exception) {
@@ -313,9 +315,7 @@ private void completeAndClearEmitter(String key) {
 
         if (emitter != null) {
             this.emitter.invalidate(key);
-        }
 
-        if (emitter != null) {
             try {
                 emitter.complete();
             } catch (Exception exception) {
@@ -379,9 +379,7 @@ private void registerEmitter(String key, long jobId, SseEmitter emitter) {
 
         if (bufferedEvents != null) {
             pendingEvents.invalidate(key);
-        }
 
-        if (bufferedEvents != null) {
             for (SseEmitter.SseEventBuilder eventBuilder : bufferedEvents) {
                 sendToEmitter(key, eventBuilder);
             }
@@ -422,9 +420,7 @@ private void unregisterListeners(String key) {
 
         if (handles != null) {
             listenerHandles.invalidate(key);
-        }
 
-        if (handles != null) {
             for (AutoCloseable handle : handles) {
                 try {
                     handle.close();

server/libs/platform/platform-workflow/platform-workflow-test/platform-workflow-test-rest/src/test/java/com/bytechef/platform/workflow/test/web/rest/WorkflowTestApiControllerTest.java

@@ -364,21 +364,19 @@ void testPendingEventsBoundedAndFlushedOnAttach() throws Exception {
         // Obtain the key from the controller's 'runs' map (wait briefly if needed)
         String key = waitForRunKey();
 
-        // Remove emitters to force buffering
-        Object emitters = ReflectionTestUtils.getField(controller, "emitter");
+        // Remove emitter to force buffering
+        Object emitter = ReflectionTestUtils.getField(controller, "emitter");
 
-        assert emitters != null;
+        assert emitter != null;
 
-        if (emitters instanceof Cache<?, ?> cache) {
+        if (emitter instanceof Cache<?, ?> cache) {
             @SuppressWarnings("unchecked")
-            Cache<String, CopyOnWriteArrayList<SseEmitter>> emitterCache =
-                (Cache<String, CopyOnWriteArrayList<SseEmitter>>) cache;
+            Cache<String, SseEmitter> emitterCache = (Cache<String, SseEmitter>) cache;
 
             emitterCache.invalidate(key);
-        } else if (emitters instanceof ConcurrentMap<?, ?>) {
+        } else if (emitter instanceof ConcurrentMap<?, ?>) {
             @SuppressWarnings("unchecked")
-            ConcurrentMap<String, CopyOnWriteArrayList<SseEmitter>> emitterMap =
-                (ConcurrentMap<String, CopyOnWriteArrayList<SseEmitter>>) emitters;
+            ConcurrentMap<String, SseEmitter> emitterMap = (ConcurrentMap<String, SseEmitter>) emitter;
 
             emitterMap.remove(key);
         }
@@ -477,14 +475,12 @@ void testPendingEventsClearedOnStop() throws Exception {
 
         if (emitter instanceof Cache<?, ?> cache) {
             @SuppressWarnings("unchecked")
-            Cache<String, CopyOnWriteArrayList<SseEmitter>> emitterCache =
-                (Cache<String, CopyOnWriteArrayList<SseEmitter>>) cache;
+            Cache<String, SseEmitter> emitterCache = (Cache<String, SseEmitter>) cache;
 
             emitterCache.invalidate(key);
         } else if (emitter instanceof ConcurrentMap<?, ?>) {
             @SuppressWarnings("unchecked")
-            ConcurrentMap<String, CopyOnWriteArrayList<SseEmitter>> emitterMap =
-                (ConcurrentMap<String, CopyOnWriteArrayList<SseEmitter>>) emitter;
+            ConcurrentMap<String, SseEmitter> emitterMap = (ConcurrentMap<String, SseEmitter>) emitter;
 
             emitterMap.remove(key);
         }