Refactoring to remove the sse package

Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>

Author: ehsavoie

vertx-web-client/src/main/asciidoc/index.adoc

@@ -374,19 +374,11 @@ WARNING: this is only valid for the response decoded as a buffer.
 
 === Server-Sent Events
 
-Using a specific body codec: {@link io.vertx.ext.web.codec.sse.SseBodyCodec}, you can decode a Server-Sent Events stream into a list of events:
-
+Using a specific body codec: {@link io.vertx.ext.web.codec.impl.SseBodyCodec}, you can decode a Server-Sent Events stream into a list of events:
 [source,$lang]
 ----
-client.get("/eventsource").as(SseBodyCodec.sseStream(stream -> {
-      stream.handler(event -> //process the event);
-      stream.endHandler(v -> {
-        async.complete();
-      });
-    })).send();
+{@link examples.WebClientExamples#receiveResponseAsServerSentEvents(Vertx vertx, int servicePort)}
 ----
-
-
 [[http-response-expectations]]
 === Response expectations
 

vertx-web-client/src/main/java/examples/WebClientExamples.java

@@ -38,15 +38,11 @@
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.client.WebClientOptions;
 import io.vertx.ext.web.codec.BodyCodec;
-import io.vertx.ext.web.codec.sse.SseBodyCodec;
-import io.vertx.ext.web.codec.sse.SseEvent;
 import io.vertx.ext.web.multipart.MultipartForm;
 import io.vertx.uritemplate.ExpandOptions;
 import io.vertx.uritemplate.UriTemplate;
-import java.util.ArrayList;
 
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -766,7 +762,7 @@ public static void clientSideConsistentHashing(Vertx vertx, int servicePort) {
     server.listen(servicePort);
   }
 
-  public static void receiveServerSideEvents(Vertx vertx, int servicePort) {
+  public static void receiveResponseAsServerSentEvents(Vertx vertx, int servicePort) {
     WebClient client = WebClient.create(vertx, new WebClientOptions().setDefaultPort(servicePort).setDefaultHost("localhost"));
 
     HttpServer server = vertx.createHttpServer()
@@ -795,9 +791,9 @@ public void handle(Long timerId) {
           });
         });
     server.listen(servicePort);
-    final List<SseEvent> events = new ArrayList<>();
-    client.get("/basic?count=5").as(SseBodyCodec.sseStream(stream -> {
-      stream.handler(events::add);
+
+    client.get("/basic?count=5").as(BodyCodec.sseStream(stream -> {
+      stream.handler(v -> System.out.println("Event received " + v));
       stream.endHandler(v ->  System.out.println("End of stream " + v));
       })).send().expecting(HttpResponseExpectation.SC_OK)
           .onSuccess(res ->

vertx-web-client/src/test/java/io/vertx/ext/web/client/tests/SseClientTest.java

@@ -8,8 +8,7 @@
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.client.WebClientOptions;
-import io.vertx.ext.web.codec.sse.SseBodyCodec;
-import io.vertx.ext.web.codec.sse.SseEvent;
+import io.vertx.ext.web.codec.spi.SseEvent;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;

vertx-web-common/src/main/java/io/vertx/ext/web/codec/BodyCodec.java

@@ -17,15 +17,19 @@
 
 import io.vertx.codegen.annotations.GenIgnore;
 import io.vertx.codegen.annotations.VertxGen;
+import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.parsetools.JsonParser;
+import io.vertx.core.streams.ReadStream;
 import io.vertx.core.streams.WriteStream;
 import io.vertx.ext.web.codec.impl.BodyCodecImpl;
 import io.vertx.ext.web.codec.impl.JsonStreamBodyCodec;
+import io.vertx.ext.web.codec.impl.SseBodyCodec;
 import io.vertx.ext.web.codec.impl.StreamingBodyCodec;
 import io.vertx.ext.web.codec.spi.BodyStream;
+import io.vertx.ext.web.codec.impl.SseEventImpl;
 
 import java.util.function.Function;
 
@@ -137,6 +141,16 @@ static BodyCodec<Void> jsonStream(JsonParser parser) {
     return new JsonStreamBodyCodec(parser);
   }
 
+  /**
+   * A body codec that parse the response as a Server-SentEvent stream.
+   *
+   * @param handler the non-null hander to handle the stream of Server-Sent Events.
+   * @return the body codec for a write stream
+   */
+  static BodyCodec<Void> sseStream(Handler<ReadStream<SseEventImpl>> handler) {
+    return new SseBodyCodec(handler);
+  }
+
   /**
    * Create the {@link BodyStream}.
    * <p>

…/ext/web/codec/sse/SseBodyCodecImpl.java …rtx/ext/web/codec/impl/SseBodyCodec.javavertx-web-common/src/main/java/io/vertx/ext/web/codec/sse/SseBodyCodecImpl.java renamed to vertx-web-common/src/main/java/io/vertx/ext/web/codec/impl/SseBodyCodec.java vertx-web-common/src/main/java/io/vertx/ext/web/codec/sse/SseBodyCodecImpl.java renamed to vertx-web-common/src/main/java/io/vertx/ext/web/codec/impl/SseBodyCodec.java

@@ -1,22 +1,23 @@
-package io.vertx.ext.web.codec.sse;
+package io.vertx.ext.web.codec.impl;
 
 import io.vertx.codegen.annotations.Nullable;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.streams.ReadStream;
 import io.vertx.core.streams.WriteStream;
+import io.vertx.ext.web.codec.BodyCodec;
 import io.vertx.ext.web.codec.spi.BodyStream;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * A codec for processing and decoding Server Sent Event from streaming HTTP body content .
  */
-public class SseBodyCodecImpl implements SseBodyCodec<Void> {
+public class SseBodyCodec implements BodyCodec<Void> {
 
-  private final Handler<ReadStream<SseEvent>> handler;
+  private final Handler<ReadStream<SseEventImpl>> handler;
 
-  SseBodyCodecImpl(Handler<ReadStream<SseEvent>> handler) {
+  public SseBodyCodec(Handler<ReadStream<SseEventImpl>> handler) {
     this.handler = handler;
   }
 
@@ -27,12 +28,12 @@ public BodyStream<Void> stream() throws Exception {
     return stream;
   }
 
-  static class SseBodyStream implements BodyStream<Void>, ReadStream<SseEvent> {
+  static class SseBodyStream implements BodyStream<Void>, ReadStream<SseEventImpl> {
 
     private static final int LOW_WATERMARK = 1024;
     private static final int HIGH_WATERMARK = 4 * 1024;
 
-    private Handler<SseEvent> handler;
+    private Handler<SseEventImpl> handler;
     private Handler<Void> endHandler;
     private final AtomicLong demand = new AtomicLong(Long.MAX_VALUE);
     private Buffer content = Buffer.buffer();
@@ -43,26 +44,26 @@ static class SseBodyStream implements BodyStream<Void>, ReadStream<SseEvent> {
     private volatile boolean failed;
 
     @Override
-    public ReadStream<SseEvent> handler(@Nullable Handler<SseEvent> handler) {
+    public ReadStream<SseEventImpl> handler(@Nullable Handler<SseEventImpl> handler) {
       this.handler = handler;
       return this;
     }
 
     @Override
-    public ReadStream<SseEvent> pause() {
+    public ReadStream<SseEventImpl> pause() {
       demand.set(0L);
       return this;
     }
 
     @Override
-    public ReadStream<SseEvent> resume() {
+    public ReadStream<SseEventImpl> resume() {
       demand.set(Long.MAX_VALUE);
       check();
       return this;
     }
 
     @Override
-    public ReadStream<SseEvent> fetch(long l) {
+    public ReadStream<SseEventImpl> fetch(long l) {
       if (l <= 0) {
         return this;
       }
@@ -72,13 +73,13 @@ public ReadStream<SseEvent> fetch(long l) {
     }
 
     @Override
-    public ReadStream<SseEvent> endHandler(@Nullable Handler<Void> handler) {
+    public ReadStream<SseEventImpl> endHandler(@Nullable Handler<Void> handler) {
       this.endHandler = handler;
       return this;
     }
 
-    SseEvent nextSseEvent() {
-      SseEvent.Builder eventBuilder = SseEvent.builder();
+    SseEventImpl nextSseEvent() {
+      SseEventImpl.Builder eventBuilder = SseEventImpl.builder();
       int lineStart = 0;
       byte[] bytes = content.getBytes();
 
@@ -110,7 +111,7 @@ void check() {
         if (demand.get() == 0L) {
           break;
         }
-        SseEvent event;
+        SseEventImpl event;
         try {
           synchronized (this) {
             event = nextSseEvent();
@@ -129,7 +130,7 @@ void check() {
           break;
         }
         demand.updateAndGet(d -> d == Long.MAX_VALUE ? d : d - 1);
-        Handler<SseEvent> h = handler;
+        Handler<SseEventImpl> h = handler;
         if (h != null) {
           h.handle(event);
         }

…io/vertx/ext/web/codec/sse/SseEvent.java …rtx/ext/web/codec/impl/SseEventImpl.javavertx-web-common/src/main/java/io/vertx/ext/web/codec/sse/SseEvent.java renamed to vertx-web-common/src/main/java/io/vertx/ext/web/codec/impl/SseEventImpl.java vertx-web-common/src/main/java/io/vertx/ext/web/codec/sse/SseEvent.java renamed to vertx-web-common/src/main/java/io/vertx/ext/web/codec/impl/SseEventImpl.java

@@ -1,20 +1,21 @@
-package io.vertx.ext.web.codec.sse;
+package io.vertx.ext.web.codec.impl;
 
 import io.vertx.codegen.annotations.DataObject;
+import io.vertx.ext.web.codec.spi.SseEvent;
 
 /**
  * This represents a Server-Sent Event.
  * @see https://html.spec.whatwg.org/multipage/server-sent-events.html
  */
 @DataObject
-public class SseEvent {
+public class SseEventImpl implements SseEvent {
 
   private final String id;
   private final String event;
   private final String data;
   private final int retry;
 
-  public SseEvent(String id, String event, String data, int retry) {
+  public SseEventImpl(String id, String event, String data, int retry) {
     this.id = id;
     this.event = event;
     this.data = data;
@@ -25,6 +26,7 @@ public SseEvent(String id, String event, String data, int retry) {
    * Returns the event id.
    * @return the event id.
    */
+  @Override
   public String id() {
     return id;
   }
@@ -33,6 +35,7 @@ public String id() {
    * Returns the type of the event.
    * @return the type of the event.
    */
+  @Override
   public String event() {
     return event;
   }
@@ -41,6 +44,7 @@ public String event() {
    * Returns the payload of the event.
    * @return the payload of the event.
    */
+  @Override
   public String data() {
     return data;
   }
@@ -49,6 +53,7 @@ public String data() {
    * Returns the reconnection time.
    * @return the reconnection time.
    */
+  @Override
   public int retry() {
     return retry;
   }
@@ -61,7 +66,7 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-    SseEvent sseEvent = (SseEvent) o;
+    SseEventImpl sseEvent = (SseEventImpl) o;
     return retry == sseEvent.retry
         && java.util.Objects.equals(id, sseEvent.id)
         && java.util.Objects.equals(event, sseEvent.event)
@@ -82,33 +87,33 @@ public String toString() {
         + '\n';
   }
 
-  public static Builder builder() {
+  static Builder builder() {
     return new Builder();
   }
 
-  public static class Builder {
+  static class Builder {
 
     private String id;
     private String event = "message";
     private StringBuilder data = new StringBuilder();
     private int retry;
 
-    public Builder id(String id) {
+    Builder id(String id) {
       this.id = id;
       return this;
     }
 
-    public Builder event(String event) {
+    Builder event(String event) {
       this.event = event;
       return this;
     }
 
-    public Builder data(String data) {
+    Builder data(String data) {
       this.data.append(data);
       return this;
     }
 
-    public Builder retry(int retry) {
+    Builder retry(int retry) {
       this.retry = retry;
       return this;
     }
@@ -159,8 +164,8 @@ private void processField(String field, String value) {
       }
     }
 
-    public SseEvent build() {
-      return new SseEvent(this.id, this.event, this.data.toString(), this.retry);
+    public SseEventImpl build() {
+      return new SseEventImpl(this.id, this.event, this.data.toString(), this.retry);
     }
   }
 }

vertx-web-common/src/main/java/io/vertx/ext/web/codec/spi/SseEvent.java

@@ -0,0 +1,36 @@
+package io.vertx.ext.web.codec.spi;
+
+import io.vertx.codegen.annotations.DataObject;
+
+/**
+ * This represents a Server-Sent Event.
+ * @see https://html.spec.whatwg.org/multipage/server-sent-events.html
+ */
+@DataObject
+public interface SseEvent {
+
+  /**
+   * Returns the payload of the event.
+   * @return the payload of the event.
+   */
+  String data();
+
+  /**
+   * Returns the type of the event.
+   * @return the type of the event.
+   */
+  String event();
+
+  /**
+   * Returns the event id.
+   * @return the event id.
+   */
+  String id();
+
+  /**
+   * Returns the reconnection time.
+   * @return the reconnection time.
+   */
+  int retry();
+    
+}