Provides grpc configuration DSL and GrpcServer algebras (#13)

* Provides grpc configuration DSL and GrpcServer algebras

Server Demo has been also adapated to the new introduced changes, where the rpc core provides helpers in order to bootstrap the server in the runtime monad, according to the server configuration.

Author: juanpedromoreno

README.md

@@ -96,7 +96,7 @@ You could generate a reverse proxy and writing an endpoint as it's described [he
 To run the gateway:
 
 ```bash
-go run demo/gateway/server/entry.go
+go run demo/http/gateway/server/entry.go
 ```
 
 Then, you could use `curl` or similar to fetch the user over `HTTP`:

build.sbt

@@ -16,7 +16,10 @@ lazy val rpc = project
   .settings(moduleName := "freestyle-rpc")
   .settings(
     Seq(
-      libraryDependencies ++= commonDeps ++ freestyleCoreDeps()
+      libraryDependencies ++= commonDeps ++ freestyleCoreDeps() ++
+        Seq(
+          "io.grpc" % "grpc-all" % "1.4.0"
+        )
     ): _*
   )
 

demo/greeting/src/main/scala/greeting/GreetingClientApp.scala

@@ -22,7 +22,8 @@ object GreetingClientApp {
   def main(args: Array[String]): Unit = {
 
     val request = MessageRequest("Freestyle")
-    val client  = new GreetingClient(host, port)
+    val client  = new GreetingClient(host, portNode1)
+    // val client  = new GreetingClient(host, portNode2)
 
     // http://www.grpc.io/docs/guides/concepts.html
 

demo/greeting/src/main/scala/greeting/GreetingServerApp.scala

@@ -17,18 +17,16 @@
 package freestyle.rpc.demo
 package greeting
 
-import scala.concurrent.ExecutionContext
+import cats.implicits._
+import freestyle.rpc.server._
+import freestyle.rpc.server.implicits._
+import runtime.implicits._
 
-object GreetingServerApp {
-
-  def main(args: Array[String]): Unit = {
+import scala.concurrent.duration.Duration
+import scala.concurrent.Await
 
-    val serverServiceDefinition =
-      GreeterGrpc.bindService(new GreetingService, ExecutionContext.global)
-    val server = new GrpcServer(serverServiceDefinition)
-
-    server.start()
-    server.blockUntilShutdown()
-  }
+object GreetingServerApp {
 
+  def main(args: Array[String]): Unit =
+    Await.result(server[GrpcServer.Op].bootstrapFuture, Duration.Inf)
 }

demo/greeting/src/main/scala/GrpcServer.scala renamed to demo/greeting/src/main/scala/greeting/runtime/implicits.scala

@@ -15,30 +15,27 @@
  */
 
 package freestyle.rpc.demo
-package greeting
+package greeting.runtime
 
-import io.grpc.{Server, ServerBuilder, ServerServiceDefinition}
+import freestyle._
+import freestyle.implicits._
+import freestyle.rpc.demo.greeting._
+import freestyle.rpc.server._
+import freestyle.rpc.server.implicits._
+import freestyle.rpc.server.handlers._
 
-class GrpcServer(serverServiceDefinition: ServerServiceDefinition) {
+import scala.concurrent.{ExecutionContext, Future}
 
-  var server: Option[Server] = None
+object implicits {
 
-  def start(): Unit = {
-    server = Option(
-      ServerBuilder
-        .forPort(port)
-        .addService(serverServiceDefinition)
-        .build
-        .start
-    )
+  implicit val config: Config       = Config(portNode1)
+  implicit val ec: ExecutionContext = ExecutionContext.Implicits.global
 
-    Runtime.getRuntime.addShutdownHook(new Thread() {
-      override def run(): Unit = stopServer()
-    })
-  }
+  implicit val grpcConfigs: List[GrpcConfig] = List(
+    AddService(GreeterGrpc.bindService(new GreetingService, ec))
+  )
 
-  def stopServer(): Unit = server.foreach(_.shutdown())
+  implicit val grpcServerHandler =
+    new GrpcServerHandler[Future] andThen new GrpcConfigInterpreter[Future]
 
-  def blockUntilShutdown(): Unit =
-    server.foreach(_.awaitTermination())
 }

demo/greeting/src/main/scala/package.scala

@@ -18,7 +18,8 @@ package freestyle.rpc
 
 package object demo {
 
-  val host = "localhost"
-  val port = 50051
+  val host      = "localhost"
+  val portNode1 = 50051
+  val portNode2 = 50052
 
 }

demo/http/src/main/scala/user/UserClientApp.scala

@@ -24,7 +24,7 @@ object UserClientApp {
 
   def main(args: Array[String]): Unit = {
     val request = UserPassword("frees", "password")
-    val client  = new UserClient(host, port)
+    val client  = new UserClient(host, portNode1)
 
     val response = client.login(request)
 

demo/http/src/main/scala/user/UserServerApp.scala

@@ -17,20 +17,16 @@
 package freestyle.rpc.demo
 package user
 
-import freestyle.rpc.demo.greeting.GrpcServer
+import cats.implicits._
+import freestyle.rpc.server._
+import freestyle.rpc.server.implicits._
+import runtime.implicits._
 
-import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.Duration
+import scala.concurrent.Await
 
 object UserServerApp {
 
-  def main(args: Array[String]): Unit = {
-
-    val serverServiceDefinition =
-      UserServiceGrpc.bindService(new UserService, ExecutionContext.global)
-    val server = new GrpcServer(serverServiceDefinition)
-
-    server.start()
-    server.blockUntilShutdown()
-  }
-
+  def main(args: Array[String]): Unit =
+    Await.result(server[GrpcServer.Op].bootstrapFuture, Duration.Inf)
 }

demo/http/src/main/scala/user/runtime/implicits.scala

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2017 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * 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 freestyle.rpc.demo
+package user.runtime
+
+import freestyle._
+import freestyle.implicits._
+import freestyle.rpc.demo.user._
+import freestyle.rpc.server._
+import freestyle.rpc.server.implicits._
+import freestyle.rpc.server.handlers._
+
+import scala.concurrent.{ExecutionContext, Future}
+
+object implicits {
+
+  implicit val config: Config       = Config(portNode1)
+  implicit val ec: ExecutionContext = ExecutionContext.Implicits.global
+
+  implicit val grpcConfigs: List[GrpcConfig] = List(
+    AddService(UserServiceGrpc.bindService(new UserService, ExecutionContext.global))
+  )
+
+  implicit val grpcServerHandler =
+    new GrpcServerHandler[Future] andThen new GrpcConfigInterpreter[Future]
+
+}

rpc/src/main/scala/server/GrpcConfig.scala

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2017 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * 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 freestyle.rpc
+package server
+
+import java.io.File
+import java.util.concurrent.Executor
+
+import io.grpc._
+
+case class Config(port: Int)
+
+sealed trait GrpcConfig extends Product with Serializable
+
+case object DirectExecutor extends GrpcConfig
+
+case class SetExecutor(executor: Executor) extends GrpcConfig
+
+case class AddService(service: ServerServiceDefinition) extends GrpcConfig
+
+case class AddBindableService(bindableService: BindableService) extends GrpcConfig
+
+case class AddTransportFilter(filter: ServerTransportFilter) extends GrpcConfig
+
+case class AddStreamTracerFactory(factory: ServerStreamTracer.Factory) extends GrpcConfig
+
+case class SetFallbackHandlerRegistry(fallbackRegistry: HandlerRegistry) extends GrpcConfig
+
+case class UseTransportSecurity(certChain: File, privateKey: File) extends GrpcConfig
+
+case class SetDecompressorRegistry(registry: DecompressorRegistry) extends GrpcConfig
+
+case class SetCompressorRegistry(registry: CompressorRegistry) extends GrpcConfig

rpc/src/main/scala/server/GrpcServer.scala

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2017 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * 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 freestyle.rpc
+package server
+
+import freestyle._
+import io.grpc._
+
+import scala.concurrent.duration.TimeUnit
+
+@free
+trait GrpcServer {
+
+  def start(): FS[Server]
+
+  def getPort: FS[Int]
+
+  def getServices: FS[List[ServerServiceDefinition]]
+
+  def getImmutableServices: FS[List[ServerServiceDefinition]]
+
+  def getMutableServices: FS[List[ServerServiceDefinition]]
+
+  def shutdown(): FS[Server]
+
+  def shutdownNow(): FS[Server]
+
+  def isShutdown: FS[Boolean]
+
+  def isTerminated: FS[Boolean]
+
+  def awaitTerminationTimeout(timeout: Long, unit: TimeUnit): FS[Boolean]
+
+  def awaitTermination(): FS[Unit]
+
+}

rpc/src/main/scala/server/handlers/GrpcServerHandler.scala

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2017 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * 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 freestyle.rpc
+package server.handlers
+
+import cats.data.Kleisli
+import freestyle.rpc.server.{GrpcServer, GrpcServerOps}
+import freestyle.Capture
+import io.grpc.{Server, ServerServiceDefinition}
+
+import scala.collection.JavaConverters._
+import scala.concurrent.duration.TimeUnit
+
+class GrpcServerHandler[F[_]](implicit C: Capture[F])
+    extends GrpcServer.Handler[GrpcServerOps[F, ?]] {
+
+  def start: GrpcServerOps[F, Server] = {
+
+    Runtime.getRuntime.addShutdownHook(new Thread() {
+      override def run(): Unit = {
+        shutdown
+        (): Unit
+      }
+    })
+
+    captureWithServer(_.start())
+  }
+
+  def getPort: GrpcServerOps[F, Int] = captureWithServer(_.getPort)
+
+  def getServices: GrpcServerOps[F, List[ServerServiceDefinition]] =
+    captureWithServer(_.getServices.asScala.toList)
+
+  def getImmutableServices: GrpcServerOps[F, List[ServerServiceDefinition]] =
+    captureWithServer(_.getImmutableServices.asScala.toList)
+
+  def getMutableServices: GrpcServerOps[F, List[ServerServiceDefinition]] =
+    captureWithServer(_.getMutableServices.asScala.toList)
+
+  def shutdown: GrpcServerOps[F, Server] = captureWithServer(_.shutdown())
+
+  def shutdownNow: GrpcServerOps[F, Server] = captureWithServer(_.shutdownNow())
+
+  def isShutdown: GrpcServerOps[F, Boolean] = captureWithServer(_.isShutdown)
+
+  def isTerminated: GrpcServerOps[F, Boolean] = captureWithServer(_.isTerminated)
+
+  def awaitTerminationTimeout(timeout: Long, unit: TimeUnit): GrpcServerOps[F, Boolean] =
+    captureWithServer(_.awaitTermination(timeout, unit))
+
+  def awaitTermination: GrpcServerOps[F, Unit] = captureWithServer(_.awaitTermination())
+
+  private[this] def captureWithServer[A](f: Server => A): GrpcServerOps[F, A] =
+    Kleisli(s => C.capture(f(s)))
+
+}