cxf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r...@apache.org
Subject cxf git commit: CXF-7085: Introduce support for Server Sent Events (Client). Adding more test cases, many fixes in SSE protocol (reader / writer) implementation
Date Fri, 18 Aug 2017 00:55:06 GMT
Repository: cxf
Updated Branches:
  refs/heads/master 9890add2c -> 4cef686cb


CXF-7085: Introduce support for Server Sent Events (Client). Adding more test cases, many
fixes in SSE protocol (reader / writer) implementation


Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/4cef686c
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/4cef686c
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/4cef686c

Branch: refs/heads/master
Commit: 4cef686cbbfcc8ec99042b525b44e5cc12647e1d
Parents: 9890add
Author: reta <drreta@gmail.com>
Authored: Thu Aug 17 20:54:47 2017 -0400
Committer: reta <drreta@gmail.com>
Committed: Thu Aug 17 20:54:47 2017 -0400

----------------------------------------------------------------------
 .../src/main/resources/web-ui/index.html        |   2 +-
 .../demo/jaxrs/sse/StatsRestServiceImpl.java    |  16 +--
 .../src/main/resources/web-ui/index.html        |   2 +-
 .../jaxrs/sse/OutboundSseEventBodyWriter.java   |   2 +-
 .../jaxrs/sse/client/InboundSseEventImpl.java   |  10 +-
 .../sse/client/InboundSseEventProcessor.java    |  36 +++---
 .../sse/client/SseEventSourceImplTest.java      | 118 ++++++++++++++++++-
 7 files changed, 154 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/distribution/src/main/release/samples/jax_rs/sse_cdi/src/main/resources/web-ui/index.html
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/sse_cdi/src/main/resources/web-ui/index.html
b/distribution/src/main/release/samples/jax_rs/sse_cdi/src/main/resources/web-ui/index.html
index d79f804..e7b4743 100644
--- a/distribution/src/main/release/samples/jax_rs/sse_cdi/src/main/resources/web-ui/index.html
+++ b/distribution/src/main/release/samples/jax_rs/sse_cdi/src/main/resources/web-ui/index.html
@@ -33,7 +33,7 @@
 	if( !!window.EventSource ) {
 	    var event = new EventSource("http://localhost:8686/rest/api/stats/sse");
 	
-	    event.addEventListener('message', function( event ) {	
+	    event.addEventListener('stats', function( event ) {	
 	    	var datapoint = jQuery.parseJSON( event.data );
 	    	 
 	    	chart.series[ 0 ].addPoint({

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/distribution/src/main/release/samples/jax_rs/sse_spring/src/main/java/demo/jaxrs/sse/StatsRestServiceImpl.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/sse_spring/src/main/java/demo/jaxrs/sse/StatsRestServiceImpl.java
b/distribution/src/main/release/samples/jax_rs/sse_spring/src/main/java/demo/jaxrs/sse/StatsRestServiceImpl.java
index 28730a3..0a15153 100644
--- a/distribution/src/main/release/samples/jax_rs/sse_spring/src/main/java/demo/jaxrs/sse/StatsRestServiceImpl.java
+++ b/distribution/src/main/release/samples/jax_rs/sse_spring/src/main/java/demo/jaxrs/sse/StatsRestServiceImpl.java
@@ -53,21 +53,21 @@ public class StatsRestServiceImpl {
             public void run() {
                 try {
                     final Builder builder = sse.newEventBuilder();
-                    sink.send(createStatsEvent(builder.name("stats"), 1));
+                    sink.send(createStatsEvent(builder, 1));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 2));
+                    sink.send(createStatsEvent(builder, 2));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 3));
+                    sink.send(createStatsEvent(builder, 3));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 4));
+                    sink.send(createStatsEvent(builder, 4));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 5));
+                    sink.send(createStatsEvent(builder, 5));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 6));
+                    sink.send(createStatsEvent(builder, 6));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 7));
+                    sink.send(createStatsEvent(builder, 7));
                     Thread.sleep(1000);
