Update Akka gRPC to 0.8.0 (#211)

* Update Akka gRPC to 0.8.0 Unfortunately running into akka/akka-grpc#628 now * Update to 0.8.1 * Update akka-grpc To a version that has akka/akka-grpc#943 to fix compilation errors. Compile/test now works, but assembly still has problems, and the PR needs a rebase. * Remove reflection proto Reflection is provided by akka-grpc. Perhaps we'd like to make the .proto available here for some reason, but we shouldn't generate code for it. * reuse writer, shorter codec negotiation * Update to published akka-grpc snapshot * Expect server reflection in the reflection in the TCK Co-authored-by: Arnout Engelen <[email protected]>
cloudstateio · Apr 30, 2020 · baa43a8 · baa43a8
1 parent 091142f
commit baa43a8
Show file tree

Hide file tree

Showing 43 changed files with 141 additions and 399 deletions.
diff --git a/build.sbt b/build.sbt
@@ -10,8 +10,8 @@ inThisBuild(
     version := dynverGitDescribeOutput.value.mkVersion(versionFmt, "latest"),
     dynver := sbtdynver.DynVer.getGitDescribeOutput(new Date).mkVersion(versionFmt, "latest"),
     scalaVersion := "2.12.11",
-    // Needed for the fixed HTTP/2 connection cleanup version of akka-http
-    resolvers += Resolver.bintrayRepo("akka", "snapshots"), // TODO: Remove once we're switching to akka-http 10.1.11
+    // Needed for the akka-grpc 0.8.4 snapshot
+    resolvers += Resolver.bintrayRepo("akka", "maven"), // TODO: Remove once we're switching to akka-grpc 0.8.5/1.0.0
     organizationName := "Lightbend Inc.",
     organizationHomepage := Some(url("https://lightbend.com")),
     startYear := Some(2019),
@@ -71,11 +71,6 @@ def common: Seq[Setting[_]] = Seq(
       de.heikoseeberger.sbtheader.FileType("proto") -> HeaderCommentStyle.cppStyleLineComment,
       de.heikoseeberger.sbtheader.FileType("js") -> HeaderCommentStyle.cStyleBlockComment
     ),
-  // Akka gRPC adds all protobuf files from the classpath to this, which we don't want because it includes
-  // all the Google protobuf files which are already compiled and on the classpath by ScalaPB. So we set it
-  // back to just our source directory.
-  PB.protoSources in Compile := Seq(),
-  PB.protoSources in Test := Seq(),
   // Akka gRPC overrides the default ScalaPB setting including the file base name, let's override it right back.
   akkaGrpcCodeGeneratorSettings := Seq(),
   excludeFilter in headerResources := HiddenFileFilter || GlobFilter("reflection.proto"),
@@ -422,18 +417,14 @@ lazy val `proxy-core` = (project in file("proxy/core"))
     },
     PB.protoSources in Compile ++= {
       val baseDir = (baseDirectory in ThisBuild).value / "protocols"
-      Seq(baseDir / "proxy", baseDir / "frontend", baseDir / "protocol", (sourceDirectory in Compile).value / "protos")
+      Seq(baseDir / "proxy", baseDir / "frontend", baseDir / "protocol")
+    },
+    PB.protoSources in Test ++= {
+      val baseDir = (baseDirectory in ThisBuild).value / "protocols"
+      Seq(baseDir / "frontend")
     },
     // For Google Cloud Pubsub API
     PB.protoSources in Compile += target.value / "protobuf_external" / "google" / "pubsub" / "v1",
-    // This adds the test/protos dir and enables the ProtocPlugin to generate protos in the Test scope
-    inConfig(Test)(
-      sbtprotoc.ProtocPlugin.protobufConfigSettings ++ Seq(
-        PB.protoSources ++= Seq(sourceDirectory.value / "protos"),
-        akkaGrpcCodeGeneratorSettings := Seq(),
-        akkaGrpcGeneratedSources := Seq(AkkaGrpc.Server, AkkaGrpc.Client)
-      )
-    ),
     javaAgents += "org.mortbay.jetty.alpn" % "jetty-alpn-agent" % "2.0.9" % "runtime;test",
     dockerSettings,
     fork in run := true,

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/CrdtImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/CrdtImpl.scala
@@ -80,7 +80,7 @@ class CrdtImpl(system: ActorSystem, services: Map[String, CrdtStatefulService],
   def handle(in: Source[CrdtStreamIn, NotUsed]): Source[CrdtStreamOut, NotUsed] =
     in.prefixAndTail(1)
       .flatMapConcat {
-        case (Seq(CrdtStreamIn(In.Init(init))), source) =>
+        case (Seq(CrdtStreamIn(In.Init(init), _)), source) =>
           source.via(runEntity(init))
         case _ =>
           // todo better error

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/FlagImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/FlagImpl.scala
@@ -28,12 +28,12 @@ private[crdt] final class FlagImpl extends InternalCrdt with Flag {
   override def state: CrdtState.State = CrdtState.State.Flag(FlagState(value))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Flag(FlagDelta(value)) =>
+    case CrdtDelta.Delta.Flag(FlagDelta(value, _)) =>
       this.value |= value
   }
 
   override val applyState = {
-    case CrdtState.State.Flag(FlagState(value)) =>
+    case CrdtState.State.Flag(FlagState(value, _)) =>
       this.value = value
   }
 

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/GCounterImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/GCounterImpl.scala
@@ -31,12 +31,12 @@ private[crdt] final class GCounterImpl extends InternalCrdt with GCounter {
   override def state: CrdtState.State = CrdtState.State.Gcounter(GCounterState(value))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Gcounter(GCounterDelta(increment)) =>
+    case CrdtDelta.Delta.Gcounter(GCounterDelta(increment, _)) =>
       value += increment
   }
 
   override val applyState = {
-    case CrdtState.State.Gcounter(GCounterState(value)) =>
+    case CrdtState.State.Gcounter(GCounterState(value, _)) =>
       this.value = value
   }
 

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/GSetImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/GSetImpl.scala
@@ -48,12 +48,12 @@ private[crdt] final class GSetImpl[T](anySupport: AnySupport)
   override def state: CrdtState.State = CrdtState.State.Gset(GSetState(value.asScala.toSeq.map(anySupport.encodeScala)))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Gset(GSetDelta(added)) =>
+    case CrdtDelta.Delta.Gset(GSetDelta(added, _)) =>
       value.addAll(added.map(e => anySupport.decode(e).asInstanceOf[T]).asJava)
   }
 
   override val applyState = {
-    case CrdtState.State.Gset(GSetState(value)) =>
+    case CrdtState.State.Gset(GSetState(value, _)) =>
       this.value.clear()
       this.value.addAll(value.map(e => anySupport.decode(e).asInstanceOf[T]).asJava)
   }

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/LWWRegisterImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/LWWRegisterImpl.scala
@@ -45,12 +45,12 @@ private[crdt] final class LWWRegisterImpl[T](anySupport: AnySupport) extends Int
     )
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Lwwregister(LWWRegisterDelta(Some(any), _, _)) =>
+    case CrdtDelta.Delta.Lwwregister(LWWRegisterDelta(Some(any), _, _, _)) =>
       this.value = anySupport.decode(any).asInstanceOf[T]
   }
 
   override val applyState = {
-    case CrdtState.State.Lwwregister(LWWRegisterState(Some(any), _, _)) =>
+    case CrdtState.State.Lwwregister(LWWRegisterState(Some(any), _, _, _)) =>
       this.value = anySupport.decode(any).asInstanceOf[T]
   }
 

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/ORMapImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/ORMapImpl.scala
@@ -173,13 +173,13 @@ private[crdt] final class ORMapImpl[K, V <: InternalCrdt](anySupport: AnySupport
     )
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Ormap(ORMapDelta(cleared, removed, updated, added)) =>
+    case CrdtDelta.Delta.Ormap(ORMapDelta(cleared, removed, updated, added, _)) =>
       if (cleared) {
         value.clear()
       }
       removed.foreach(key => value.remove(anySupport.decode(key)))
       updated.foreach {
-        case ORMapEntryDelta(Some(key), Some(delta)) =>
+        case ORMapEntryDelta(Some(key), Some(delta), _) =>
           val crdt = value.get(anySupport.decode(key))
           if (crdt == null) {
             ORMapImpl.log.warn(s"ORMap entry to update with key $key not found in map")
@@ -188,17 +188,17 @@ private[crdt] final class ORMapImpl[K, V <: InternalCrdt](anySupport: AnySupport
           }
       }
       added.foreach {
-        case ORMapEntry(Some(key), Some(state)) =>
+        case ORMapEntry(Some(key), Some(state), _) =>
           value.put(anySupport.decode(key).asInstanceOf[K],
                     CrdtStateTransformer.create(state, anySupport).asInstanceOf[V])
       }
   }
 
   override val applyState = {
-    case CrdtState.State.Ormap(ORMapState(values)) =>
+    case CrdtState.State.Ormap(ORMapState(values, _)) =>
       value.clear()
       values.foreach {
-        case ORMapEntry(Some(key), Some(state)) =>
+        case ORMapEntry(Some(key), Some(state), _) =>
           value.put(anySupport.decode(key).asInstanceOf[K],
                     CrdtStateTransformer.create(state, anySupport).asInstanceOf[V])
       }

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/ORSetImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/ORSetImpl.scala
@@ -101,7 +101,7 @@ private[crdt] class ORSetImpl[T](anySupport: AnySupport) extends util.AbstractSe
     CrdtState.State.Orset(ORSetState(value.asScala.toSeq.map(anySupport.encodeScala)))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Orset(ORSetDelta(cleared, removed, added)) =>
+    case CrdtDelta.Delta.Orset(ORSetDelta(cleared, removed, added, _)) =>
       if (cleared) {
         value.clear()
       }
@@ -110,7 +110,7 @@ private[crdt] class ORSetImpl[T](anySupport: AnySupport) extends util.AbstractSe
   }
 
   override val applyState = {
-    case CrdtState.State.Orset(ORSetState(value)) =>
+    case CrdtState.State.Orset(ORSetState(value, _)) =>
       this.value.clear()
       this.value.addAll(value.map(e => anySupport.decode(e).asInstanceOf[T]).asJava)
   }

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/PNCounterImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/PNCounterImpl.scala
@@ -30,12 +30,12 @@ private[crdt] final class PNCounterImpl extends InternalCrdt with PNCounter {
   override def state: CrdtState.State = CrdtState.State.Pncounter(PNCounterState(value))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Pncounter(PNCounterDelta(increment)) =>
+    case CrdtDelta.Delta.Pncounter(PNCounterDelta(increment, _)) =>
       value += increment
   }
 
   override val applyState = {
-    case CrdtState.State.Pncounter(PNCounterState(value)) =>
+    case CrdtState.State.Pncounter(PNCounterState(value, _)) =>
       this.value = value
   }
 

