@file:JvmMultifileClass @file:JvmName("FlowKt") @file:Suppress("UNCHECKED_CAST") package kotlinx.coroutines.flow import kotlinx.coroutines.flow.internal.* import kotlin.jvm.* import kotlinx.coroutines.flow.flow as safeFlow import kotlinx.coroutines.flow.internal.unsafeFlow as flow /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. * * It can be demonstrated with the following example: * ``` * val flow = flowOf(1, 2).onEach { delay(10) } * val flow2 = flowOf("a", "b", "c").onEach { delay(15) } * flow.combine(flow2) { i, s -> i.toString() + s }.collect { * println(it) // Will print "1a 2a 2b 2c" * } * ``` * * This function is a shorthand for `flow.combineTransform(flow2) { a, b -> emit(transform(a, b)) } */ @JvmName("flowCombine") public fun Flow.combine(flow: Flow, transform: suspend (a: T1, b: T2) -> R): Flow = flow { combineInternal(arrayOf(this@combine, flow), nullArrayFactory(), { emit(transform(it[0] as T1, it[1] as T2)) }) } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. * * It can be demonstrated with the following example: * ``` * val flow = flowOf(1, 2).onEach { delay(10) } * val flow2 = flowOf("a", "b", "c").onEach { delay(15) } * combine(flow, flow2) { i, s -> i.toString() + s }.collect { * println(it) // Will print "1a 2a 2b 2c" * } * ``` * * This function is a shorthand for `combineTransform(flow, flow2) { a, b -> emit(transform(a, b)) } */ public fun combine(flow: Flow, flow2: Flow, transform: suspend (a: T1, b: T2) -> R): Flow = flow.combine(flow2, transform) /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. * * Its usage can be demonstrated with the following example: * ``` * val flow = requestFlow() * val flow2 = searchEngineFlow() * flow.combineTransform(flow2) { request, searchEngine -> * emit("Downloading in progress") * val result = download(request, searchEngine) * emit(result) * } * ``` */ @JvmName("flowCombineTransform") public fun Flow.combineTransform( flow: Flow, @BuilderInference transform: suspend FlowCollector.(a: T1, b: T2) -> Unit ): Flow = combineTransformUnsafe(this, flow) { args: Array<*> -> transform( args[0] as T1, args[1] as T2 ) } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. * * Its usage can be demonstrated with the following example: * ``` * val flow = requestFlow() * val flow2 = searchEngineFlow() * combineTransform(flow, flow2) { request, searchEngine -> * emit("Downloading in progress") * val result = download(request, searchEngine) * emit(result) * } * ``` */ public fun combineTransform( flow: Flow, flow2: Flow, @BuilderInference transform: suspend FlowCollector.(a: T1, b: T2) -> Unit ): Flow = combineTransformUnsafe(flow, flow2) { args: Array<*> -> transform( args[0] as T1, args[1] as T2 ) } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. */ public fun combine( flow: Flow, flow2: Flow, flow3: Flow, @BuilderInference transform: suspend (T1, T2, T3) -> R ): Flow = combineUnsafe(flow, flow2, flow3) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3 ) } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. */ public fun combineTransform( flow: Flow, flow2: Flow, flow3: Flow, @BuilderInference transform: suspend FlowCollector.(T1, T2, T3) -> Unit ): Flow = combineTransformUnsafe(flow, flow2, flow3) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3 ) } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. */ public fun combine( flow: Flow, flow2: Flow, flow3: Flow, flow4: Flow, transform: suspend (T1, T2, T3, T4) -> R ): Flow = combineUnsafe(flow, flow2, flow3, flow4) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3, args[3] as T4 ) } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. */ public fun combineTransform( flow: Flow, flow2: Flow, flow3: Flow, flow4: Flow, @BuilderInference transform: suspend FlowCollector.(T1, T2, T3, T4) -> Unit ): Flow = combineTransformUnsafe(flow, flow2, flow3, flow4) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3, args[3] as T4 ) } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. */ public fun combine( flow: Flow, flow2: Flow, flow3: Flow, flow4: Flow, flow5: Flow, transform: suspend (T1, T2, T3, T4, T5) -> R ): Flow = combineUnsafe(flow, flow2, flow3, flow4, flow5) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3, args[3] as T4, args[4] as T5 ) } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. */ public fun combineTransform( flow: Flow, flow2: Flow, flow3: Flow, flow4: Flow, flow5: Flow, @BuilderInference transform: suspend FlowCollector.(T1, T2, T3, T4, T5) -> Unit ): Flow = combineTransformUnsafe(flow, flow2, flow3, flow4, flow5) { args: Array<*> -> transform( args[0] as T1, args[1] as T2, args[2] as T3, args[3] as T4, args[4] as T5 ) } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. */ public inline fun combine( vararg flows: Flow, crossinline transform: suspend (Array) -> R ): Flow = flow { combineInternal(flows, { arrayOfNulls(flows.size) }, { emit(transform(it)) }) } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. */ public inline fun combineTransform( vararg flows: Flow, @BuilderInference crossinline transform: suspend FlowCollector.(Array) -> Unit ): Flow = safeFlow { combineInternal(flows, { arrayOfNulls(flows.size) }, { transform(it) }) } /* * Same as combine, but does not copy array each time, deconstructing existing * array each time. Used in overloads that accept FunctionN instead of Function> */ private inline fun combineUnsafe( vararg flows: Flow, crossinline transform: suspend (Array) -> R ): Flow = flow { combineInternal(flows, nullArrayFactory(), { emit(transform(it)) }) } /* * Same as combineTransform, but does not copy array each time, deconstructing existing * array each time. Used in overloads that accept FunctionN instead of Function> */ private inline fun combineTransformUnsafe( vararg flows: Flow, @BuilderInference crossinline transform: suspend FlowCollector.(Array) -> Unit ): Flow = safeFlow { combineInternal(flows, nullArrayFactory(), { transform(it) }) } // Saves bunch of anonymous classes private fun nullArrayFactory(): () -> Array? = { null } /** * Returns a [Flow] whose values are generated with [transform] function by combining * the most recently emitted values by each flow. */ public inline fun combine( flows: Iterable>, crossinline transform: suspend (Array) -> R ): Flow { val flowArray = flows.toList().toTypedArray() return flow { combineInternal( flowArray, arrayFactory = { arrayOfNulls(flowArray.size) }, transform = { emit(transform(it)) }) } } /** * Returns a [Flow] whose values are generated by [transform] function that process the most recently emitted values by each flow. * * The receiver of the [transform] is [FlowCollector] and thus `transform` is a * generic function that may transform emitted element, skip it or emit it multiple times. */ public inline fun combineTransform( flows: Iterable>, @BuilderInference crossinline transform: suspend FlowCollector.(Array) -> Unit ): Flow { val flowArray = flows.toList().toTypedArray() return safeFlow { combineInternal(flowArray, { arrayOfNulls(flowArray.size) }, { transform(it) }) } } /** * Zips values from the current flow (`this`) with [other] flow using provided [transform] function applied to each pair of values. * The resulting flow completes as soon as one of the flows completes and cancel is called on the remaining flow. * * It can be demonstrated with the following example: * ``` * val flow = flowOf(1, 2, 3).onEach { delay(10) } * val flow2 = flowOf("a", "b", "c", "d").onEach { delay(15) } * flow.zip(flow2) { i, s -> i.toString() + s }.collect { * println(it) // Will print "1a 2b 3c" * } * ``` * * ### Buffering * * The upstream flow is collected sequentially in the same coroutine without any buffering, while the * [other] flow is collected concurrently as if `buffer(0)` is used. See documentation in the [buffer] operator * for explanation. You can use additional calls to the [buffer] operator as needed for more concurrency. */ public fun Flow.zip(other: Flow, transform: suspend (T1, T2) -> R): Flow = zipImpl(this, other, transform)