Vertx 4 (#206)

* Something wrong

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Update to Vert.x 4

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Removed wrong pom bits

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Updated readme

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Updated the sample code of vertx

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Avoid recreating the JsonFormat

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

Author: slinkydeveloper

examples/vertx/src/main/java/io/cloudevents/examples/vertx/SampleHTTPClient.java

@@ -5,9 +5,12 @@
 import io.cloudevents.core.message.MessageReader;
 import io.cloudevents.http.vertx.VertxMessageFactory;
 import io.cloudevents.jackson.JsonFormat;
+import io.vertx.core.Future;
 import io.vertx.core.Vertx;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.client.WebClient;
+
 import java.net.URI;
 import java.util.UUID;
 
@@ -24,7 +27,7 @@ public static void main(String[] args) {
         final String eventSink = args[0];
 
         final Vertx vertx = Vertx.vertx();
-        final HttpClient httpClient = vertx.createHttpClient();
+        final WebClient webClient = WebClient.create(vertx);
 
         // Create an event template to set basic CloudEvent attributes.
         CloudEventBuilder eventTemplate = CloudEventBuilder.v1()
@@ -33,39 +36,28 @@ public static void main(String[] args) {
 
         // Send NUM_EVENTS events.
         for (int i = 0; i < NUM_EVENTS; i++) {
-
-            // create HTTP request.
-            final HttpClientRequest request = httpClient.postAbs(eventSink)
-                .handler(response -> {
-
-                    // We need to read the event from the HTTP response we get, so create a MessageReader.
-                    VertxMessageFactory.createReader(response)
-                        // Covert the MessageReader to a CloudEvent.
-                        .map(MessageReader::toEvent)
-                        // Print out the event.
-                        .onSuccess(System.out::println)
-                        .onFailure(System.err::println);
-
-                })
-                .exceptionHandler(System.err::println);
-
-            String id = UUID.randomUUID().toString();
             String data = "Event number " + i;
 
             // Create the event starting from the template
             final CloudEvent event = eventTemplate.newBuilder()
-                .withId(id)
+                .withId(UUID.randomUUID().toString())
                 .withData("text/plain", data.getBytes())
                 .build();
 
+            Future<HttpResponse<Buffer>> responseFuture;
             // We need to write the event to the request, so create a MessageWriter.
             if (i % 2 == 0) {
-                VertxMessageFactory.createWriter(request)
+                responseFuture = VertxMessageFactory.createWriter(webClient.postAbs(eventSink))
                     .writeBinary(event); // Use binary mode.
             } else {
-                VertxMessageFactory.createWriter(request)
-                    .writeStructured(event, new JsonFormat()); // Use structured mode.
+                responseFuture = VertxMessageFactory.createWriter(webClient.postAbs(eventSink))
+                    .writeStructured(event, JsonFormat.CONTENT_TYPE); // Use structured mode.
             }
+            responseFuture
+                .map(VertxMessageFactory::createReader) // Let's convert the response to message reader...
+                .map(MessageReader::toEvent) // ...then to event
+                .onSuccess(System.out::println) // Print the received message
+                .onFailure(System.err::println); // Print the eventual failure
         }
     }
 }

http/vertx/README.md

@@ -58,41 +58,34 @@ public class CloudEventServerVerticle extends AbstractVerticle {
 Below is a sample on how to use the client to send and receive a CloudEvent:
 
 ```java
+import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
-import io.cloudevents.core.message.MessageReader;
 import io.cloudevents.http.vertx.VertxMessageFactory;
-import io.cloudevents.CloudEvent;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpClient;
 import io.vertx.core.AbstractVerticle;
+import io.vertx.ext.web.client.WebClient;
+
 import java.net.URI;
 
 public class CloudEventClientVerticle extends AbstractVerticle {
 
   public void start() {
-    HttpClient client = vertx.createHttpClient();
-
-    HttpClientRequest request = client.postAbs("http://localhost:8080")
-        .handler(httpClientResponse -> {
-          VertxMessageFactory
-            .createReader(httpClientResponse)
-            .onComplete(result -> {
-              if (result.succeeded()) {
-                CloudEvent event = result.result().toEvent();
-              }
-          });
-        });
+    WebClient client = WebClient.create(vertx);
 
-    CloudEvent event = CloudEventBuilder.v1()
-      .withId("hello")
-      .withType("example.vertx")
-      .withSource(URI.create("http://localhost"))
-      .build();
+    CloudEvent reqEvent = CloudEventBuilder.v1()
+        .withId("hello")
+        .withType("example.vertx")
+        .withSource(URI.create("http://localhost"))
+        .build();
 
-    // Write request as binary
     VertxMessageFactory
-      .createWriter(request)
-      .writeBinary(event);
+        .createWriter(client.postAbs("http://localhost:8080"))
+        .writeBinary(reqEvent)
+        .onSuccess(res -> {
+          CloudEvent resEvent = VertxMessageFactory
+              .createReader(res)
+              .toEvent();
+        })
+        .onFailure(Throwable::printStackTrace);
   }
 }
 ```

http/vertx/pom.xml

@@ -31,7 +31,7 @@
     <packaging>jar</packaging>
 
     <properties>
-        <vertx.version>3.9.2</vertx.version>
+        <vertx.version>4.0.0.Beta1</vertx.version>
         <module-name>io.cloudevents.http.vertx</module-name>
     </properties>
 
@@ -53,6 +53,11 @@
             <artifactId>vertx-core</artifactId>
             <version>${vertx.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-web-client</artifactId>
+            <version>${vertx.version}</version>
+        </dependency>
 
         <!-- Test deps -->
         <dependency>

http/vertx/src/main/java/io/cloudevents/http/vertx/VertxMessageFactory.java

@@ -7,19 +7,22 @@
 import io.cloudevents.core.message.impl.UnknownEncodingMessageReader;
 import io.cloudevents.http.vertx.impl.BinaryVertxMessageReaderImpl;
 import io.cloudevents.http.vertx.impl.CloudEventsHeaders;
-import io.cloudevents.http.vertx.impl.VertxHttpClientRequestMessageWriterImpl;
+import io.cloudevents.http.vertx.impl.VertxWebClientRequestMessageWriterImpl;
 import io.cloudevents.http.vertx.impl.VertxHttpServerResponseMessageWriterImpl;
 import io.cloudevents.rw.CloudEventWriter;
 import io.vertx.core.*;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.*;
+import io.vertx.core.http.HttpHeaders;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.http.HttpServerResponse;
+import io.vertx.ext.web.client.HttpRequest;
+import io.vertx.ext.web.client.HttpResponse;
 
 import javax.annotation.ParametersAreNonnullByDefault;
-import java.util.function.Consumer;
 
 /**
  * This class provides a collection of methods to create {@link io.cloudevents.core.message.MessageReader}
- * and {@link io.cloudevents.core.message.MessageWriter} for Vert.x HTTP Client and Server.
+ * and {@link io.cloudevents.core.message.MessageWriter} for Vert.x HTTP Server and Web Client.
  */
 @ParametersAreNonnullByDefault
 public final class VertxMessageFactory {
@@ -52,7 +55,17 @@ public static MessageReader createReader(MultiMap headers, Buffer body) throws I
      * @return
      */
     public static Future<MessageReader> createReader(HttpServerRequest request) {
-        return createReader(request.headers(), request::exceptionHandler, request::bodyHandler);
+        Promise<MessageReader> prom = Promise.promise();
+
+        request.exceptionHandler(prom::tryFail);
+        request.bodyHandler(b -> {
+            try {
+                prom.complete(createReader(request.headers(), b));
+            } catch (IllegalArgumentException e) {
+                prom.fail(e);
+            }
+        });
+        return prom.future();
     }
 
     /**
@@ -65,42 +78,13 @@ public static void createReader(HttpServerRequest request, Handler<AsyncResult<M
     }
 
     /**
-     * Build a {@link MessageReader} starting from an {@link HttpClientResponse}
+     * Build a {@link MessageReader} starting from an {@link io.vertx.ext.web.client.HttpResponse}
      *
      * @param response
      * @return
      */
-    public static Future<MessageReader> createReader(HttpClientResponse response) {
-        return createReader(response.headers(), response::exceptionHandler, response::bodyHandler);
-    }
-
-    /**
-     * @see #createReader(HttpClientResponse)
-     *
-     * @param response
-     * @param handler
-     */
-    public static void createReader(HttpClientResponse response, Handler<AsyncResult<MessageReader>> handler) {
-        createReader(response).onComplete(handler);
-    }
-
-    private static Future<MessageReader> createReader(
-        MultiMap headers,
-        Consumer<Handler<Throwable>> fail,
-        Consumer<Handler<Buffer>> success) {
-
-        Promise<MessageReader> prom = Promise.promise();
-
-        fail.accept(prom::tryFail);
-        success.accept(b -> {
-            try {
-                prom.complete(createReader(headers, b));
-            } catch (IllegalArgumentException e) {
-                prom.fail(e);
-            }
-        });
-        return prom.future();
-
+    public static MessageReader createReader(HttpResponse<Buffer> response) {
+        return createReader(response.headers(), response.body());
     }
 
     /**
@@ -115,13 +99,14 @@ public static MessageWriter<CloudEventWriter<HttpServerResponse>, HttpServerResp
     }
 
     /**
-     * Creates a {@link MessageWriter} that can write both structured and binary messages to a {@link HttpClientRequest}.
-     * When the writer finished to write the {@link MessageReader}, the request is ended with {@link HttpClientRequest#end(io.vertx.core.buffer.Buffer)}
+     * Creates a {@link MessageWriter} that can write both structured and binary messages to a {@link io.vertx.ext.web.client.HttpRequest}.
+     * When the writer finished to write the {@link MessageReader}, the request is sent with {@link io.vertx.ext.web.client.HttpRequest#sendBuffer(Buffer)}
+     * and it returns the {@link Future} containing the response.
      *
      * @param req the request to write
      * @return the message writer
      */
-    public static MessageWriter<CloudEventWriter<HttpClientRequest>, HttpClientRequest> createWriter(HttpClientRequest req) {
-        return new VertxHttpClientRequestMessageWriterImpl(req);
+    public static MessageWriter<CloudEventWriter<Future<HttpResponse<Buffer>>>, Future<HttpResponse<Buffer>>> createWriter(HttpRequest<Buffer> req) {
+        return new VertxWebClientRequestMessageWriterImpl(req);
     }
 }

http/vertx/src/main/java/io/cloudevents/http/vertx/impl/VertxHttpClientRequestMessageWriterImpl.java renamed to http/vertx/src/main/java/io/cloudevents/http/vertx/impl/VertxWebClientRequestMessageWriterImpl.java

@@ -22,56 +22,56 @@
 import io.cloudevents.core.message.MessageWriter;
 import io.cloudevents.rw.CloudEventRWException;
 import io.cloudevents.rw.CloudEventWriter;
+import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpHeaders;
+import io.vertx.ext.web.client.HttpRequest;
+import io.vertx.ext.web.client.HttpResponse;
 
-public class VertxHttpClientRequestMessageWriterImpl implements MessageWriter<CloudEventWriter<HttpClientRequest>, HttpClientRequest>, CloudEventWriter<HttpClientRequest> {
+public class VertxWebClientRequestMessageWriterImpl implements MessageWriter<CloudEventWriter<Future<HttpResponse<Buffer>>>, Future<HttpResponse<Buffer>>>, CloudEventWriter<Future<HttpResponse<Buffer>>> {
 
-    private final HttpClientRequest request;
+    private final HttpRequest<Buffer> request;
 
-    public VertxHttpClientRequestMessageWriterImpl(HttpClientRequest request) {
+    public VertxWebClientRequestMessageWriterImpl(HttpRequest<Buffer> request) {
         this.request = request;
     }
 
     // Binary visitor factory
 
     @Override
-    public CloudEventWriter<HttpClientRequest> create(SpecVersion version) {
-        this.request.putHeader(CloudEventsHeaders.SPEC_VERSION, version.toString());
+    public CloudEventWriter<Future<HttpResponse<Buffer>>> create(SpecVersion version) {
+        this.request.headers().add(CloudEventsHeaders.SPEC_VERSION, version.toString());
         return this;
     }
 
     // Binary visitor
 
     @Override
     public void setAttribute(String name, String value) throws CloudEventRWException {
-        this.request.putHeader(CloudEventsHeaders.ATTRIBUTES_TO_HEADERS.get(name), value);
+        this.request.headers().add(CloudEventsHeaders.ATTRIBUTES_TO_HEADERS.get(name), value);
     }
 
     @Override
     public void setExtension(String name, String value) throws CloudEventRWException {
-        this.request.putHeader("ce-" + name, value);
+        this.request.headers().add("ce-" + name, value);
     }
 
     @Override
-    public HttpClientRequest end(byte[] value) throws CloudEventRWException {
-        this.request.end(Buffer.buffer(value));
-        return this.request;
+    public Future<HttpResponse<Buffer>> end(byte[] value) throws CloudEventRWException {
+        return this.request.sendBuffer(Buffer.buffer(value));
     }
 
     @Override
-    public HttpClientRequest end() {
-        this.request.end();
-        return this.request;
+    public Future<HttpResponse<Buffer>> end() {
+        return this.request.send();
     }
 
     // Structured visitor
 
     @Override
-    public HttpClientRequest setEvent(EventFormat format, byte[] value) throws CloudEventRWException {
-        this.request.putHeader(HttpHeaders.CONTENT_TYPE, format.serializedContentType());
-        this.request.end(Buffer.buffer(value));
-        return this.request;
+    public Future<HttpResponse<Buffer>> setEvent(EventFormat format, byte[] value) throws CloudEventRWException {
+        this.request.headers().add(HttpHeaders.CONTENT_TYPE, format.serializedContentType());
+        return this.request.sendBuffer(Buffer.buffer(value));
     }
 }