Stream text/event-stream responses instead of buffering them

spring-boot-starter/spring-boot-starter-web/src/main/java/com/getyourguide/openapi/validation/factory/ContentCachingWrapperFactory.java

@@ -1,5 +1,6 @@
 package com.getyourguide.openapi.validation.factory;
 
+import com.getyourguide.openapi.validation.filter.EventStreamAwareContentCachingResponseWrapper;
 import com.getyourguide.openapi.validation.filter.MultiReadContentCachingRequestWrapper;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
@@ -22,7 +23,7 @@ public ContentCachingResponseWrapper buildContentCachingResponseWrapper(HttpServ
             return cachingResponse;
         }
 
-        return new ContentCachingResponseWrapper(response);
+        return new EventStreamAwareContentCachingResponseWrapper(response);
     }
 
     @Nullable

spring-boot-starter/spring-boot-starter-web/src/main/java/com/getyourguide/openapi/validation/filter/EventStreamAwareContentCachingResponseWrapper.java

@@ -0,0 +1,46 @@
+package com.getyourguide.openapi.validation.filter;
+
+import jakarta.servlet.ServletOutputStream;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Locale;
+import org.springframework.http.MediaType;
+import org.springframework.web.util.ContentCachingResponseWrapper;
+
+/**
+ * Caches the response body for validation, except for Server-Sent Events: a {@code text/event-stream}
+ * response is written and flushed straight through so each event reaches the client as it is sent.
+ * Its body is never validated anyway, as only JSON, XML and HTML responses are.
+ */
+public class EventStreamAwareContentCachingResponseWrapper extends ContentCachingResponseWrapper {
+
+    public EventStreamAwareContentCachingResponseWrapper(HttpServletResponse response) {
+        super(response);
+    }
+
+    @Override
+    public ServletOutputStream getOutputStream() throws IOException {
+        return isEventStream() ? getResponse().getOutputStream() : super.getOutputStream();
+    }
+
+    @Override
+    public PrintWriter getWriter() throws IOException {
+        return isEventStream() ? getResponse().getWriter() : super.getWriter();
+    }
+
+    @Override
+    public void flushBuffer() throws IOException {
+        if (isEventStream()) {
+            getResponse().flushBuffer();
+        } else {
+            super.flushBuffer();
+        }
+    }
+
+    private boolean isEventStream() {
+        var contentType = getContentType();
+        return contentType != null
+            && contentType.toLowerCase(Locale.ROOT).startsWith(MediaType.TEXT_EVENT_STREAM_VALUE);
+    }
+}

spring-boot-starter/spring-boot-starter-web/src/test/java/com/getyourguide/openapi/validation/filter/EventStreamAwareContentCachingResponseWrapperTest.java

@@ -0,0 +1,53 @@
+package com.getyourguide.openapi.validation.filter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+import org.springframework.mock.web.MockHttpServletResponse;
+
+class EventStreamAwareContentCachingResponseWrapperTest {
+
+    private final MockHttpServletResponse response = new MockHttpServletResponse();
+    private final EventStreamAwareContentCachingResponseWrapper wrapper =
+        new EventStreamAwareContentCachingResponseWrapper(response);
+
+    @Test
+    public void whenJsonResponseThenBodyIsCachedUntilCopied() throws IOException {
+        wrapper.setContentType("application/json");
+        wrapper.getOutputStream().write("{}".getBytes(StandardCharsets.UTF_8));
+        wrapper.flushBuffer();
+
+        assertEquals("", response.getContentAsString());
+        assertFalse(response.isCommitted());
+        assertEquals("{}", new String(wrapper.getContentAsByteArray(), StandardCharsets.UTF_8));
+
+        wrapper.copyBodyToResponse();
+
+        assertEquals("{}", response.getContentAsString());
+    }
+
+    @Test
+    public void whenEventStreamThenEachWriteAndFlushReachesTheResponse() throws IOException {
+        wrapper.setContentType("text/event-stream;charset=UTF-8");
+        wrapper.getOutputStream().write("data: first\n\n".getBytes(StandardCharsets.UTF_8));
+        wrapper.flushBuffer();
+
+        assertEquals("data: first\n\n", response.getContentAsString());
+        assertTrue(response.isCommitted());
+        assertEquals(0, wrapper.getContentSize());
+    }
+
+    @Test
+    public void whenEventStreamWrittenThroughWriterThenItReachesTheResponse() throws IOException {
+        wrapper.setContentType("TEXT/EVENT-STREAM");
+        wrapper.getWriter().write("data: first\n\n");
+        wrapper.getWriter().flush();
+
+        assertEquals("data: first\n\n", response.getContentAsString());
+        assertEquals(0, wrapper.getContentSize());
+    }
+}

spring-boot-starter/spring-boot-starter-web/src/test/java/com/getyourguide/openapi/validation/integration/EventStreamIntegrationTest.java

@@ -0,0 +1,46 @@
+package com.getyourguide.openapi.validation.integration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import com.getyourguide.openapi.validation.integration.controller.EventStreamController;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.web.servlet.MockMvc;
+
+@SpringBootTest
+@AutoConfigureMockMvc
+@ExtendWith(SpringExtension.class)
+public class EventStreamIntegrationTest {
+
+    @Autowired
+    private MockMvc mockMvc;
+
+    @Test
+    public void whenEventStreamThenEventsReachTheClientBeforeTheStreamCompletes() throws Exception {
+        var result = mockMvc.perform(get("/sse").accept(MediaType.TEXT_EVENT_STREAM))
+            .andExpect(request().asyncStarted())
+            .andReturn();
+
+        assertEquals("data:first\n\n", result.getResponse().getContentAsString());
+
+        var emitter = EventStreamController.CURRENT.get();
+        emitter.send("second");
+        emitter.complete();
+
+        mockMvc.perform(asyncDispatch(result))
+            .andExpectAll(
+                status().isOk(),
+                content().string("data:first\n\ndata:second\n\n")
+            );
+    }
+}

spring-boot-starter/spring-boot-starter-web/src/test/java/com/getyourguide/openapi/validation/integration/controller/EventStreamController.java

@@ -0,0 +1,21 @@
+package com.getyourguide.openapi.validation.integration.controller;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+@RestController
+public class EventStreamController {
+    public static final AtomicReference<SseEmitter> CURRENT = new AtomicReference<>();
+
+    @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public SseEmitter stream() throws IOException {
+        var emitter = new SseEmitter(10_000L);
+        emitter.send("first");
+        CURRENT.set(emitter);
+        return emitter;
+    }
+}