Non suspend flow (#299)

* KRPC-133 Don't require `suspend` modifier in functions returning `Flow`

* gold tests

* update compiler tests

* Added cancellation tests and GC tests, checked exceptions

* apiDump

* detekt

* uncomment subplugin options

* fix timeout

* review comments

Author: Mr3zee

compiler-plugin/compiler-plugin-backend/src/main/core/kotlinx/rpc/codegen/extension/RpcIrContext.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.rpc.codegen.extension
@@ -176,6 +176,10 @@ internal class RpcIrContext(
             rpcClient.namedFunction("call")
         }
 
+        val rpcClientCallServerStreaming by lazy {
+            rpcClient.namedFunction("callServerStreaming")
+        }
+
         val provideStubContext by lazy {
             rpcClient.namedFunction("provideStubContext")
         }

compiler-plugin/compiler-plugin-backend/src/main/core/kotlinx/rpc/codegen/extension/RpcStubGenerator.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.rpc.codegen.extension
@@ -530,7 +530,7 @@ internal class RpcStubGenerator(
             returnType = method.function.returnType
             modality = Modality.OPEN
 
-            isSuspend = true
+            isSuspend = method.function.isSuspend
         }.apply {
             val functionThisReceiver = vsApi {
                 stubClassThisReceiver.copyToVS(this@apply, origin = IrDeclarationOrigin.DEFINED)
@@ -550,6 +550,20 @@ internal class RpcStubGenerator(
             overriddenSymbols = listOf(method.function.symbol)
 
             body = irBuilder(symbol).irBlockBody {
+                if (method.function.isNonSuspendingWithFlowReturn()) {
+                    +irReturn(
+                        irRpcMethodClientCall(
+                            method = method,
+                            functionThisReceiver = functionThisReceiver,
+                            isMethodObject = isMethodObject,
+                            methodClass = methodClass,
+                            arguments = arguments,
+                        )
+                    )
+
+                    return@irBlockBody
+                }
+
                 +irReturn(
                     irCall(
                         callee = ctx.functions.scopedClientCall,
@@ -742,8 +756,14 @@ internal class RpcStubGenerator(
         methodClass: IrClass,
         arguments: List<IrValueParameter>,
     ): IrCall {
+        val callee = if (method.function.isNonSuspendingWithFlowReturn()) {
+            ctx.functions.rpcClientCallServerStreaming.symbol
+        } else {
+            ctx.functions.rpcClientCall.symbol
+        }
+
         val call = irCall(
-            callee = ctx.functions.rpcClientCall.symbol,
+            callee = callee,
             type = method.function.returnType,
             typeArgumentsCount = 1,
         ).apply {
@@ -1218,13 +1238,15 @@ internal class RpcStubGenerator(
      *         ),
      *         ...
      *     ),
+     *     isNonSuspendFunction = !function.isSuspend,
      * )
      *```
      *
      * Where:
      *  - `<callable-name>` - the name of the method (field)
      *  - `<callable-data-type>` - a method class for a method and `FieldDataObject` for fields
-     *  - `<callable-return-type>` - the return type for the method and the field type for a field
+     *  - `<callable-return-type>` - the return type for the method and the field type for a field.
+     *  For a non-suspending flow the return type is its element type
      *  - `<callable-invokator>` - an invokator, previously generated by [generateInvokators]
      *  - `<method-parameter-name-k>` - if a method, its k-th parameter name
      *  - `<method-parameter-type-k>` - if a method, its k-th parameter type
@@ -1253,7 +1275,16 @@ internal class RpcStubGenerator(
             putValueArgument(1, irRpcTypeCall(dataType))
 
             val returnType = when (callable) {
-                is ServiceDeclaration.Method -> callable.function.returnType
+                is ServiceDeclaration.Method -> when {
+                    callable.function.isNonSuspendingWithFlowReturn() -> {
+                        (callable.function.returnType as IrSimpleType).arguments.single().typeOrFail
+                    }
+
+                    else -> {
+                        callable.function.returnType
+                    }
+                }
+
                 is ServiceDeclaration.FlowField -> callable.property.getterOrFail.returnType
             }
 
@@ -1321,9 +1352,14 @@ internal class RpcStubGenerator(
             }
 
             putValueArgument(4, arrayOfCall)
+            putValueArgument(5, booleanConst(callable is ServiceDeclaration.Method && !callable.function.isSuspend))
         }
     }
 
+    private fun IrSimpleFunction.isNonSuspendingWithFlowReturn(): Boolean {
+        return returnType.classOrNull == ctx.flow && !isSuspend
+    }
+
     /**
      * Accessor function for the `callableMap` property
      * Defined in `RpcServiceDescriptor`
@@ -1525,7 +1561,7 @@ internal class RpcStubGenerator(
     }
 
     /**
-     * IR call of the `RpcType(KType, Array<Annotation>)` function
+     * IR call of the `RpcType(KType)` function
      */
     private fun irRpcTypeCall(type: IrType): IrConstructorCallImpl {
         return vsApi {
@@ -1644,6 +1680,13 @@ internal class RpcStubGenerator(
         value = value,
     )
 
+    private fun booleanConst(value: Boolean) = IrConstImpl.boolean(
+        startOffset = UNDEFINED_OFFSET,
+        endOffset = UNDEFINED_OFFSET,
+        type = ctx.irBuiltIns.booleanType,
+        value = value,
+    )
+
     private fun <T> vsApi(body: VersionSpecificApi.() -> T): T {
         return ctx.versionSpecificApi.body()
     }

compiler-plugin/compiler-plugin-k2/src/main/core/kotlinx/rpc/codegen/StrictMode.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.rpc.codegen
@@ -52,8 +52,8 @@ fun CompilerConfiguration.strictModeAggregator(): StrictModeAggregator {
         stateFlow = get(StrictModeConfigurationKeys.STATE_FLOW, StrictMode.WARNING),
         sharedFlow = get(StrictModeConfigurationKeys.SHARED_FLOW, StrictMode.WARNING),
         nestedFlow = get(StrictModeConfigurationKeys.NESTED_FLOW, StrictMode.WARNING),
-        streamScopedFunctions = get(StrictModeConfigurationKeys.STREAM_SCOPED_FUNCTIONS, StrictMode.NONE),
-        suspendingServerStreaming = get(StrictModeConfigurationKeys.SUSPENDING_SERVER_STREAMING, StrictMode.NONE),
+        streamScopedFunctions = get(StrictModeConfigurationKeys.STREAM_SCOPED_FUNCTIONS, StrictMode.WARNING),
+        suspendingServerStreaming = get(StrictModeConfigurationKeys.SUSPENDING_SERVER_STREAMING, StrictMode.WARNING),
         notTopLevelServerFlow = get(StrictModeConfigurationKeys.NOT_TOP_LEVEL_SERVER_FLOW, StrictMode.WARNING),
         fields = get(StrictModeConfigurationKeys.FIELDS, StrictMode.WARNING),
     )

core/api/core.api

@@ -32,9 +32,14 @@ public final class kotlinx/rpc/RpcCall {
 public abstract interface class kotlinx/rpc/RpcClient : kotlinx/coroutines/CoroutineScope {
 	public abstract fun call (Lkotlinx/rpc/RpcCall;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun callAsync (Lkotlinx/coroutines/CoroutineScope;Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/Deferred;
+	public abstract fun callServerStreaming (Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/flow/Flow;
 	public abstract fun provideStubContext (J)Lkotlin/coroutines/CoroutineContext;
 }
 
+public final class kotlinx/rpc/RpcClient$DefaultImpls {
+	public static fun callServerStreaming (Lkotlinx/rpc/RpcClient;Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/flow/Flow;
+}
+
 public abstract interface annotation class kotlinx/rpc/RpcEagerField : java/lang/annotation/Annotation {
 }
 
@@ -59,12 +64,13 @@ public abstract interface annotation class kotlinx/rpc/annotations/Rpc : java/la
 }
 
 public final class kotlinx/rpc/descriptor/RpcCallable {
-	public fun <init> (Ljava/lang/String;Lkotlinx/rpc/descriptor/RpcType;Lkotlinx/rpc/descriptor/RpcType;Lkotlinx/rpc/descriptor/RpcInvokator;[Lkotlinx/rpc/descriptor/RpcParameter;)V
+	public fun <init> (Ljava/lang/String;Lkotlinx/rpc/descriptor/RpcType;Lkotlinx/rpc/descriptor/RpcType;Lkotlinx/rpc/descriptor/RpcInvokator;[Lkotlinx/rpc/descriptor/RpcParameter;Z)V
 	public final fun getDataType ()Lkotlinx/rpc/descriptor/RpcType;
 	public final fun getInvokator ()Lkotlinx/rpc/descriptor/RpcInvokator;
 	public final fun getName ()Ljava/lang/String;
 	public final fun getParameters ()[Lkotlinx/rpc/descriptor/RpcParameter;
 	public final fun getReturnType ()Lkotlinx/rpc/descriptor/RpcType;
+	public final fun isNonSuspendFunction ()Z
 }
 
 public abstract interface class kotlinx/rpc/descriptor/RpcInvokator {

core/src/commonMain/kotlin/kotlinx/rpc/RpcClient.kt

@@ -6,6 +6,7 @@ package kotlinx.rpc
 
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.flow.Flow
 import kotlin.coroutines.CoroutineContext
 
 @Deprecated("Use RpcClient instead", ReplaceWith("RpcClient"), level = DeprecationLevel.ERROR)
@@ -36,8 +37,25 @@ public interface RpcClient : CoroutineScope {
      * that is needed to route it properly to the server.
      * @return actual result of the call, for example, data from the server
      */
+    @Deprecated(
+        "This method was primarily used for fields in RPC services, which are now deprecated. " +
+                "See https://kotlin.github.io/kotlinx-rpc/strict-mode.html fields guide for more information"
+    )
     public fun <T> callAsync(serviceScope: CoroutineScope, call: RpcCall): Deferred<T>
 
+    /**
+     * This method is used by generated clients to perform a call to the server
+     * that returns a streaming flow.
+     *
+     * @param T type of the result
+     * @param call an object that contains all required information about the called method,
+     * that is needed to route it properly to the server.
+     * @return the actual result of the call, for example, data from the server
+     */
+    public fun <T> callServerStreaming(call: RpcCall): Flow<T> {
+        error("Non-suspending server streaming is not supported by this client")
+    }
+
     /**
      * Provides child [CoroutineContext] for a new [RemoteService] service stub.
      *

core/src/commonMain/kotlin/kotlinx/rpc/descriptor/RpcServiceDescriptor.kt

@@ -62,6 +62,7 @@ public class RpcCallable<@Rpc T : Any>(
     public val returnType: RpcType,
     public val invokator: RpcInvokator<T>,
     public val parameters: Array<RpcParameter>,
+    public val isNonSuspendFunction: Boolean,
 )
 
 @ExperimentalRpcApi

gradle-plugin/src/main/kotlin/kotlinx/rpc/Extensions.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 @file:Suppress("unused")
@@ -64,16 +64,16 @@ open class RpcStrictModeExtension @Inject constructor(objects: ObjectFactory) {
     val nestedFlow: Property<RpcStrictMode> = objects.strictModeProperty()
 
     /**
-     * WIP: https://youtrack.jetbrains.com/issue/KRPC-133
-     * Will be enabled later, when an alternative is ready.
+     * StreamScoped functions are deprecated.
      */
-    private val streamScopedFunctions: Property<RpcStrictMode> = objects.strictModeProperty(RpcStrictMode.NONE)
+    val streamScopedFunctions: Property<RpcStrictMode> = objects.strictModeProperty()
 
     /**
-     * WIP: https://youtrack.jetbrains.com/issue/KRPC-133
-     * Will be enabled later, when an alternative is ready.
+     * Suspending functions with server-streaming are deprecated in RPC.
+     *
+     * Consider returning a Flow in a non-suspending function.
      */
-    private val suspendingServerStreaming: Property<RpcStrictMode> = objects.strictModeProperty(RpcStrictMode.NONE)
+    val suspendingServerStreaming: Property<RpcStrictMode> = objects.strictModeProperty()
 
     /**
      * Not top-level flows in the return value are deprecated in RPC for streaming.

gradle-plugin/src/main/kotlin/kotlinx/rpc/compilerPlugins.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2024 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 @file:Suppress("detekt.ClassNaming", "ClassName")
@@ -36,12 +36,11 @@ class CompilerPluginCli : KotlinCompilerPluginSupportPlugin by compilerPlugin({
                 SubpluginOption("strict-stateFlow", strict.stateFlow.get().toCompilerArg()),
                 SubpluginOption("strict-sharedFlow", strict.sharedFlow.get().toCompilerArg()),
                 SubpluginOption("strict-nested-flow", strict.nestedFlow.get().toCompilerArg()),
-                // WIP: https://youtrack.jetbrains.com/issue/KRPC-133
-//                SubpluginOption("strict-stream-scope", strict.streamScopedFunctions.get().toCompilerArg()),
-//                SubpluginOption(
-//                    "strict-suspending-server-streaming",
-//                    strict.suspendingServerStreaming.get().toCompilerArg()
-//                ),
+                SubpluginOption("strict-stream-scope", strict.streamScopedFunctions.get().toCompilerArg()),
+                SubpluginOption(
+                    "strict-suspending-server-streaming",
+                    strict.suspendingServerStreaming.get().toCompilerArg()
+                ),
                 SubpluginOption("strict-not-top-level-server-flow", strict.notTopLevelServerFlow.get().toCompilerArg()),
                 SubpluginOption("strict-fields", strict.fields.get().toCompilerArg()),
                 @OptIn(RpcDangerousApi::class)

krpc/krpc-client/api/krpc-client.api

@@ -2,6 +2,7 @@ public abstract class kotlinx/rpc/krpc/client/KrpcClient : kotlinx/rpc/krpc/inte
 	public fun <init> (Lkotlinx/rpc/krpc/KrpcConfig$Client;Lkotlinx/rpc/krpc/KrpcTransport;)V
 	public final fun call (Lkotlinx/rpc/RpcCall;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun callAsync (Lkotlinx/coroutines/CoroutineScope;Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/Deferred;
+	public fun callServerStreaming (Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/flow/Flow;
 	protected final fun getConfig ()Lkotlinx/rpc/krpc/KrpcConfig$Client;
 	public synthetic fun getConfig ()Lkotlinx/rpc/krpc/KrpcConfig;
 	public final fun getCoroutineContext ()Lkotlin/coroutines/CoroutineContext;