ServerEventDecoder { public void onReceivedContentPart(ByteBuf content) throws KingHttpException { String contentString = content.toString(StandardCharsets.UTF_8); try { char[] chars = contentString.toCharArray(); for (int i = 0; i < chars.length; i++) { char c = chars[i]; if (chars[i] == '\r') { continue; } if (isNewLine(c)) { String line = buffer.toString(); buffer.setLength(0); parseLine(line); } else { buffer.append(c); } } } catch (Exception e) { throw new KingHttpException("Failed to parse incoming content in SSE stream", e); } } ServerEventDecoder(SseClientCallback sseClientCallback); void reset(); void onReceivedContentPart(ByteBuf content);  }
@Test public void simpleDataEvent() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("data: test\n\n")); assertEquals(1, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertNull(event.event); assertNull(event.lastSentId); assertEquals("test", event.data); }
@Test public void endOfLines() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("data: test1\n\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test3\r\n\r\n")); assertEquals(2, sseCallback.count); Event event1 = sseCallback.poll(); Event event3 = sseCallback.poll(); assertEquals("test1", event1.data); assertEquals("test3", event3.data); }
@Test public void twoDataEvents() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("data: test\n\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test\n\n")); assertEquals(2, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertNull(event.event); assertNull(event.lastSentId); assertEquals("test", event.data); Event event2 = sseCallback.poll(); assertNotNull(event2); assertEquals(event, event2); }
@Test public void differentFormattedDataEvents() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("data:test1\n\n")); serverEventDecoder.onReceivedContentPart(buffer("data :test2\n\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test3\n\n")); assertEquals(3, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertNull(event.event); assertNull(event.lastSentId); assertEquals("test1", event.data); Event event2 = sseCallback.poll(); Event event3 = sseCallback.poll(); assertEquals("test2", event2.data); assertEquals("test3", event3.data); }
@Test public void emptyLinesShouldNotTriggerEvents() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); assertEquals(0, sseCallback.count); }
@Test public void partialEmitOfDataEvents() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("da")); serverEventDecoder.onReceivedContentPart(buffer("ta")); serverEventDecoder.onReceivedContentPart(buffer(":")); serverEventDecoder.onReceivedContentPart(buffer(" te")); serverEventDecoder.onReceivedContentPart(buffer("st")); serverEventDecoder.onReceivedContentPart(buffer("\r\n")); assertEquals(0, sseCallback.count); serverEventDecoder.onReceivedContentPart(buffer("\n")); assertEquals(1, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertNull(event.event); assertNull(event.lastSentId); assertEquals("test", event.data); }
@Test public void eventName() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("event: testEvent\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test\n")); serverEventDecoder.onReceivedContentPart(buffer("\n")); assertEquals(1, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertEquals("testEvent", event.event); assertEquals("test", event.data); assertNull(event.lastSentId); }
@Test public void dataWithNewLineInOneEvent() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("event: some_event\r\ndata: data1\r\ndata: data2\r\n\r\n")); assertEquals(1, sseCallback.count); Event event = sseCallback.poll(); assertNotNull(event); assertEquals("some_event", event.event); assertEquals("data1\ndata2", event.data); assertNull(event.lastSentId); }
@Test public void id() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("id: 1\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test\n\n")); Event event = sseCallback.poll(); assertNotNull(event); assertNull(event.event); assertEquals("1", event.lastSentId); assertEquals("test", event.data); }
@Test public void lastSentIdIsReused() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("id: 1\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test1\n\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test2\n\n")); Event event1 = sseCallback.poll(); assertNotNull(event1); assertNull(event1.event); assertEquals("1", event1.lastSentId); assertEquals("test1", event1.data); Event event2 = sseCallback.poll(); assertEquals("1", event2.lastSentId); assertEquals("test2", event2.data); serverEventDecoder.onReceivedContentPart(buffer("id: 2\n")); serverEventDecoder.onReceivedContentPart(buffer("data: test3\n\n")); Event event3 = sseCallback.poll(); assertEquals("2", event3.lastSentId); assertEquals("test3", event3.data); }
@Test public void eventIdData() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("event: testEvent1\ndata: test1\nid: 1\n\n")); serverEventDecoder.onReceivedContentPart(buffer("event: testEvent2\ndata: test2\nid: 2\n\n")); serverEventDecoder.onReceivedContentPart(buffer("event: testEvent3\ndata: test3\nid: 3\n\n")); assertEquals(3, sseCallback.count); Event event1 = sseCallback.poll(); Event event2 = sseCallback.poll(); Event event3 = sseCallback.poll(); assertEquals("1", event1.lastSentId); assertEquals("2", event2.lastSentId); assertEquals("3", event3.lastSentId); assertEquals("test1", event1.data); assertEquals("test2", event2.data); assertEquals("test3", event3.data); assertEquals("testEvent1", event1.event); assertEquals("testEvent2", event2.event); assertEquals("testEvent3", event3.event); }
@Test public void verifyBrokenUpCRLF() throws Exception { ServerEventDecoder serverEventDecoder = new ServerEventDecoder(sseCallback); serverEventDecoder.onReceivedContentPart(buffer("event: ClientCommand\r\ndata: test1\r\n\n")); serverEventDecoder.onReceivedContentPart(buffer("event: ClientCommand\r")); serverEventDecoder.onReceivedContentPart(buffer("\ndata: test2\r\n\n")); assertEquals(2, sseCallback.count); Event event1 = sseCallback.poll(); Event event2 = sseCallback.poll(); assertEquals("ClientCommand", event1.event); assertEquals("ClientCommand", event2.event); }