Extract crud module for clarity

README.md

@@ -49,11 +49,12 @@ For demonstration purposes readside module also implements a direct CRUD-like ac
 
 ```
 
-curl -H "Content-Type: application/json" -X POST -d '{"message": "Hi"}' http://localhost:9000/api/readside/hello/Alice
+curl -H "Content-Type: application/json" -X POST -d '{"message": "Hi"}' http://localhost:9000/crud-api/hello/Alice
 { "done" : true }✔
 
-curl http://localhost:9000/api/readside/hello/Alice
-Hi, Alice!✔
+curl http://localhost:9000/crud-api/hello/Alice
+Hello, Alice!✔
+
 ```
 
 Couchbase

crud-api/pom.xml

@@ -0,0 +1,26 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+                      http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.lightbend</groupId>
+        <artifactId>readside</artifactId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>crud-api</artifactId>
+
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.lightbend.lagom</groupId>
+            <artifactId>lagom-javadsl-api_${scala.binary.version}</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+    </dependencies>
+</project>

crud-api/src/main/java/com/lightbend/readside/api/GreetingMessage.java

@@ -0,0 +1,18 @@
+package com.lightbend.readside.api;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.google.common.base.Preconditions;
+import lombok.Value;
+
+@Value
+@JsonDeserialize
+public final class GreetingMessage {
+
+  public final String message;
+
+  @JsonCreator
+  public GreetingMessage(String message) {
+    this.message = Preconditions.checkNotNull(message, "message");
+  }
+}

crud-api/src/main/java/com/lightbend/readside/api/ReadSideService.java

@@ -0,0 +1,29 @@
+package com.lightbend.readside.api;
+
+import static com.lightbend.lagom.javadsl.api.Service.named;
+import static com.lightbend.lagom.javadsl.api.Service.pathCall;
+
+import akka.Done;
+import akka.NotUsed;
+import com.lightbend.lagom.javadsl.api.Descriptor;
+import com.lightbend.lagom.javadsl.api.Service;
+import com.lightbend.lagom.javadsl.api.ServiceCall;
+
+public interface ReadSideService extends Service {
+
+  /**
+   * Example: curl http://localhost:9000/api/readside/hello/Alice
+   */
+  ServiceCall<NotUsed, String> hello(String id);
+
+  ServiceCall<GreetingMessage, Done> useGreeting(String id);
+
+    @Override
+  default Descriptor descriptor() {
+    return named("crud")
+            .withCalls(
+                pathCall("/crud-api/hello/:id",  this::hello),
+                pathCall("/crud-api/hello/:id", this::useGreeting)
+            ).withAutoAcl(true);
+  }
+}

crud-impl/pom.xml

@@ -0,0 +1,67 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+                      http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.lightbend</groupId>
+        <artifactId>readside</artifactId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>crud-impl</artifactId>
+
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>crud-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.lightbend.lagom</groupId>
+            <artifactId>lagom-javadsl-server_${scala.binary.version}</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.lightbend.lagom</groupId>
+            <artifactId>lagom-javadsl-kafka-client_${scala.binary.version}</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.lightbend.lagom</groupId>
+            <artifactId>lagom-logback_${scala.binary.version}</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.play</groupId>
+            <artifactId>play-akka-http-server_${scala.binary.version}</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.couchbase.client</groupId>
+            <artifactId>java-client</artifactId>
+            <version>2.5.9</version>
+        </dependency>
+        <dependency>
+            <groupId>com.lightbend.lagom</groupId>
+            <artifactId>lagom-javadsl-testkit_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.lightbend</groupId>
+            <artifactId>couchbase</artifactId>
+            <version>1.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>com.lightbend.lagom</groupId>
+                <artifactId>lagom-maven-plugin</artifactId>
+                <configuration>
+                    <lagomService>true</lagomService>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

crud-impl/src/main/java/com/lightbend/readside/impl/CrudModule.java

@@ -0,0 +1,17 @@
+package com.lightbend.readside.impl;
+
+import com.google.inject.AbstractModule;
+import com.lightbend.lagom.javadsl.server.ServiceGuiceSupport;
+import com.lightbend.readside.api.ReadSideService;
+
+/**
+ * The module that binds the ReadSideService so that it can be served.
+ */
+public class CrudModule extends AbstractModule implements ServiceGuiceSupport {
+  @Override
+  protected void configure() {
+    // Bind the ReadSideService service
+    bindService(ReadSideService.class, CrudServiceImpl.class);
+
+  }
+}

