[cxf] branch 3.2.x-fixes updated: CXF-8207: SseEventSource loses lastEventId on the 2nd reconnect attempt. Adding a test case for LAST_EVENT_ID

[re...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

reta pushed a commit to branch 3.2.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git


The following commit(s) were added to refs/heads/3.2.x-fixes by this push:
     new 81ef500  CXF-8207: SseEventSource loses lastEventId on the 2nd reconnect attempt. Adding a test case for LAST_EVENT_ID
81ef500 is described below

commit 81ef500319fe4989066f1a2a957bb90427682c2a
Author: reta <dr...@gmail.com>
AuthorDate: Sun Feb 2 21:27:40 2020 -0500

    CXF-8207: SseEventSource loses lastEventId on the 2nd reconnect attempt. Adding a test case for LAST_EVENT_ID
---
 .../jaxrs/sse/client/SseEventSourceImplTest.java   | 34 ++++++++++++++++++++++
 1 file changed, 34 insertions(+)

diff --git a/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java b/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
index 70d5b46..01e552c 100644
--- a/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
+++ b/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
@@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit;
 
 import javax.ws.rs.client.Invocation;
 import javax.ws.rs.core.Configuration;
+import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.sse.InboundSseEvent;
@@ -518,6 +519,39 @@ public class SseEventSourceImplTest extends Assert {
         }
     }
     
+    @Test
+    public void testConnectWithLastEventId() throws InterruptedException, IOException {
+        final String payload = EVENT_NO_RETRY.replaceAll("id: 1", "id: 11");
+        
+        when(configuration.getProperty(HttpHeaders.LAST_EVENT_ID_HEADER)).thenReturn("10");
+        when(builder.header(HttpHeaders.LAST_EVENT_ID_HEADER, "10")).thenAnswer(invocation -> {
+            try (InputStream is = new ByteArrayInputStream(payload.getBytes(StandardCharsets.UTF_8))) {
+                when(response.getStatus()).thenReturn(200);
+                when(response.readEntity(InputStream.class)).thenReturn(is);
+                return builder;
+            }
+        });
+
+        final List<InboundSseEvent> events = new ArrayList<>();
+        try (SseEventSource eventSource = withReconnect()) {
+            eventSource.register(events::add);
+            eventSource.open();
+           
+            assertThat(eventSource.isOpen(), equalTo(true));
+            verify(response, times(1)).getStatus();
+           
+            // Allow the event processor to pull for events (150ms)
+            Thread.sleep(150L);
+        }
+   
+        assertThat(events.size(), equalTo(1));
+        assertThat(events.get(0).getId(), equalTo("11"));
+        assertThat(events.get(0).getReconnectDelay(), equalTo(-1L));
+        assertThat(events.get(0).getComment(), equalTo("test comment"));
+        assertThat(events.get(0).readData(), equalTo("test data"));
+    }
+
+    
     private SseEventSource withNoReconnect() {
         return SseEventSource.target(target).build();
     }