diff --git a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jsonb/deployment/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/sse/SseParserTest.java b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jsonb/deployment/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/sse/SseParserTest.java index 36cca2e23779e..12c47625797e5 100644 --- a/extensions/resteasy-reactive/quarkus-resteasy-reactive-jsonb/deployment/src/test/java/io/quarkus/resteasy/reactive/jsonb/deployment/test/sse/SseParserTest.java +++ b/extensions/resteasy-reactive/quarkus-resteasy-reactive-jsonb/deployment/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/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java index 46bb82858514a..45db7ae4ac6fb 100644 --- a/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java +++ b/independent-projects/resteasy-reactive/client/runtime/src/main/java/org/jboss/resteasy/reactive/client/impl/SseParser.java @@ -40,7 +40,10 @@ public class SseParser implements Handler { * 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() {