-                    sink.send(createStatsEvent(builder.name("stats"), 8));
+                    sink.send(createStatsEvent(builder, 8));
                     sink.close();
                 } catch (final Exception e) {
                     e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/distribution/src/main/release/samples/jax_rs/sse_tomcat/src/main/resources/web-ui/index.html
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/jax_rs/sse_tomcat/src/main/resources/web-ui/index.html
b/distribution/src/main/release/samples/jax_rs/sse_tomcat/src/main/resources/web-ui/index.html
index 2aaed6e..2b58b30 100644
--- a/distribution/src/main/release/samples/jax_rs/sse_tomcat/src/main/resources/web-ui/index.html
+++ b/distribution/src/main/release/samples/jax_rs/sse_tomcat/src/main/resources/web-ui/index.html
@@ -33,7 +33,7 @@
 	if( !!window.EventSource ) {
 	    var event = new EventSource("http://localhost:8686/rest/api/stats/sse/1");
 	
-	    event.addEventListener('message', function( event ) {	
+	    event.addEventListener('stats', function( event ) {	
 	    	var datapoint = jQuery.parseJSON( event.data );
 	    	 
 	    	chart.series[ 0 ].addPoint({

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/OutboundSseEventBodyWriter.java
----------------------------------------------------------------------
diff --git a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/OutboundSseEventBodyWriter.java
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/OutboundSseEventBodyWriter.java
index 085df51..a97020f 100644
--- a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/OutboundSseEventBodyWriter.java
+++ b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/OutboundSseEventBodyWriter.java
@@ -43,7 +43,7 @@ public class OutboundSseEventBodyWriter implements MessageBodyWriter<OutboundSse
     public static final MediaType SERVER_SENT_EVENTS_TYPE = MediaType.valueOf(SERVER_SENT_EVENTS);
 
     private static final byte[] COMMENT = ": ".getBytes(StandardCharsets.UTF_8);
-    private static final byte[] EVENT = "    ".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] EVENT = "event: ".getBytes(StandardCharsets.UTF_8);
     private static final byte[] ID = "id: ".getBytes(StandardCharsets.UTF_8);
     private static final byte[] RETRY = "retry: ".getBytes(StandardCharsets.UTF_8);
     private static final byte[] DATA = "data: ".getBytes(StandardCharsets.UTF_8);

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventImpl.java
----------------------------------------------------------------------
diff --git a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventImpl.java
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventImpl.java
index f173576..a13b591 100644
--- a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventImpl.java
+++ b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventImpl.java
@@ -50,20 +50,24 @@ public final class InboundSseEventImpl implements InboundSseEvent {
     static class Builder {
         private static final Logger LOG = LogUtils.getL7dLogger(Builder.class);
 
-        private final String name;
+        private String name; /* the default event type would be "message" */
         private String id;
         private String comment;
         private OptionalLong reconnectDelay = OptionalLong.empty();
         private String data;
 
-        Builder(String name) {
-            this.name = name;
+        Builder() {
         }
 
         Builder id(String i) {
             this.id = i;
             return this;
         }
+        
+        Builder name(String n) {
+            this.name = n;
+            return this;
+        }
 
         Builder comment(String cmt) {
             this.comment = cmt;

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventProcessor.java
----------------------------------------------------------------------
diff --git a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventProcessor.java
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventProcessor.java
index 2c0b8b5..4d88ee7 100644
--- a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventProcessor.java
+++ b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/InboundSseEventProcessor.java
@@ -35,6 +35,7 @@ import org.apache.cxf.common.util.StringUtils;
 import org.apache.cxf.endpoint.Endpoint;
 import org.apache.cxf.jaxrs.client.ClientProviderFactory;
 import org.apache.cxf.jaxrs.impl.ResponseImpl;
+import org.apache.cxf.jaxrs.sse.client.InboundSseEventImpl.Builder;
 import org.apache.cxf.message.Message;
 
 public class InboundSseEventProcessor {
@@ -42,7 +43,7 @@ public class InboundSseEventProcessor {
     public static final MediaType SERVER_SENT_EVENTS_TYPE = MediaType.valueOf(SERVER_SENT_EVENTS);
 
     private static final String COMMENT = ": ";
-    private static final String EVENT = "    ";
+    private static final String EVENT = "event: ";
     private static final String ID = "id: ";
     private static final String RETRY = "retry: ";
     private static final String DATA = "data: ";
@@ -82,23 +83,21 @@ public class InboundSseEventProcessor {
                 InboundSseEventImpl.Builder builder = null;
 
                 while (line != null && !Thread.interrupted() && !closed)
{
-                    if (!StringUtils.isEmpty(line) && line.startsWith(EVENT)) {
-                        if (builder == null) {
-                            builder = new InboundSseEventImpl.Builder(line.substring(EVENT.length()));
-                        } else {
-                            final InboundSseEvent event = builder.build(factory, message);
-                            builder = new InboundSseEventImpl.Builder(line.substring(EVENT.length()));
-                            listener.onNext(event);
-                        }
-                    } else if (builder != null) {
-                        if (line.startsWith(ID)) {
-                            builder.id(line.substring(ID.length()));
+                    if (StringUtils.isEmpty(line) && builder != null) { /* empty
new line */
+                        final InboundSseEvent event = builder.build(factory, message);
+                        builder = null; /* reset the builder for next event */
+                        listener.onNext(event);
+                    } else {
+                        if (line.startsWith(EVENT)) {
+                            builder = getOrCreate(builder).name(line.substring(EVENT.length()));
+                        } else if (line.startsWith(ID)) {
+                            builder = getOrCreate(builder).id(line.substring(ID.length()));
                         } else if (line.startsWith(COMMENT)) {
-                            builder.comment(line.substring(COMMENT.length()));
+                            builder = getOrCreate(builder).comment(line.substring(COMMENT.length()));
                         } else if (line.startsWith(RETRY)) {
-                            builder.reconnectDelay(line.substring(RETRY.length()));
+                            builder = getOrCreate(builder).reconnectDelay(line.substring(RETRY.length()));
                         } else if (line.startsWith(DATA)) {
-                            builder.data(line.substring(DATA.length()));
+                            builder = getOrCreate(builder).data(line.substring(DATA.length()));
                         }
                     }
                     line = reader.readLine();
@@ -140,4 +139,11 @@ public class InboundSseEventProcessor {
             return false;
         }
     }
+    
+    /**
+     * Create builder on-demand, without explicit event demarcation
+     */
+    private static Builder getOrCreate(final Builder builder) {
+        return (builder == null) ? new InboundSseEventImpl.Builder() : builder;
+    }
 }

http://git-wip-us.apache.org/repos/asf/cxf/blob/4cef686c/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
----------------------------------------------------------------------
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 9d28972..9db391f 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
@@ -57,6 +57,7 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.times;
@@ -65,26 +66,36 @@ import static org.mockito.Mockito.when;
 
 @RunWith(MockitoJUnitRunner.class)
 public class SseEventSourceImplTest extends Assert {
-    private static final String EVENT = "    event\n"
+    private static final String EVENT = "event: event\n"
         + "id: 1\n"
         + "data: test data\n"
         + "retry: 10000\n"
         + ": test comment\n"
         + "\n";
     
-    private static final String EVENT_NO_RETRY = "    event\n"
+    private static final String EVENT_JUST_DATA = "\n"
+        + "data: just test data\n"
+        + "\n";
+    
+    private static final String EVENT_JUST_NAME = "\n"
+        + "event: just name\n";
+    
+    private static final String EVENT_NO_RETRY = "event: event\n"
         + "id: 1\n"
         + "data: test data\n"
         + ": test comment\n"
         + "\n";
     
-    private static final String EVENT_BAD_RETRY = "    event\n"
+    private static final String EVENT_BAD_RETRY = "event: event\n"
         + "id: 1\n"
         + "data: test data\n"
         + "retry: blba\n"
         + ": test comment\n"
         + "\n";
     
+    private static final String EVENT_MIXED = EVENT_JUST_DATA + EVENT;
+    private static final String EVENT_BAD_NEW_LINES =  "\n\n\n\n\n\n";
+    
     @Rule
     public ExpectedException exception = ExpectedException.none();
     
@@ -239,6 +250,104 @@ public class SseEventSourceImplTest extends Assert {
             assertThat(events.get(0).getId(), equalTo("1"));
             assertThat(events.get(0).getReconnectDelay(), equalTo(10000L));
             assertThat(events.get(0).getComment(), equalTo("test comment"));
+            assertThat(events.get(0).readData(), equalTo("test data"));
+        }
+    }
+    
+    @Test
+    public void testNoReconnectAndJustDataEventIsReceived() throws InterruptedException,
IOException {
+        try (InputStream is = new ByteArrayInputStream(EVENT_JUST_DATA.getBytes(StandardCharsets.UTF_8)))
{
+            when(response.getStatus()).thenReturn(200);
+            when(response.readEntity(InputStream.class)).thenReturn(is);
+            
+            final List<InboundSseEvent> events = new ArrayList<>();
+            try (SseEventSource eventSource = withNoReconnect()) {
+                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(150);
+            }
+            
+            assertThat(events.size(), equalTo(1));
+            assertThat(events.get(0).getName(), nullValue());
+            assertThat(events.get(0).readData(), equalTo("just test data"));
+        }
+    }
+    
+    @Test
+    public void testNoReconnectAndJustEventNameIsReceived() throws InterruptedException,
IOException {
+        try (InputStream is = new ByteArrayInputStream(EVENT_JUST_NAME.getBytes(StandardCharsets.UTF_8)))
{
+            when(response.getStatus()).thenReturn(200);
+            when(response.readEntity(InputStream.class)).thenReturn(is);
+            
+            final List<InboundSseEvent> events = new ArrayList<>();
+            try (SseEventSource eventSource = withNoReconnect()) {
+                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(150);
+            }
+            
+            assertThat(events.size(), equalTo(1));
+            assertThat(events.get(0).getName(), equalTo("just name"));
+        }
+    }
+    
+    @Test
+    public void testNoReconnectAndMixedEventsAreReceived() throws InterruptedException, IOException
{
+        try (InputStream is = new ByteArrayInputStream(EVENT_MIXED.getBytes(StandardCharsets.UTF_8)))
{
+            when(response.getStatus()).thenReturn(200);
+            when(response.readEntity(InputStream.class)).thenReturn(is);
+            
+            final List<InboundSseEvent> events = new ArrayList<>();
+            try (SseEventSource eventSource = withNoReconnect()) {
+                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(150);
+            }
+            
+            assertThat(events.size(), equalTo(2));
+            assertThat(events.get(0).getName(), nullValue());
+            assertThat(events.get(0).readData(), equalTo("just test data"));
+            assertThat(events.get(1).getId(), equalTo("1"));
+            assertThat(events.get(1).getReconnectDelay(), equalTo(10000L));
+            assertThat(events.get(1).getComment(), equalTo("test comment"));
+            assertThat(events.get(1).readData(), equalTo("test data"));
+        }
+    }
+    
+    @Test
+    public void testNoReconnectAndNoEventsAreDetected() throws InterruptedException, IOException
{
+        try (InputStream is = new ByteArrayInputStream(EVENT_BAD_NEW_LINES.getBytes(StandardCharsets.UTF_8)))
{
+            when(response.getStatus()).thenReturn(200);
+            when(response.readEntity(InputStream.class)).thenReturn(is);
+            
+            final List<InboundSseEvent> events = new ArrayList<>();
+            try (SseEventSource eventSource = withNoReconnect()) {
+                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(150);
+            }
+            
+            assertThat(events.size(), equalTo(0));
         }
     }
     
@@ -269,8 +378,10 @@ public class SseEventSourceImplTest extends Assert {
             assertThat(events.size(), equalTo(2));
             assertThat(events.get(0).getId(), equalTo("1"));
             assertThat(events.get(0).getComment(), equalTo("test comment"));
+            assertThat(events.get(0).readData(), equalTo("test data"));
             assertThat(events.get(1).getId(), equalTo("1"));
             assertThat(events.get(1).getComment(), equalTo("test comment"));
+            assertThat(events.get(1).readData(), equalTo("test data"));
         } finally {
             for (final InputStream is: closeables) {
                 is.close();
@@ -323,6 +434,7 @@ public class SseEventSourceImplTest extends Assert {
             assertThat(events.get(0).getId(), equalTo("1"));
             assertThat(events.get(0).getReconnectDelay(), equalTo(-1L));
             assertThat(events.get(0).getComment(), equalTo("test comment"));
+            assertThat(events.get(0).readData(), equalTo("test data"));
         }
     }
 


Mime
View raw message