Merge pull request #3284 from jclyne/jclyne/client-streaming-enhancements

Client Streaming Enhancements

Author: oldergod

gen-tests.gradle.kts

@@ -564,6 +564,7 @@ val generateGrpcTests by tasks.creating(JavaExec::class) {
   args = listOf(
           "--proto_path=wire-grpc-tests/src/test/proto",
           "--kotlin_out=wire-grpc-tests/src/test/proto-grpc",
+          "--kotlin_explicit_streaming_calls",
           "routeguide/RouteGuideProto.proto"
   ) + GRPC_PROTOS
 }

wire-compiler/api/wire-compiler.api

@@ -8,7 +8,7 @@ public final class com/squareup/wire/DryRunFileSystem : okio/ForwardingFileSyste
 public final class com/squareup/wire/WireCompiler {
 	public static final field CODE_GENERATED_BY_WIRE Ljava/lang/String;
 	public static final field Companion Lcom/squareup/wire/WireCompiler$Companion;
-	public synthetic fun <init> (Lokio/FileSystem;Lcom/squareup/wire/WireLogger;Ljava/util/List;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/util/List;Ljava/util/List;Ljava/util/List;Ljava/util/Map;ZZZZZZZZIZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZLjava/lang/String;ZZZLjava/util/List;Ljava/util/Map;Ljava/util/List;IILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public synthetic fun <init> (Lokio/FileSystem;Lcom/squareup/wire/WireLogger;Ljava/util/List;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/util/List;Ljava/util/List;Ljava/util/List;Ljava/util/Map;ZZZZZZZZIZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZLjava/lang/String;ZZZZLjava/util/List;Ljava/util/Map;Ljava/util/List;IILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun compile ()V
 	public static final fun forArgs (Ljava/nio/file/FileSystem;Lcom/squareup/wire/WireLogger;[Ljava/lang/String;)Lcom/squareup/wire/WireCompiler;
 	public static final fun forArgs (Lokio/FileSystem;Lcom/squareup/wire/WireLogger;[Ljava/lang/String;)Lcom/squareup/wire/WireCompiler;
@@ -31,6 +31,7 @@ public final class com/squareup/wire/WireCompiler {
 	public final fun getKotlinBuildersOnly ()Z
 	public final fun getKotlinEscapeKeywords ()Z
 	public final fun getKotlinExclusive ()Z
+	public final fun getKotlinExplicitStreamingCalls ()Z
 	public final fun getKotlinNameSuffix ()Ljava/lang/String;
 	public final fun getKotlinOut ()Ljava/lang/String;
 	public final fun getKotlinRpcCallStyle ()Lcom/squareup/wire/kotlin/RpcCallStyle;
@@ -117,8 +118,8 @@ public final class com/squareup/wire/schema/JavaTarget : com/squareup/wire/schem
 }
 
 public final class com/squareup/wire/schema/KotlinTarget : com/squareup/wire/schema/Target {
-	public fun <init> (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZ)V
-	public synthetic fun <init> (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZZ)V
+	public synthetic fun <init> (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZZILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun component1 ()Ljava/util/List;
 	public final fun component10 ()Lcom/squareup/wire/kotlin/RpcRole;
 	public final fun component11 ()Z
@@ -135,8 +136,8 @@ public final class com/squareup/wire/schema/KotlinTarget : com/squareup/wire/sch
 	public final fun component7 ()Z
 	public final fun component8 ()Z
 	public final fun component9 ()Lcom/squareup/wire/kotlin/RpcCallStyle;
-	public final fun copy (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZ)Lcom/squareup/wire/schema/KotlinTarget;
-	public static synthetic fun copy$default (Lcom/squareup/wire/schema/KotlinTarget;Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZILjava/lang/Object;)Lcom/squareup/wire/schema/KotlinTarget;
+	public final fun copy (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZZ)Lcom/squareup/wire/schema/KotlinTarget;
+	public static synthetic fun copy$default (Lcom/squareup/wire/schema/KotlinTarget;Ljava/util/List;Ljava/util/List;ZLjava/lang/String;ZZZZLcom/squareup/wire/kotlin/RpcCallStyle;Lcom/squareup/wire/kotlin/RpcRole;ZILjava/lang/String;ZZLcom/squareup/wire/kotlin/EnumMode;ZZZILjava/lang/Object;)Lcom/squareup/wire/schema/KotlinTarget;
 	public fun copyTarget (Ljava/util/List;Ljava/util/List;ZLjava/lang/String;)Lcom/squareup/wire/schema/Target;
 	public fun equals (Ljava/lang/Object;)Z
 	public final fun getAndroid ()Z

wire-compiler/src/main/java/com/squareup/wire/WireCompiler.kt

@@ -142,6 +142,7 @@ class WireCompiler internal constructor(
   val kotlinBuildersOnly: Boolean,
   val kotlinEscapeKeywords: Boolean,
   val emitProtoReader32: Boolean,
+  val kotlinExplicitStreamingCalls: Boolean,
   val eventListenerFactoryClasses: List<String>,
   val customOptions: Map<String, String>,
   val opaqueTypes: List<String> = listOf(),
@@ -177,6 +178,7 @@ class WireCompiler internal constructor(
         buildersOnly = kotlinBuildersOnly,
         escapeKotlinKeywords = kotlinEscapeKeywords,
         emitProtoReader32 = emitProtoReader32,
+        explicitStreamingCalls = kotlinExplicitStreamingCalls,
       )
     }
     if (swiftOut != null) {
@@ -290,6 +292,7 @@ class WireCompiler internal constructor(
     private const val EMIT_PROTO_READER_32 = "--emit_proto_reader_32"
     private const val CUSTOM_OPTION_FLAG = "--custom_option="
     private const val OPAQUE_TYPES_FLAG = "--opaque_types="
+    private const val KOTLIN_EXPLICIT_STREAMING_CALLS = "--kotlin_explicit_streaming_calls"
 
     @Throws(IOException::class)
     @JvmStatic
@@ -353,6 +356,7 @@ class WireCompiler internal constructor(
       var kotlinBuildersOnly = false
       var kotlinEscapeKeywords = false
       var emitProtoReader32 = false
+      var kotlinExplicitStreamingCalls = false
       var dryRun = false
       val customOptions = mutableMapOf<String, String>()
       val opaqueTypes = mutableListOf<String>()
@@ -473,6 +477,7 @@ class WireCompiler internal constructor(
           arg == LOAD_EXHAUSTIVELY -> loadExhaustively = true
           arg == JAVA_INTEROP -> javaInterop = true
           arg == EMIT_PROTO_READER_32 -> emitProtoReader32 = true
+          arg == KOTLIN_EXPLICIT_STREAMING_CALLS -> kotlinExplicitStreamingCalls = true
           arg.startsWith("--") -> throw IllegalArgumentException("Unknown argument '$arg'.")
           else -> sourceFileNames.add(arg)
         }
@@ -527,6 +532,7 @@ class WireCompiler internal constructor(
         eventListenerFactoryClasses = eventListenerFactoryClasses,
         customOptions = customOptions,
         opaqueTypes = opaqueTypes,
+        kotlinExplicitStreamingCalls = kotlinExplicitStreamingCalls,
       )
     }
   }

wire-compiler/src/main/java/com/squareup/wire/schema/Target.kt

@@ -147,6 +147,13 @@ data class KotlinTarget(
    * If true, the generated classes will be mutable..
    */
   private val mutableTypes: Boolean = false,
+
+  /**
+   * If true, the generated gRPC client will use explicit classes for client, server,
+   * and bidirectional streaming calls.
+   */
+  private val explicitStreamingCalls: Boolean = false,
+
 ) : Target() {
   override fun newHandler(): SchemaHandler {
     return KotlinSchemaHandler(
@@ -165,6 +172,7 @@ data class KotlinTarget(
       enumMode = enumMode,
       emitProtoReader32 = emitProtoReader32,
       mutableTypes = mutableTypes,
+      explicitStreamingCalls = explicitStreamingCalls,
     )
   }
 

wire-gradle-plugin/api/wire-gradle-plugin.api

@@ -52,6 +52,7 @@ public class com/squareup/wire/gradle/KotlinOutput : com/squareup/wire/gradle/Wi
 	public final fun getEscapeKotlinKeywords ()Z
 	public final fun getExcludes ()Ljava/util/List;
 	public final fun getExclusive ()Z
+	public final fun getExplicitStreamingCalls ()Z
 	public final fun getGrpcServerCompatible ()Z
 	public final fun getIncludes ()Ljava/util/List;
 	public final fun getJavaInterop ()Z
@@ -70,6 +71,7 @@ public class com/squareup/wire/gradle/KotlinOutput : com/squareup/wire/gradle/Wi
 	public final fun setEscapeKotlinKeywords (Z)V
 	public final fun setExcludes (Ljava/util/List;)V
 	public final fun setExclusive (Z)V
+	public final fun setExplicitStreamingCalls (Z)V
 	public final fun setGrpcServerCompatible (Z)V
 	public final fun setIncludes (Ljava/util/List;)V
 	public final fun setJavaInterop (Z)V

wire-gradle-plugin/src/main/kotlin/com/squareup/wire/gradle/WireOutput.kt

@@ -159,6 +159,12 @@ open class KotlinOutput @Inject constructor() : WireOutput() {
    */
   var mutableTypes: Boolean = false
 
+  /**
+   * If true, the generated gRPC client will use explicit classes for client, server,
+   * and bidirectional streaming calls.
+   */
+  var explicitStreamingCalls: Boolean = false
+
   override fun toTarget(outputDirectory: String): KotlinTarget {
     if (grpcServerCompatible) {
       throw IllegalArgumentException(
@@ -203,6 +209,7 @@ open class KotlinOutput @Inject constructor() : WireOutput() {
       enumMode = enumMode,
       emitProtoReader32 = emitProtoReader32,
       mutableTypes = mutableTypes,
+      explicitStreamingCalls = explicitStreamingCalls,
     )
   }
 }

wire-grpc-client/api/wire-grpc-client.api

@@ -20,6 +20,8 @@ public abstract interface class com/squareup/wire/GrpcCall$Callback {
 
 public final class com/squareup/wire/GrpcCalls {
 	public static final fun grpcCall (Lkotlin/jvm/functions/Function1;)Lcom/squareup/wire/GrpcCall;
+	public static final fun grpcClientStreamingCall (Lkotlin/jvm/functions/Function2;)Lcom/squareup/wire/GrpcClientStreamingCall;
+	public static final fun grpcServerStreamingCall (Lkotlin/jvm/functions/Function3;)Lcom/squareup/wire/GrpcServerStreamingCall;
 	public static final fun grpcStreamingCall (Lkotlin/jvm/functions/Function3;)Lcom/squareup/wire/GrpcStreamingCall;
 }
 
@@ -28,6 +30,8 @@ public abstract class com/squareup/wire/GrpcClient {
 	public final fun create (Lkotlin/reflect/KClass;)Lcom/squareup/wire/Service;
 	public final fun newBuilder ()Lcom/squareup/wire/GrpcClient$Builder;
 	public abstract fun newCall (Lcom/squareup/wire/GrpcMethod;)Lcom/squareup/wire/GrpcCall;
+	public abstract fun newClientStreamingCall (Lcom/squareup/wire/GrpcMethod;)Lcom/squareup/wire/GrpcClientStreamingCall;
+	public abstract fun newServerStreamingCall (Lcom/squareup/wire/GrpcMethod;)Lcom/squareup/wire/GrpcServerStreamingCall;
 	public abstract fun newStreamingCall (Lcom/squareup/wire/GrpcMethod;)Lcom/squareup/wire/GrpcStreamingCall;
 }
 
@@ -41,6 +45,25 @@ public final class com/squareup/wire/GrpcClient$Builder {
 	public final fun minMessageToCompress (J)Lcom/squareup/wire/GrpcClient$Builder;
 }
 
+public abstract interface class com/squareup/wire/GrpcClientStreamingCall {
+	public abstract fun cancel ()V
+	public abstract fun clone ()Lcom/squareup/wire/GrpcClientStreamingCall;
+	public abstract fun executeBlocking ()Lkotlin/Pair;
+	public abstract fun executeIn (Lkotlinx/coroutines/CoroutineScope;)Lkotlin/Pair;
+	public abstract fun getMethod ()Lcom/squareup/wire/GrpcMethod;
+	public abstract fun getRequestMetadata ()Ljava/util/Map;
+	public abstract fun getResponseMetadata ()Ljava/util/Map;
+	public abstract fun getTimeout ()Lokio/Timeout;
+	public abstract fun isCanceled ()Z
+	public abstract fun isExecuted ()Z
+	public abstract fun setRequestMetadata (Ljava/util/Map;)V
+}
+
+public abstract interface class com/squareup/wire/GrpcDeferredResponse {
+	public abstract fun close ()V
+	public abstract fun get ()Ljava/lang/Object;
+}
+
 public final class com/squareup/wire/GrpcException : java/io/IOException {
 	public synthetic fun <init> (Lcom/squareup/wire/GrpcStatus;Ljava/lang/String;[B)V
 	public synthetic fun <init> (Lcom/squareup/wire/GrpcStatus;Ljava/lang/String;[BILkotlin/jvm/internal/DefaultConstructorMarker;)V
@@ -63,6 +86,20 @@ public final class com/squareup/wire/GrpcMethod {
 	public final fun getResponseAdapter ()Lcom/squareup/wire/ProtoAdapter;
 }
 
+public abstract interface class com/squareup/wire/GrpcServerStreamingCall {
+	public abstract fun cancel ()V
+	public abstract fun clone ()Lcom/squareup/wire/GrpcServerStreamingCall;
+	public abstract fun executeBlocking (Ljava/lang/Object;)Lcom/squareup/wire/MessageSource;
+	public abstract fun executeIn (Lkotlinx/coroutines/CoroutineScope;Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun getMethod ()Lcom/squareup/wire/GrpcMethod;
+	public abstract fun getRequestMetadata ()Ljava/util/Map;
+	public abstract fun getResponseMetadata ()Ljava/util/Map;
+	public abstract fun getTimeout ()Lokio/Timeout;
+	public abstract fun isCanceled ()Z
+	public abstract fun isExecuted ()Z
+	public abstract fun setRequestMetadata (Ljava/util/Map;)V
+}
+
 public final class com/squareup/wire/GrpcStatus : java/io/Serializable {
 	public static final field ABORTED Lcom/squareup/wire/GrpcStatus;
 	public static final field ALREADY_EXISTS Lcom/squareup/wire/GrpcStatus;
@@ -106,6 +143,13 @@ public abstract interface class com/squareup/wire/GrpcStreamingCall {
 	public abstract fun setRequestMetadata (Ljava/util/Map;)V
 }
 
+public final class com/squareup/wire/GrpcStreamingCallExtensionsKt {
+	public static final fun bidirectionalStream (Lcom/squareup/wire/GrpcStreamingCall;Lkotlinx/coroutines/CoroutineScope;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun bidirectionalStreamBlocking (Lcom/squareup/wire/GrpcStreamingCall;Lkotlin/jvm/functions/Function2;)Lcom/squareup/wire/MessageSource;
+	public static final fun clientStream (Lcom/squareup/wire/GrpcClientStreamingCall;Lkotlinx/coroutines/CoroutineScope;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun clientStreamBlocking (Lcom/squareup/wire/GrpcClientStreamingCall;Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
+}
+
 public abstract interface annotation class com/squareup/wire/WireGrpcExperimental : java/lang/annotation/Annotation {
 }
 

wire-grpc-client/src/commonMain/kotlin/com/squareup/wire/GrpcClient.kt

@@ -18,4 +18,6 @@ package com.squareup.wire
 expect abstract class GrpcClient() {
   abstract fun <S : Any, R : Any> newCall(method: GrpcMethod<S, R>): GrpcCall<S, R>
   abstract fun <S : Any, R : Any> newStreamingCall(method: GrpcMethod<S, R>): GrpcStreamingCall<S, R>
+  abstract fun <S : Any, R : Any> newClientStreamingCall(method: GrpcMethod<S, R>): GrpcClientStreamingCall<S, R>
+  abstract fun <S : Any, R : Any> newServerStreamingCall(method: GrpcMethod<S, R>): GrpcServerStreamingCall<S, R>
 }

wire-grpc-client/src/commonMain/kotlin/com/squareup/wire/GrpcClientStreamingCall.kt

@@ -0,0 +1,98 @@
+/*
+ * Copyright (C) 2025 Square, 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
+ *
+ *      https://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 com.squareup.wire
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.SendChannel
+import okio.IOException
+import okio.Timeout
+
+/**
+ * A single streaming call to a remote server.
+ *
+ *  * This class handles a client streaming call . The send channel or message sink accept zero or more
+ *    messages. The returned deferred repsonse will ensure that the corresponding send channel or message sink
+ *    is closed before returning the single response message.
+ *
+ * A gRPC call cannot be executed twice.
+ *
+ * gRPC calls can be [suspending][executeIn] or [blocking][executeBlocking]. Use whichever mechanism
+ * works at your call site: the bytes transmitted on the network are the same.
+ */
+interface GrpcClientStreamingCall<S : Any, R : Any> {
+  /** The method invoked by this call. */
+  val method: GrpcMethod<S, R>
+
+  /**
+   * Configures how long the call can take to complete before it is automatically canceled. The
+   * timeout applies to the full set of messages transmitted. For long-running streams you must
+   * configure a sufficiently long timeout.
+   */
+  val timeout: Timeout
+
+  /**
+   * A map containing request metadata. This is initially empty; it can be assigned to any other map
+   * of metadata before the call is executed. It is an error to set this value after the call is
+   * executed.
+   */
+  var requestMetadata: Map<String, String>
+
+  /**
+   * A map containing response metadata. This is null until the call has executed, at which point
+   * it will be non-null if the call completed successfully. It may also be non-null in failure
+   * cases if the failure was not a problem of connectivity. For example, if the gRPC call fails
+   * with an HTTP 503 error, response metadata will be present.
+   */
+  val responseMetadata: Map<String, String>?
+
+  /**
+   * Attempts to cancel the call. This function is safe to call concurrently with execution. When
+   * canceled, execution fails with an immediate [IOException] rather than waiting to complete
+   * normally.
+   */
+  fun cancel()
+
+  /** True if [cancel] was called. */
+  fun isCanceled(): Boolean
+
+  /**
+   * Enqueues this call for execution and returns a channel to send the call's messages and a deferred
+   * response that will ensure the sender is closed and return the single response.
+   * This uses the [Dispatchers.IO] to transmit outbound messages.
+   */
+  fun executeIn(scope: CoroutineScope): Pair<SendChannel<S>, Deferred<R>>
+
+  /**
+   * Enqueues this call for execution and returns a stream to send call's messages and a deferred
+   * response that will ensure the sender is closed and return the single response.
+   * Reads and writes on the returned streams are blocking.
+   */
+  fun executeBlocking(): Pair<MessageSink<S>, GrpcDeferredResponse<R>>
+
+  /**
+   * Returns true if [executeIn] or [executeBlocking] was called. It is an error to execute a call
+   * more than once.
+   */
+  fun isExecuted(): Boolean
+
+  /**
+   * Create a new, identical gRPC call to this one which can be enqueued or executed even if this
+   * call has already been.
+   */
+  fun clone(): GrpcClientStreamingCall<S, R>
+}

wire-grpc-client/src/commonMain/kotlin/com/squareup/wire/GrpcDeferredResponse.kt

@@ -0,0 +1,23 @@
+/*
+ * Copyright (C) 2025 Square, 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
+ *
+ *      https://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 com.squareup.wire
+
+interface GrpcDeferredResponse<T : Any> {
+
+  fun get(): T
+
+  fun close()
+}