Make Flow.fold inlineable

    * It is not that useful in an application code (-> is extracted into method), so won't bloat bytecode too much
    * It is crucial enough as building block to avoid excess allocations
    * Additionally mark inlined builders with labeled return to workaround KT-28938

Author: qwwdfsad

kotlinx-coroutines-core/common/src/flow/operators/Context.kt

@@ -50,7 +50,7 @@ public fun <T> Flow<T>.flowOn(flowContext: CoroutineContext, bufferSize: Int = 1
         coroutineScope {
             val channel = produce(flowContext, capacity = bufferSize) {
                 collect { value ->
-                    send(value)
+                    return@collect send(value)
                 }
             }
             channel.consumeEach { value ->
@@ -98,7 +98,7 @@ public fun <T, R> Flow<T>.flowWith(
         val originalContext = coroutineContext.minusKey(Job)
         val prepared = source.flowOn(originalContext, bufferSize)
         builder(prepared).flowOn(flowContext, bufferSize).collect { value ->
-            emit(value)
+            return@collect emit(value)
         }
     }
 }

kotlinx-coroutines-core/common/src/flow/operators/Transform.kt

@@ -29,7 +29,8 @@ import kotlinx.coroutines.flow.unsafeFlow as flow
 public inline fun <T, R> Flow<T>.transform(@BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit): Flow<R> {
     return flow {
         collect { value ->
-            transform(value)
+            // kludge, without it Unit will be returned and TCE won't kick in, KT-28938
+            return@collect transform(value)
         }
     }
 }
@@ -40,7 +41,7 @@ public inline fun <T, R> Flow<T>.transform(@BuilderInference crossinline transfo
 @FlowPreview
 public inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T> = flow {
     collect { value ->
-        if (predicate(value)) emit(value)
+        if (predicate(value)) return@collect emit(value)
     }
 }
 
@@ -50,7 +51,7 @@ public inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boole
 @FlowPreview
 public inline fun <T> Flow<T>.filterNot(crossinline predicate: suspend (T) -> Boolean): Flow<T> = flow {
     collect { value ->
-        if (!predicate(value)) emit(value)
+        if (!predicate(value)) return@collect emit(value)
     }
 }
 
@@ -66,15 +67,15 @@ public inline fun <reified R> Flow<*>.filterIsInstance(): Flow<R> = filter { it
  */
 @FlowPreview
 public fun <T: Any> Flow<T?>.filterNotNull(): Flow<T> = flow<T> {
-    collect { value -> if (value != null) emit(value) }
+    collect { value -> if (value != null) return@collect  emit(value) }
 }
 
 /**
  * Returns a flow containing the results of applying the given [transform] function to each value of the original flow.
  */
 @FlowPreview
 public inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -> R): Flow<R> = transform { value ->
-    emit(transform(value))
+   return@transform emit(transform(value))
 }
 
 /**
@@ -83,7 +84,7 @@ public inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -
 @FlowPreview
 public inline fun <T, R: Any> Flow<T>.mapNotNull(crossinline transform: suspend (value: T) -> R?): Flow<R> = transform { value ->
     val transformed = transform(value) ?: return@transform
-    emit(transformed)
+    return@transform emit(transformed)
 }
 
 /**

kotlinx-coroutines-core/common/src/flow/terminal/Reduce.kt

@@ -38,9 +38,9 @@ public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S,
  * Accumulates value starting with [initial] value and applying [operation] current accumulator value and each element
  */
 @FlowPreview
-public suspend fun <T, R> Flow<T>.fold(
+public suspend inline fun <T, R> Flow<T>.fold(
     initial: R,
-    operation: suspend (acc: R, value: T) -> R
+    crossinline operation: suspend (acc: R, value: T) -> R
 ): R {
     var accumulator = initial
     collect { value ->