Add TCK model test for CRDT entities (#491)

java-support/src/main/java/io/cloudstate/javasupport/crdt/CrdtContext.java

@@ -31,4 +31,18 @@ public interface CrdtContext extends EntityContext {
    *     </code> type.
    */
   <T extends Crdt> Optional<T> state(Class<T> crdtClass) throws IllegalStateException;
+
+  /**
+   * Get the current write consistency setting for replication of CRDT state.
+   *
+   * @return the current write consistency
+   */
+  WriteConsistency getWriteConsistency();
+
+  /**
+   * Set the write consistency setting for replication of CRDT state.
+   *
+   * @param writeConsistency the new write consistency to use
+   */
+  void setWriteConsistency(WriteConsistency writeConsistency);
 }

java-support/src/main/java/io/cloudstate/javasupport/crdt/WriteConsistency.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2019 Lightbend Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.cloudstate.javasupport.crdt;
+
+/** Write consistency setting for replication of state updates for CRDTs. */
+public enum WriteConsistency {
+  /**
+   * Updates will only be written to the local replica immediately, and then asynchronously
+   * distributed to other replicas in the background.
+   */
+  LOCAL,
+
+  /**
+   * Updates will be written immediately to a majority of replicas, and then asynchronously
+   * distributed to remaining replicas in the background.
+   */
+  MAJORITY,
+
+  /** Updates will be written immediately to all replicas. */
+  ALL
+}

java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/AnnotationBasedCrdtSupport.scala

@@ -45,7 +45,8 @@ import io.cloudstate.javasupport.crdt.{
   StreamCancelledContext,
   StreamedCommandContext,
   SubscriptionContext,
-  Vote
+  Vote,
+  WriteConsistency
 }
 import io.cloudstate.javasupport.impl.ReflectionHelper.{
   CommandHandlerInvoker,
@@ -288,6 +289,9 @@ private final class AdaptedStreamedCommandContext(val delegate: StreamedCommandC
   override def newLWWRegister[T](value: T): LWWRegister[T] = delegate.newLWWRegister(value)
   override def newORMap[K, V <: Crdt](): ORMap[K, V] = delegate.newORMap()
   override def newVote(): Vote = delegate.newVote()
+
+  override def getWriteConsistency: WriteConsistency = delegate.getWriteConsistency
+  override def setWriteConsistency(consistency: WriteConsistency): Unit = delegate.setWriteConsistency(consistency)
 }
 
 private final class EntityConstructorInvoker(constructor: Constructor[_]) extends (CrdtCreationContext => AnyRef) {

java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/CrdtImpl.scala

@@ -31,7 +31,8 @@ import io.cloudstate.javasupport.crdt.{
   CrdtEntityFactory,
   StreamCancelledContext,
   StreamedCommandContext,
-  SubscriptionContext
+  SubscriptionContext,
+  WriteConsistency
 }
 import io.cloudstate.javasupport.impl.{
   AbstractClientActionContext,
@@ -139,7 +140,9 @@ class CrdtImpl(system: ActorSystem, services: Map[String, CrdtStatefulService],
         ctx.deactivate()
       }
     }
-    verifyNoDelta("creation")
+    // Doesn't make sense to verify that there's no delta here.
+    // LWWRegister can have its value set on creation, so there may be a delta.
+    // verifyNoDelta("creation")
 
     private def verifyNoDelta(scope: String): Unit =
       crdt match {
@@ -399,6 +402,19 @@ class CrdtImpl(system: ActorSystem, services: Map[String, CrdtStatefulService],
       override final def entityId(): String = EntityRunner.this.entityId
 
       override def serviceCallFactory(): ServiceCallFactory = rootContext.serviceCallFactory()
+
+      private var writeConsistency = WriteConsistency.LOCAL
+
+      override final def getWriteConsistency: WriteConsistency = writeConsistency
+
+      override final def setWriteConsistency(writeConsistency: WriteConsistency): Unit =
+        this.writeConsistency = writeConsistency
+
+      def crdtWriteConsistency: CrdtWriteConsistency = writeConsistency match {
+        case WriteConsistency.LOCAL => CrdtWriteConsistency.LOCAL
+        case WriteConsistency.MAJORITY => CrdtWriteConsistency.MAJORITY
+        case WriteConsistency.ALL => CrdtWriteConsistency.ALL
+      }
     }
 
     trait CapturingCrdtFactory extends AbstractCrdtFactory with AbstractCrdtContext {
@@ -430,11 +446,11 @@ class CrdtImpl(system: ActorSystem, services: Map[String, CrdtStatefulService],
       final def createCrdtAction(): Option[CrdtStateAction] = crdt match {
         case Some(c) =>
           if (deleted) {
-            Some(CrdtStateAction(action = CrdtStateAction.Action.Delete(CrdtDelete())))
+            Some(CrdtStateAction(action = CrdtStateAction.Action.Delete(CrdtDelete()), crdtWriteConsistency))
           } else if (c.hasDelta) {
             val delta = c.delta
             c.resetDelta()
-            Some(CrdtStateAction(action = CrdtStateAction.Action.Update(CrdtDelta(delta))))
+            Some(CrdtStateAction(action = CrdtStateAction.Action.Update(CrdtDelta(delta)), crdtWriteConsistency))
           } else {
             None
           }

java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/LWWRegisterImpl.scala

@@ -33,9 +33,11 @@ private[crdt] final class LWWRegisterImpl[T](anySupport: AnySupport) extends Int
   override def set(value: T, clock: LWWRegister.Clock, customClockValue: Long): T = {
     Objects.requireNonNull(value)
     val old = this.value
-    if (this.value != value) {
+    if (this.value != value || this.clock != clock || this.customClockValue != customClockValue) {
       deltaValue = Some(anySupport.encodeScala(value))
       this.value = value
+      this.clock = clock
+      this.customClockValue = customClockValue
     }
     old
   }
@@ -55,6 +57,7 @@ private[crdt] final class LWWRegisterImpl[T](anySupport: AnySupport) extends Int
 
   override val applyDelta = {
     case CrdtDelta.Delta.Lwwregister(LWWRegisterDelta(Some(any), _, _, _)) =>
+      resetDelta()
       this.value = anySupport.decode(any).asInstanceOf[T]
   }
 

java-support/src/test/scala/io/cloudstate/javasupport/impl/crdt/AnnotationBasedCrdtSupportSpec.scala

@@ -23,7 +23,7 @@ import com.google.protobuf.any.{Any => ScalaPbAny}
 import com.google.protobuf.{ByteString, Any => JavaPbAny}
 import io.cloudstate.javasupport.impl.{AnySupport, ResolvedServiceMethod, ResolvedType}
 import io.cloudstate.javasupport._
-import io.cloudstate.javasupport.crdt.{Crdt, CrdtContext, CrdtCreationContext, CrdtEntity, Vote}
+import io.cloudstate.javasupport.crdt.{Crdt, CrdtContext, CrdtCreationContext, CrdtEntity, Vote, WriteConsistency}
 import org.scalatest.{Matchers, WordSpec}
 
 import scala.compat.java8.OptionConverters._
@@ -58,6 +58,8 @@ class AnnotationBasedCrdtSupportSpec extends WordSpec with Matchers {
           s"The current ${wrongType} CRDT state doesn't match requested type of ${crdtType.getSimpleName}"
         )
     }
+    override def getWriteConsistency: WriteConsistency = WriteConsistency.LOCAL
+    override def setWriteConsistency(writeConsistency: WriteConsistency): Unit = ()
   }
 
   object WrappedResolvedType extends ResolvedType[Wrapped] {

java-support/tck/src/main/java/io/cloudstate/javasupport/tck/JavaSupportTck.java

@@ -22,10 +22,13 @@
 import io.cloudstate.javasupport.tck.model.valuebased.ValueEntityTwoEntity;
 import io.cloudstate.javasupport.tck.model.action.ActionTckModelBehavior;
 import io.cloudstate.javasupport.tck.model.action.ActionTwoBehavior;
+import io.cloudstate.javasupport.tck.model.crdt.CrdtTckModelEntity;
+import io.cloudstate.javasupport.tck.model.crdt.CrdtTwoEntity;
 import io.cloudstate.javasupport.tck.model.eventsourced.EventSourcedTckModelEntity;
 import io.cloudstate.javasupport.tck.model.eventsourced.EventSourcedTwoEntity;
 import io.cloudstate.samples.shoppingcart.ShoppingCartEntity;
 import io.cloudstate.tck.model.Action;
+import io.cloudstate.tck.model.Crdt;
 import io.cloudstate.tck.model.Eventsourced;
 import io.cloudstate.tck.model.valueentity.Valueentity;
 
@@ -51,6 +54,11 @@ public static final void main(String[] args) throws Exception {
             ShoppingCartEntity.class,
             Shoppingcart.getDescriptor().findServiceByName("ShoppingCart"),
             com.example.valueentity.shoppingcart.persistence.Domain.getDescriptor())
+        .registerCrdtEntity(
+            CrdtTckModelEntity.class,
+            Crdt.getDescriptor().findServiceByName("CrdtTckModel"),
+            Crdt.getDescriptor())
+        .registerCrdtEntity(CrdtTwoEntity.class, Crdt.getDescriptor().findServiceByName("CrdtTwo"))
         .registerEventSourcedEntity(
             EventSourcedTckModelEntity.class,
             Eventsourced.getDescriptor().findServiceByName("EventSourcedTckModel"),