diff --git a/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/VoteImpl.scala b/java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/VoteImpl.scala
@@ -43,13 +43,13 @@ private[crdt] final class VoteImpl extends InternalCrdt with Vote {
   override def state: CrdtState.State = CrdtState.State.Vote(VoteState(votesFor, voters, selfVote))
 
   override val applyDelta = {
-    case CrdtDelta.Delta.Vote(VoteDelta(_, votesFor, totalVoters)) =>
+    case CrdtDelta.Delta.Vote(VoteDelta(_, votesFor, totalVoters, _)) =>
       this.voters = totalVoters
       this.votesFor = votesFor
   }
 
   override val applyState = {
-    case CrdtState.State.Vote(VoteState(votesFor, voters, selfVote)) =>
+    case CrdtState.State.Vote(VoteState(votesFor, voters, selfVote, _)) =>
       this.voters = voters
       this.votesFor = votesFor
       this.selfVote = selfVote

diff --git a/...support/src/main/scala/io/cloudstate/javasupport/impl/eventsourced/EventSourcedImpl.scala b/...support/src/main/scala/io/cloudstate/javasupport/impl/eventsourced/EventSourcedImpl.scala
@@ -96,7 +96,7 @@ final class EventSourcedImpl(_system: ActorSystem,
   ): akka.stream.scaladsl.Source[EventSourcedStreamOut, akka.NotUsed] =
     in.prefixAndTail(1)
       .flatMapConcat {
-        case (Seq(EventSourcedStreamIn(InInit(init))), source) =>
+        case (Seq(EventSourcedStreamIn(InInit(init), _)), source) =>
           source.via(runEntity(init))
         case _ =>
           // todo better error

diff --git a/project/plugins.sbt b/project/plugins.sbt
@@ -1,7 +1,8 @@
 addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.17")
 addSbtPlugin("com.dwijnand" % "sbt-dynver" % "3.3.0")
 
-addSbtPlugin("com.lightbend.akka.grpc" % "sbt-akka-grpc" % "0.7.1")
+resolvers += Resolver.bintrayRepo("akka", "maven")
+addSbtPlugin("com.lightbend.akka.grpc" % "sbt-akka-grpc" % "0.8.4+25-52f006d6")
 addSbtPlugin("com.lightbend.sbt" % "sbt-javaagent" % "0.1.4")
 
 addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9")

diff --git a/protocols/proxy/grpc/reflection/v1alpha/reflection.proto b/protocols/proxy/grpc/reflection/v1alpha/reflection.proto
diff --git a/.../protos/cloudstate/proxy/autoscaler.proto → ...n/proto/cloudstate/proxy/autoscaler.proto b/.../protos/cloudstate/proxy/autoscaler.proto → ...n/proto/cloudstate/proxy/autoscaler.proto
diff --git a/...tos/cloudstate/proxy/crdt_protobufs.proto → ...oto/cloudstate/proxy/crdt_protobufs.proto b/...tos/cloudstate/proxy/crdt_protobufs.proto → ...oto/cloudstate/proxy/crdt_protobufs.proto
diff --git a/...rotos/cloudstate/proxy/statemanager.proto → ...proto/cloudstate/proxy/statemanager.proto b/...rotos/cloudstate/proxy/statemanager.proto → ...proto/cloudstate/proxy/statemanager.proto