crud-impl/src/main/java/com/lightbend/readside/impl/CrudRepository.java

@@ -0,0 +1,78 @@
+package com.lightbend.readside.impl;
+
+import akka.Done;
+import com.couchbase.client.java.AsyncBucket;
+import com.couchbase.client.java.document.JsonDocument;
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.document.json.JsonObject;
+import com.couchbase.client.java.query.N1qlQuery;
+import com.couchbase.client.java.query.ParameterizedN1qlQuery;
+import com.couchbase.client.java.query.dsl.Expression;
+import com.couchbase.client.java.query.dsl.functions.ConditionalFunctions;
+import com.couchbase.client.java.query.dsl.path.UpdateSetPath;
+import com.lightbend.couchbase.Couchbase;
+import rx.Observable;
+import rx.Single;
+import utils.RxJava8Utils;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import static com.couchbase.client.java.query.Update.update;
+import static com.couchbase.client.java.query.dsl.Expression.*;
+import static com.couchbase.client.java.query.dsl.functions.ArrayFunctions.arrayIfNull;
+import static com.couchbase.client.java.query.dsl.functions.ArrayFunctions.arrayPrepend;
+import static com.couchbase.client.java.query.dsl.functions.ConditionalFunctions.*;
+
+@Singleton
+public class CrudRepository {
+
+    private final Couchbase couchbase;
+
+    @Inject
+    public CrudRepository(Couchbase couchbase) {
+        this.couchbase = couchbase;
+    }
+
+    public CompletionStage<Done> updateMessage(String name, String message) {
+
+        AsyncBucket bucket = couchbase.getBucket();
+
+        JsonObject obj = JsonObject.create()
+                .put("messages", JsonArray.from(message))
+                .put("message", message);
+
+        String docId = userMessageDocId(name);
+        JsonDocument doc = JsonDocument.create(docId, obj);
+
+        String queryText = "UPDATE test USE KEYS $1 SET messages = ARRAY_PREPEND($2, IFNULL(messages, [])), message = $2;";
+        ParameterizedN1qlQuery query = N1qlQuery.parameterized(queryText, JsonArray.from(docId, message));
+
+        Observable<Done> result = bucket
+                .insert(doc).map(x -> Done.getInstance())
+                .onErrorResumeNext(e -> bucket.query(query).map(x -> Done.getInstance()));
+
+        return RxJava8Utils.fromSingleObservable(result);
+    }
+
+    private String userMessageDocId(String name) {
+        return "crud_user_messages:" + name;
+    }
+
+    public CompletionStage<Optional<String>> getMessage(String name) {
+
+        String docId = userMessageDocId(name);
+
+        AsyncBucket bucket = couchbase.getBucket();
+
+        Observable<Optional<String>> result = bucket
+                .get(docId)
+                .map(v -> Optional.ofNullable(v.content().getString("message")));
+
+        return RxJava8Utils.fromSingleOptOptObservable(result);
+    }
+}
+

crud-impl/src/main/java/com/lightbend/readside/impl/CrudServiceImpl.java

@@ -0,0 +1,32 @@
+package com.lightbend.readside.impl;
+
+import akka.Done;
+import akka.NotUsed;
+import com.lightbend.lagom.javadsl.api.ServiceCall;
+import com.lightbend.readside.api.GreetingMessage;
+import com.lightbend.readside.api.ReadSideService;
+
+import javax.inject.Inject;
+
+public class CrudServiceImpl implements ReadSideService {
+
+    private final CrudRepository repository;
+
+    @Inject
+    public CrudServiceImpl(CrudRepository repository) {
+        this.repository = repository;
+    }
+
+    @Override
+    public ServiceCall<NotUsed, String> hello(String name) {
+        return request ->
+            repository.getMessage(name).thenApply(message ->
+                String.format("%s, %s!", message.orElse("Hello (default)"), name)
+            );
+    }
+
+    @Override
+    public ServiceCall<GreetingMessage, Done> useGreeting(String id) {
+        return request -> repository.updateMessage(id, request.getMessage());
+    }
+}

crud-impl/src/main/resources/application.conf

