RR client: fix chunking between \n\n in SSEParser quarkusio#37625

Fixes quarkusio#37625
gsmet · Dec 20, 2023 · 5070e60 · 5070e60
1 parent e4d93a1
commit 5070e60
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 1 deletion.
diff --git a/...t/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/sse/SseParserTest.java b/...t/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/sse/SseParserTest.java
@@ -108,6 +108,22 @@ public void testParser() {
         testParser(Arrays.asList("data:f", "oo\n\n"),
                 Collections.singletonList(new InboundSseEventImpl(null, null)
                         .setData("foo")));
+        testParser(Arrays.asList("dat", "a:foo\n\n"),
+                Collections.singletonList(new InboundSseEventImpl(null, null)
+                        .setData("foo")));
+        testParser(Arrays.asList("data", ":foo\n\n"),
+                Collections.singletonList(new InboundSseEventImpl(null, null)
+                        .setData("foo")));
+        testParser(Arrays.asList("data:", "foo\n\n"),
+                Collections.singletonList(new InboundSseEventImpl(null, null)
+                        .setData("foo")));
+        // chunk at the worst possible place, make sure we don't drop events
+        testParser(Arrays.asList("data:foo\n", "\n"),
+                Collections.singletonList(new InboundSseEventImpl(null, null)
+                        .setData("foo")));
+        testParser(Arrays.asList("data:foo\n", "data:bar\n", "\n"),
+                Collections.singletonList(new InboundSseEventImpl(null, null)
+                        .setData("foo\nbar")));
         // one event in two buffers within a UTF-8 char
         testParserWithBytes(
                 Arrays.asList(new byte[] { 'd', 'a', 't', 'a', ':', (byte) 0b11000010 },

diff --git a/...ctive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java b/...ctive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java
@@ -40,7 +40,10 @@ public class SseParser implements Handler<Buffer> {
      * True if we're at the very beginning of the data stream and could see a BOM
      */
     private boolean firstByte = true;
-
+    /**
+     * True if we've started to read at least one byte of an event
+     */
+    private boolean startedEvent = false;
     /**
      * The event type we're reading. Defaults to "message" and changes with "event" fields
      */
@@ -95,6 +98,7 @@ public void handle(Buffer event) {
 
         while (hasByte()) {
             boolean lastFirstByte = firstByte;
+            startedEvent = false;
             nameBuffer.setLength(0);
             valueBuffer.setLength(0);
             commentBuffer.setLength(0);
@@ -105,10 +109,19 @@ public void handle(Buffer event) {
             eventReconnectTime = SseEvent.RECONNECT_NOT_SET;
             // SSE spec says ID is persistent
 
+            boolean needsMoreData = false;
             int lastEventStart = i;
             try {
                 parseEvent();
+                // if we started an event but did not fire it, it means we lacked a final end-of-line and must
+                // wait for more data
+                if (startedEvent) {
+                    needsMoreData = true;
+                }
             } catch (NeedsMoreDataException x) {
+                needsMoreData = true;
+            }
+            if (needsMoreData) {
                 // save the remaining bytes for later
                 i = lastEventStart;
                 // be ready to rescan the BOM, but only if we didn't already see it in a previous event
@@ -133,8 +146,10 @@ private void parseEvent() {
             int c = readChar();
             firstByte = false;
             if (c == COLON) {
+                startedEvent = true;
                 parseComment();
             } else if (isNameChar(c)) {
+                startedEvent = true;
                 parseField(c);
             } else if (isEofWithSideEffect(c)) {
                 dispatchEvent();
@@ -164,6 +179,8 @@ private void dispatchEvent() {
         event.setReconnectDelay(eventReconnectTime);
         event.setMediaType(contentType != null ? MediaType.valueOf(contentType) : null);
         sseEventSource.fireEvent(event);
+        // make sure we mark that we are done with this event
+        startedEvent = false;
     }
 
     private byte peekByte() {