Session preparation

barista-kafka/src/main/java/me/escoffier/quarkus/coffeeshop/Beverage.java

@@ -1,5 +1,8 @@
 package me.escoffier.quarkus.coffeeshop;
 
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
+@RegisterForReflection(methods = true, fields = true)
 public class Beverage {
 
     private String beverage;

barista-kafka/src/main/java/me/escoffier/quarkus/coffeeshop/KafkaBarista.java

@@ -29,15 +29,18 @@ public CompletionStage<String> prepare(String message) {
                 .thenApply(beverage -> PreparationState.ready(order, beverage));
     }
 
+
+
+
     private CompletionStage<Beverage> makeIt(Order order) {
         return CompletableFuture.supplyAsync(() -> {
             System.out.println("Preparing a " + order.getProduct());
-            nap();
+            prepare();
             return new Beverage(order, name);
         });
     }
 
-    private void nap() {
+    private void prepare() {
         try {
             Thread.sleep(random.nextInt(5000));
         } catch (InterruptedException e) {

barista-kafka/src/main/java/me/escoffier/quarkus/coffeeshop/Order.java

@@ -2,7 +2,7 @@
 
 import io.quarkus.runtime.annotations.RegisterForReflection;
 
-@RegisterForReflection
+@RegisterForReflection(methods = true, fields = true)
 public class Order {
 
     private String product;

barista-kafka/src/main/java/me/escoffier/quarkus/coffeeshop/PreparationState.java

@@ -1,8 +1,11 @@
 package me.escoffier.quarkus.coffeeshop;
 
+import io.quarkus.runtime.annotations.RegisterForReflection;
+
 import javax.json.bind.Jsonb;
 import javax.json.bind.JsonbBuilder;
 
+@RegisterForReflection
 public class PreparationState {
 
     Beverage beverage;

coffeeshop-service/draft.txt

@@ -0,0 +1,22 @@
+    @POST
+    @Path("/async")
+    public CompletionStage<Beverage> async(Order order) {
+        return barista.orderAsync(order.setOrderId(UUID.randomUUID().toString()));
+    }
+
+    @Inject
+    @Stream("orders")
+    Emitter<String> orders;
+
+    @Inject
+    @Stream("queue")
+    Emitter<String> states;
+
+    @POST
+    @Path("/messaging")
+    public Order messaging(Order order) {
+        order.setOrderId(UUID.randomUUID().toString());
+        states.send(PreparationState.queued(order));
+        orders.send(jsonb.toJson(order));
+        return order;
+    }

coffeeshop-service/src/main/java/me/escoffier/quarkus/coffeeshop/CoffeeShopResource.java

@@ -1,18 +1,20 @@
 package me.escoffier.quarkus.coffeeshop;
 
+import io.smallrye.reactive.messaging.annotations.Emitter;
+import io.smallrye.reactive.messaging.annotations.Stream;
 import me.escoffier.quarkus.coffeeshop.http.BaristaService;
-import me.escoffier.quarkus.coffeeshop.messaging.KafkaBaristas;
 import me.escoffier.quarkus.coffeeshop.model.Beverage;
 import me.escoffier.quarkus.coffeeshop.model.Order;
+import me.escoffier.quarkus.coffeeshop.model.PreparationState;
 import org.eclipse.microprofile.rest.client.inject.RestClient;
 
 import javax.inject.Inject;
+import javax.json.bind.Jsonb;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
 import java.util.UUID;
 import java.util.concurrent.CompletionStage;
 
@@ -21,6 +23,9 @@
 @Produces(MediaType.APPLICATION_JSON)
 public class CoffeeShopResource {
 
+    @Inject
+    Jsonb jsonb;
+
     @Inject
     @RestClient
     BaristaService barista;
@@ -32,14 +37,26 @@ public Beverage http(Order order) {
         return barista.order(order.setOrderId(UUID.randomUUID().toString()));
     }
 
-    @Inject
-    KafkaBaristas baristas;
+    @POST
+    @Path("/async")
+    public CompletionStage<Beverage> async(Order order) {
+        return barista.orderAsync(order.setOrderId(UUID.randomUUID().toString()));
+    }
+
+    @Inject @Stream("orders")
+    Emitter<String> orders;
+
+    @Inject @Stream("queue")
+    Emitter<String> queue;
 
     @POST
     @Path("/messaging")
-    public CompletionStage<Order> messaging(Order order) {
+    public Order messaging(Order order) {
         order.setOrderId(UUID.randomUUID().toString());
-        return baristas.order(order);
+        queue.send(PreparationState.queued(order));
+        orders.send(jsonb.toJson(order));
+        return order;
     }
 
+
 }

coffeeshop-service/src/main/java/me/escoffier/quarkus/coffeeshop/dashboard/BoardResource.java

@@ -20,7 +20,9 @@ public class BoardResource {
     @GET
     @Produces(MediaType.SERVER_SENT_EVENTS)
     public Publisher<String> getQueue() {
-        return queue.peek(s -> System.out.println("GOT: " + s)).buildRs();
+        return queue
+                .peek(s -> System.out.println("GOT: " + s))
+                .buildRs();
     }
 
 }

coffeeshop-service/src/main/java/me/escoffier/quarkus/coffeeshop/http/BaristaService.java

@@ -8,6 +8,7 @@
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
+import java.util.concurrent.CompletionStage;
 
 @Path("/barista")
 @RegisterRestClient
@@ -17,4 +18,11 @@ public interface BaristaService {
     @Produces("application/json")
     @Consumes("application/json")
     Beverage order(Order order);
+
+    @POST
+    @Produces("application/json")
+    @Consumes("application/json")
+    CompletionStage<Beverage> orderAsync(Order order);
+
+
 }

pom.xml

@@ -10,7 +10,7 @@
     <packaging>pom</packaging>
 
     <properties>
-        <quarkus.version>0.14.0</quarkus.version>
+        <quarkus.version>0.15.0</quarkus.version>
 
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>