@@ -0,0 +1,13 @@
+
+include "couchbase.conf"
+
+play.modules.enabled += com.lightbend.readside.impl.CrudModule
+
+# The properties below override Lagom default configuration with the recommended values for new projects.
+#
+# Lagom has not yet made these settings the defaults for backward-compatibility reasons.
+
+# Prefer 'ddata' over 'persistence' to share cluster sharding state for new projects.
+# See https://doc.akka.io/docs/akka/current/cluster-sharding.html#distributed-data-vs-persistence-mode
+akka.cluster.sharding.state-store-mode = ddata
+

crud-impl/src/main/resources/couchbase.conf

@@ -0,0 +1,7 @@
+
+couchbase {
+  nodes = ["couchbase://localhost:8091"]
+  bucket = "test"
+  username = "Administrator"
+  password = "test123"
+}

crud-impl/src/test/java/com/lightbend/readside/impl/N1qlDslTests.java

@@ -0,0 +1,28 @@
+package com.lightbend.readside.impl;
+
+import com.couchbase.client.java.document.json.JsonArray;
+import com.couchbase.client.java.query.dsl.path.UpdateSetPath;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static com.couchbase.client.java.query.Update.update;
+import static com.couchbase.client.java.query.dsl.Expression.x;
+import static com.couchbase.client.java.query.dsl.functions.ArrayFunctions.arrayPrepend;
+import static com.couchbase.client.java.query.dsl.functions.ConditionalFunctions.ifNull;
+
+public class N1qlDslTests {
+
+    @Test
+    public void test() {
+
+        UpdateSetPath updateSetPath =
+                update("test")
+                        .useKeys("$1")
+                        .set("messages", arrayPrepend(ifNull(x("messages"), x(JsonArray.empty())), x("$2")))
+                        .set("message", "$2");
+
+
+        Assert.assertEquals("UPDATE test USE KEYS $1 SET messages = ARRAY_PREPEND($2, IFNULL(messages, [])), message = $2;", updateSetPath.toString());
+
+    }
+}

hello-impl/src/main/java/com/lightbend/hello/impl/readside/ChangeEventProcessor.java

@@ -56,7 +56,8 @@ public CompletionStage<Offset> prepare(AggregateEventTag<HelloEvent> tag) {
     private CompletionStage<Done> updateByEvent(HelloEvent event, Offset offset, AggregateEventTag<HelloEvent> tag) {
         if (event instanceof HelloEvent.GreetingMessageChanged) {
             HelloEvent.GreetingMessageChanged evt = (HelloEvent.GreetingMessageChanged) event;
-            return repository.updateMessage(evt.name, evt.message)
+            return repository
+                    .updateMessage(evt.name, evt.message)
                     .thenCompose(v -> repository.updateOffset(tag, offset));
         }
         return CompletableFuture.completedFuture(Done.getInstance());

hello-impl/src/main/java/com/lightbend/hello/impl/readside/ReadSideRepository.java

@@ -65,7 +65,7 @@ public CompletionStage<Done> updateOffset(AggregateEventTag<HelloEvent> tag, Off
     }
 
     private String offsetDocId(AggregateEventTag<HelloEvent> tag) {
-        return "user_messages_with_offset:tag_offset:" + tag.tag();
+        return "hello_user_messages_with_offset:tag_offset:" + tag.tag();
     }
 
     public CompletionStage<Done> updateMessage(String name, String message) {
@@ -90,7 +90,7 @@ public CompletionStage<Done> updateMessage(String name, String message) {
     }
 
     private String userMessageDocId(String name) {
-        return "user_messages_with_offset:" + name;
+        return "hello_user_messages_with_offset:" + name;
     }
 
 }

pom.xml

@@ -11,6 +11,8 @@
 
     <modules>
         <module>couchbase</module>
+        <module>crud-api</module>
+        <module>crud-impl</module>
         <module>hello-api</module>
         <module>hello-impl</module>
         <module>readside-api</module>

readside-api/src/main/java/com/lightbend/readside/api/ReadSideService.java

@@ -16,14 +16,11 @@ public interface ReadSideService extends Service {
    */
   ServiceCall<NotUsed, String> hello(String id);
 
-  ServiceCall<GreetingMessage, Done> useGreeting(String id);
-
     @Override
   default Descriptor descriptor() {
     return named("readside")
             .withCalls(
-                pathCall("/api/readside/hello/:id",  this::hello),
-                pathCall("/api/readside/hello/:id", this::useGreeting)
+                pathCall("/readside-api/hello/:id",  this::hello)
             ).withAutoAcl(true);
   }
 }