adds streaming and updates from google3

Author: bshaffer

BUILD

@@ -22,7 +22,13 @@ package_group(
 
 package_group(
     name = "approved_clients",
-    packages = [],
+    packages = [
+        "//java/com/google/apps/framework/server/kotlin/...",
+        "//java/com/google/apps/tiktok/rpc/...",
+        "//java/com/google/corp/accessibility/tracker/...",
+        "//javatests/com/google/apps/tiktok/rpc/...",
+        "//javatests/com/google/corp/accessibility/tracker/...",
+    ],
 )
 
 alias(

example/src/main/kotlin/io/grpc/examples/routeguide/BUILD

@@ -17,12 +17,12 @@ kt_jvm_library(
         "//google/protobuf:duration_java_proto",
         "//java/com/google/common/base",
         "//java/com/google/common/io",
-        "//java/com/google/protobuf/util:javatime",
         "//java/com/google/protobuf/util:json",
+        "//java/com/google/protobuf/util:time",
         "//third_party/java/grpc:core",
         "//third_party/java/grpc:netty",
         "//third_party/kotlin/grpc_kotlin/example/src/main/proto:route_guide_grpc_kotlin",
-        "//third_party/kotlin/grpc_kotlin/example/src/main/proto:route_guide_proto_kotlin",
+        "//third_party/kotlin/grpc_kotlin/example/src/main/proto:route_guide_proto_java",
         "//third_party/kotlin/grpc_kotlin/example/src/main/resources/io/grpc/examples/routeguide:resources",
         "//third_party/kotlin/kotlinx_coroutines",
     ],

example/src/main/kotlin/io/grpc/examples/routeguide/RouteGuideClient.kt

@@ -1,29 +1,30 @@
 package io.grpc.examples.routeguide
 
-import com.google.protobuf.util.JavaTimeConversions
 import io.grpc.ManagedChannel
 import io.grpc.ManagedChannelBuilder
 import io.grpc.examples.routeguide.RouteGuideGrpcKt.RouteGuideCoroutineStub
-import java.io.Closeable
-import java.util.concurrent.Executors
-import java.util.concurrent.TimeUnit
-import kotlin.random.Random
-import kotlin.random.nextLong
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.asExecutor
-import kotlinx.coroutines.async
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import java.io.Closeable
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import kotlin.random.Random
+import kotlin.random.nextLong
 
 class RouteGuideClient private constructor(
   val channel: ManagedChannel,
   val stub: RouteGuideCoroutineStub,
   val printer: Printer
 ) : Closeable {
-  val random = Random(314159)
+  private val random = Random(314159)
 
   constructor(
     channelBuilder: ManagedChannelBuilder<*>,
@@ -55,13 +56,18 @@ class RouteGuideClient private constructor(
         }
       }
     }
+
+    private fun point(lat: Int, lon: Int): Point =
+      Point.newBuilder()
+        .setLatitude(lat)
+        .setLongitude(lon)
+        .build()
   }
 
   override fun close() {
     channel.shutdown().awaitTermination(5, TimeUnit.SECONDS)
   }
 
-
   interface Printer {
     companion object {
       val stdout = object : Printer {
@@ -76,10 +82,10 @@ class RouteGuideClient private constructor(
   fun getFeature(latitude: Int, longitude: Int) = runBlocking {
     printer.println("*** GetFeature: lat=$latitude lon=$longitude")
 
-    val request = point {
-      this.latitude = latitude
-      this.longitude = longitude
-    }
+    val request = Point.newBuilder()
+      .setLatitude(latitude)
+      .setLongitude(longitude)
+      .build()
     val feature = stub.getFeature(request)
 
     if (feature.exists()) {
@@ -92,90 +98,67 @@ class RouteGuideClient private constructor(
   fun listFeatures(lowLat: Int, lowLon: Int, hiLat: Int, hiLon: Int) = runBlocking {
     printer.println("*** ListFeatures: lowLat=$lowLat lowLon=$lowLon hiLat=$hiLat liLon=$hiLon")
 
-    val request = rectangle {
-      lo = point {
-        latitude = lowLat
-        longitude = lowLon
-      }
-      hi = point {
-        latitude = lowLat
-        longitude = lowLon
-      }
-    }
+    val request = Rectangle.newBuilder()
+      .setLo(point(lowLat, lowLon))
+      .setHi(point(hiLat, hiLon))
+      .build()
     var i = 1
-    for (feature in stub.listFeatures(request)) {
-      printer.println("Result #$i: $feature")
-      i++
+    stub.listFeatures(request).collect { feature ->
+      printer.println("Result #${i++}: $feature")
     }
   }
 
   fun recordRoute(features: List<Feature>, numPoints: Int) = runBlocking {
     printer.println("*** RecordRoute")
-    val requests = Channel<Point>()
+    val requests = flow {
+      for (i in 1..numPoints) {
+        val feature = features.random(random)
+        println("Visiting point ${feature.location.toStr()}")
+        emit(feature.location)
+        delay(timeMillis = random.nextLong(500L..1500L))
+      }
+    }
     val finish = launch {
       val summary = stub.recordRoute(requests)
       printer.println("Finished trip with ${summary.pointCount} points.")
       printer.println("Passed ${summary.featureCount} features.")
       printer.println("Travelled ${summary.distance} meters.")
-      val duration = JavaTimeConversions.toJavaDuration(summary.elapsedTime).seconds
+      val duration = summary.elapsedTime.seconds
       printer.println("It took $duration seconds.")
     }
-    for (i in 1..numPoints) {
-      val feature = features.random(random)
-      println("Visiting point ${feature.location.toStr()}")
-      requests.send(feature.location)
-      delay(timeMillis = random.nextLong(500L..1500L))
-    }
-    requests.close()
     finish.join()
   }
 
   fun routeChat() = runBlocking {
     printer.println("*** RouteChat")
-    val requests = Channel<RouteNote>()
-    val rpc = launch {
-      val responses = stub.routeChat(requests)
-      for (note in responses) {
-        printer.println("Got message \"${note.message}\" at ${note.location.toStr()}")
-      }
-      println("Finished RouteChat")
-    }
     val requestList = listOf(
-      routeNote {
+      RouteNote.newBuilder().apply {
         message = "First message"
-        location = point {
-          latitude = 0
-          longitude = 0
-        }
-      },
-      routeNote {
+        location = point(0, 0)
+      }.build(),
+      RouteNote.newBuilder().apply {
         message = "Second message"
-        location = point {
-          latitude = 0
-          longitude = 1
-        }
-      },
-      routeNote {
+        location = point(0, 0)
+      }.build(),
+      RouteNote.newBuilder().apply {
         message = "Third message"
-        location = point {
-          latitude = 1
-          longitude = 0
-        }
-      },
-      routeNote {
+        location = point(1, 0)
+      }.build(),
+      RouteNote.newBuilder().apply {
         message = "Fourth message"
-        location = point {
-          latitude = 1
-          longitude = 1
-        }
-      }
+        location = point(1, 1)
+      }.build()
     )
-
-    for (request in requestList) {
+    val requests = requestList.asFlow().onEach { request ->
       printer.println("Sending message \"${request.message}\" at ${request.location.toStr()}")
-      requests.send(request)
     }
-    requests.close()
+    val rpc = launch {
+      stub.routeChat(requests).collect { note ->
+        printer.println("Got message \"${note.message}\" at ${note.location.toStr()}")
+      }
+      println("Finished RouteChat")
+    }
+
     rpc.join()
   }
 }

example/src/main/kotlin/io/grpc/examples/routeguide/RouteGuideServer.kt

@@ -3,13 +3,17 @@ package io.grpc.examples.routeguide
 import com.google.common.base.Stopwatch
 import com.google.common.base.Ticker
 import com.google.common.io.ByteSource
-import com.google.protobuf.util.JavaTimeConversions
+import com.google.protobuf.util.Durations
 import io.grpc.Server
 import io.grpc.ServerBuilder
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flow
 import java.util.Collections
 import java.util.concurrent.ConcurrentHashMap
-import kotlinx.coroutines.channels.ReceiveChannel
-import kotlinx.coroutines.channels.SendChannel
+import java.util.concurrent.TimeUnit
 
 /**
  * Kotlin adaptation of RouteGuideServer from the Java gRPC example.
@@ -74,53 +78,51 @@ class RouteGuideServer private constructor(
 
     override suspend fun getFeature(request: Point): Feature {
       return features.find { it.location == request }
-        ?: feature { location = request } // No feature was found, return an unnamed feature.
+        ?: Feature.newBuilder().apply { location = request }.build()
+      // No feature was found, return an unnamed feature.
     }
 
-    override suspend fun listFeatures(request: Rectangle, responses: SendChannel<Feature>) {
-      for (feature in features) {
-        if (feature.exists() && feature.location in request) {
-          responses.send(feature)
-        }
-      }
+    override fun listFeatures(request: Rectangle): Flow<Feature> {
+      return features.asFlow().filter { it.exists() && it.location in request }
     }
 
-    override suspend fun recordRoute(requests: ReceiveChannel<Point>): RouteSummary {
+    override suspend fun recordRoute(requests: Flow<Point>): RouteSummary {
       var pointCount = 0
       var featureCount = 0
       var distance = 0
       var previous: Point? = null
       val stopwatch = Stopwatch.createStarted(ticker)
-      for (request in requests) {
+      requests.collect { request ->
         pointCount++
         if (getFeature(request).exists()) {
           featureCount++
         }
-        if (previous != null) {
-          distance += previous distanceTo request
+        val prev = previous
+        if (prev != null) {
+          distance += prev distanceTo request
         }
         previous = request
       }
-      return routeSummary {
+      return RouteSummary.newBuilder().apply {
         this.pointCount = pointCount
         this.featureCount = featureCount
         this.distance = distance
-        this.elapsedTime = JavaTimeConversions.toProtoDuration(stopwatch.elapsed())
-      }
+        this.elapsedTime = Durations.fromMicros(stopwatch.elapsed(TimeUnit.MICROSECONDS))
+      }.build()
     }
 
-    override suspend fun routeChat(
-      requests: ReceiveChannel<RouteNote>,
-      responses: SendChannel<RouteNote>
-    ) {
-      for (note in requests) {
-        val notes: MutableList<RouteNote> = routeNotes.computeIfAbsent(note.location) {
-          Collections.synchronizedList(mutableListOf<RouteNote>())
-        }
-        for (prevNote in notes.toTypedArray()) { // thread-safe snapshot
-          responses.send(prevNote)
+    override fun routeChat(requests: Flow<RouteNote>): Flow<RouteNote> {
+      return flow {
+        // could use transform, but it's currently experimental
+        requests.collect { note ->
+          val notes: MutableList<RouteNote> = routeNotes.computeIfAbsent(note.location) {
+            Collections.synchronizedList(mutableListOf<RouteNote>())
+          }
+          for (prevNote in notes.toTypedArray()) { // thread-safe snapshot
+            emit(prevNote)
+          }
+          notes += note
         }
-        notes += note
       }
     }
   }

example/src/main/proto/BUILD

@@ -1,4 +1,4 @@
-load("//tools/build_defs/kotlin:rules.bzl", "kt_jvm_grpc_library", "kt_jvm_proto_library")
+load("//tools/build_defs/kotlin:rules.bzl", "kt_jvm_grpc_library")
 
 package(
     default_visibility = ["//third_party/kotlin/grpc_kotlin:__subpackages__"],
@@ -13,8 +13,8 @@ proto_library(
     deps = ["//google/protobuf:duration"],
 )
 
-kt_jvm_proto_library(
-    name = "route_guide_proto_kotlin",
+java_proto_library(
+    name = "route_guide_proto_java",
     deps = [":route_guide_proto"],
 )