gRPC extended Demos

CHANGELOG.md

@@ -0,0 +1 @@
+# Changelog

README.md

@@ -1,8 +1,15 @@
+
+[comment]: # (Start Badges)
+
+[![Build Status](https://travis-ci.org/frees-io/freestyle-rpc.svg?branch=master)](https://travis-ci.org/frees-io/freestyle-rpc) [![codecov.io](http://codecov.io/github/frees-io/freestyle-rpc/coverage.svg?branch=master)](http://codecov.io/github/frees-io/freestyle-rpc?branch=master) [![Maven Central](https://img.shields.io/badge/maven%20central-0.0.1-green.svg)](https://oss.sonatype.org/#nexus-search;gav~io.frees~freestyle*) [![Latest version](https://img.shields.io/badge/freestyle--rpc-0.0.1-green.svg)](https://index.scala-lang.org/frees-io/freestyle-rpc) [![License](https://img.shields.io/badge/license-Apache%202-blue.svg)](https://raw.githubusercontent.com/frees-io/freestyle-rpc/master/LICENSE) [![Join the chat at https://gitter.im/47deg/freestyle](https://badges.gitter.im/47deg/freestyle.svg)](https://gitter.im/47deg/freestyle?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge) [![GitHub Issues](https://img.shields.io/github/issues/frees-io/freestyle-rpc.svg)](https://github.com/frees-io/freestyle-rpc/issues) [![Scala.js](http://scala-js.org/assets/badges/scalajs-0.6.15.svg)](http://scala-js.org)
+
+[comment]: # (End Badges)
+
 # freestyle-rpc
 
 Simple RPC with Freestyle
 
-## Demo
+## Demo
 
 Run server:
 

demo/src/main/proto/greeting.proto

@@ -5,14 +5,15 @@ package freestyle.rpc.demo;
 service Greeter {
     rpc SayHello (MessageRequest) returns (MessageReply) {}
     rpc SayGoodbye (MessageRequest) returns (MessageReply) {}
+    rpc LotsOfReplies(MessageRequest) returns (stream MessageReply) {}
+    rpc LotsOfGreetings(stream MessageRequest) returns (MessageReply) {}
+    rpc BidiHello(stream MessageRequest) returns (stream MessageReply) {}
 }
 
-// The request message containing the user's name.
 message MessageRequest {
     string name = 1;
 }
 
-// The response message containing the greetings
 message MessageReply {
     string message = 1;
 }

demo/src/main/scala/GreetingClient.scala

@@ -0,0 +1,142 @@
+/*
+ * 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 greeting
+
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+
+import io.grpc.ManagedChannelBuilder
+import freestyle.rpc.demo.greeting.GreeterGrpc._
+import io.grpc.stub.StreamObserver
+
+import scala.collection.immutable
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Await, Future, Promise}
+import scala.concurrent.duration._
+import scala.util.Random
+
+class GreetingClient(host: String, port: Int) {
+
+  private[this] val channel =
+    ManagedChannelBuilder.forAddress(host, port).usePlaintext(true).build
+
+  private[this] val asyncHelloClient: GreeterStub = GreeterGrpc.stub(channel)
+
+  def unaryDemo(request: MessageRequest): Unit = {
+
+    val response = for {
+      hi  <- asyncHelloClient.sayHello(request)
+      bye <- asyncHelloClient.sayGoodbye(request)
+    } yield (hi.message, bye.message)
+
+    println("")
+    println(s"Received -> ${Await.result(response, Duration.Inf)}")
+    println("")
+  }
+
+  def serverStreamingDemo(request: MessageRequest): Future[Unit] = {
+    val lotOfRepliesStreamingPromise = Promise[Unit]()
+    val lotOfRepliesObserver = new StreamObserver[MessageReply] {
+
+      override def onError(t: Throwable): Unit =
+        println(s"[lotOfRepliesObserver] Streaming failure: ${t.getMessage}")
+
+      override def onCompleted(): Unit = {
+        println("[lotOfRepliesObserver] Lot of Replies streaming completed")
+        lotOfRepliesStreamingPromise.success((): Unit)
+      }
+
+      override def onNext(value: MessageReply): Unit =
+        println(s"[lotOfRepliesObserver] Received by streaming -> $value")
+    }
+
+    asyncHelloClient.lotsOfReplies(request, lotOfRepliesObserver)
+
+    Await.ready(lotOfRepliesStreamingPromise.future, Duration.Inf)
+  }
+
+  def clientStreamingDemo(): Boolean = {
+    val countDownLatch = new CountDownLatch(1)
+    val responseObserver = new StreamObserver[MessageReply] {
+
+      override def onError(t: Throwable): Unit = {
+        println(s"[responseObserver] Streaming failure: ${t.getMessage}")
+        countDownLatch.countDown()
+      }
+
+      override def onCompleted(): Unit = {
+        println("[responseObserver] Lot of greetings streaming completed")
+        countDownLatch.countDown()
+      }
+
+      override def onNext(value: MessageReply): Unit =
+        println(s"[responseObserver] Received by streaming -> $value")
+    }
+
+    val requestObserver = asyncHelloClient.lotsOfGreetings(responseObserver)
+
+    val randomRequestList = 1 to math.min(5, Random.nextInt(20))
+
+    try {
+      randomRequestList foreach (i => requestObserver.onNext(MessageRequest(s"I'm Freestyle $i")))
+    } catch {
+      case t: Throwable =>
+        countDownLatch.countDown()
+        requestObserver.onError(t)
+    }
+
+    requestObserver.onCompleted()
+    countDownLatch.await(1, TimeUnit.MINUTES)
+  }
+
+  def biStreamingDemo(): Boolean = {
+
+    val countDownLatch = new CountDownLatch(1)
+    val requestObserver = asyncHelloClient.bidiHello(new StreamObserver[MessageReply] {
+
+      override def onError(t: Throwable): Unit = {
+        println(s"Bi-Streaming failure: ${t.getMessage}")
+        countDownLatch.countDown()
+      }
+
+      override def onCompleted(): Unit = {
+        println("Finished Bi-streaming")
+        countDownLatch.countDown()
+      }
+
+      override def onNext(value: MessageReply): Unit =
+        println(s"Got $value from server")
+    })
+
+    val randomRequestList: immutable.Seq[MessageRequest] = (1 to math.min(5, Random.nextInt(20))) map (
+        i => MessageRequest(s"Message $i"))
+
+    try {
+      for (request <- randomRequestList) {
+        println(s"Sending message $request")
+        requestObserver.onNext(request)
+      }
+    } catch {
+      case e: RuntimeException =>
+        requestObserver.onError(e)
+        throw e
+    }
+
+    requestObserver.onCompleted()
+    countDownLatch.await(1, TimeUnit.MINUTES)
+  }
+}

demo/src/main/scala/GreetingClientApp.scala

@@ -17,30 +17,39 @@
 package freestyle.rpc.demo
 package greeting
 
-import io.grpc.ManagedChannelBuilder
-import freestyle.rpc.demo.greeting.GreeterGrpc._
-
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.Await
-import scala.concurrent.duration._
-
 object GreetingClientApp {
 
-  private val channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext(true).build
-
   def main(args: Array[String]): Unit = {
 
     val request = MessageRequest("Freestyle")
+    val client  = new GreetingClient(host, port)
+
+    // http://www.grpc.io/docs/guides/concepts.html
+
+    // Unary RPCs where the client sends a single request to the server and
+    // gets a single response back, just like a normal function call:
+    client.unaryDemo(request)
+
+    // Server streaming RPCs where the client sends a request to the server and
+    // gets a stream to read a sequence of messages back. The client reads from
+    // the returned stream until there are no more messages.
+
+    client.serverStreamingDemo(request)
+
+    // Client streaming RPCs where the client writes a sequence of messages and sends them
+    // to the server, again using a provided stream. Once the client has finished writing the messages,
+    // it waits for the server to read them and return its response.
+
+    client.clientStreamingDemo()
 
-    val asyncHelloClient: GreeterStub = GreeterGrpc.stub(channel)
+    // Bidirectional streaming RPCs where both sides send a sequence of messages using a read-write stream.
+    // The two streams operate independently, so clients and servers can read and write in whatever order
+    // they like: for example, the server could wait to receive all the client messages before writing its
+    // responses, or it could alternately read a message then write a message, or some other combination
+    // of reads and writes. The order of messages in each stream is preserved.
 
-    val response = for {
-      hi  <- asyncHelloClient.sayHello(request)
-      bye <- asyncHelloClient.sayGoodbye(request)
-    } yield (hi.message, bye.message)
+    client.biStreamingDemo()
 
-    println("")
-    println(s"Received -> ${Await.result(response, Duration.Inf)}")
-    println("")
+    (): Unit
   }
 }

demo/src/main/scala/GreetingService.scala

@@ -17,17 +17,95 @@
 package freestyle.rpc.demo
 package greeting
 
+import java.util.concurrent.{Executors, TimeUnit}
+import java.util.concurrent.atomic.AtomicInteger
+
+import io.grpc.stub.StreamObserver
+
 import scala.concurrent.Future
 
 class GreetingService extends GreeterGrpc.Greeter {
 
+  val numberOfReplies    = 10
+  val initialDelay: Long = 0l
+  val interval: Long     = 500l
+
+  // rpc SayHello (MessageRequest) returns (MessageReply) {}
   def sayHello(request: MessageRequest): Future[MessageReply] = {
     println(s"Hi message received from ${request.name}")
     Future.successful(MessageReply(s"Hello ${request.name} from HelloService!"))
   }
 
+  // rpc SayGoodbye (MessageRequest) returns (MessageReply) {}
   def sayGoodbye(request: MessageRequest): Future[MessageReply] = {
     println(s"Goodbye message received from ${request.name}")
     Future.successful(MessageReply(s"See you soon ${request.name}!"))
   }
+
+  // rpc LotsOfReplies(MessageRequest) returns (stream MessageReply) {}
+  def lotsOfReplies(
+      request: MessageRequest,
+      responseObserver: StreamObserver[MessageReply]): Unit = {
+    val scheduler = Executors.newSingleThreadScheduledExecutor()
+    val tick = new Runnable {
+      val counter = new AtomicInteger(10)
+      def run(): Unit = {
+        val n: Int = counter.getAndDecrement()
+
+        if (n >= 0) {
+          responseObserver.onNext(
+            MessageReply(
+              s"[$n] I'm sorry to be a bore, but I wanted to say hi again ${request.name}!"))
+        } else {
+          scheduler.shutdown()
+          responseObserver.onCompleted()
+        }
+      }
+    }
+
+    scheduler.scheduleAtFixedRate(tick, initialDelay, interval, TimeUnit.MILLISECONDS)
+    (): Unit
+  }
+
+  // rpc LotsOfGreetings(stream MessageRequest) returns (MessageReply) {}
+  override def lotsOfGreetings(
+      responseObserver: StreamObserver[MessageReply]): StreamObserver[MessageRequest] =
+    new StreamObserver[MessageRequest] {
+      val loggerInfo = "lotsOfGreetings"
+      val counter    = new AtomicInteger(0)
+
+      override def onError(t: Throwable): Unit =
+        println(s"[$loggerInfo] Streaming failure: ${t.getMessage}")
+
+      override def onCompleted(): Unit = {
+        println(s"[$loggerInfo] Streaming completed.")
+
+        responseObserver.onNext(MessageReply(s"$loggerInfo - It's done ;)"))
+        responseObserver.onCompleted()
+      }
+
+      override def onNext(value: MessageRequest): Unit =
+        println(s"[$loggerInfo] This is your message ${counter.incrementAndGet()}, ${value.name}")
+    }
+
+  // rpc BidiHello(stream MessageRequest) returns (stream MessageReply) {}
+  override def bidiHello(
+      responseObserver: StreamObserver[MessageReply]): StreamObserver[MessageRequest] =
+    new StreamObserver[MessageRequest] {
+      val loggerInfo = "bidiHello"
+      val counter    = new AtomicInteger(0)
+
+      override def onError(t: Throwable): Unit =
+        println(s"[$loggerInfo] Streaming failure: ${t.getMessage}")
+
+      override def onCompleted(): Unit = {
+        println(s"[$loggerInfo] Streaming completed.")
+
+        responseObserver.onNext(MessageReply(s"$loggerInfo - It's done ;)"))
+        responseObserver.onCompleted()
+      }
+
+      override def onNext(value: MessageRequest): Unit =
+        println(s"[$loggerInfo] This is your message ${counter.incrementAndGet()}, ${value.name}")
+    }
 }

version.sbt

@@ -1 +1 @@
-version in ThisBuild := "0.0.1-SNAPSHOT"
+version in ThisBuild := "0.0.1-SNAPSHOT"