title: Wait for several Flows to finish before proceeding
body: <p>I need to upload an image to a server, and to do so I cannot use other libraries, but rather I split it (base64 encoded) in chunks and upload them all.</p>
  
  <p>I am using Kotlin coroutine Flows for it, and what I currently do is to do a first call (returning a flow) to obtain a image ID that I need to append in all the upload requests</p>
  
  <p>Here are the 2 functions I use to upload the image</p>
  
  <pre><code>fun submitImage(payload: Payload): Flow&lt;String&gt; {
      val request = requestBuilder.buildUploadImageRequest(payload)
      return client.execute(request)
          .serviceFlow({ response -&gt;
              val imageId = response.body.id
              uploadImage(payload.imageBase64, imageId)
              imageId
          }, { response -&gt;
              throw MyServerError("Error ${response.error}")
          })
  }
  
  private fun uploadImage(imageBase64: String, imageId: String) {
      val chunks = divideEncodedImageInChunksOfSize(imageBase64)
      var v = 1
      for (chunk in chunks) {
          val payload = generatePayload(imageId, v, chunk, false)
          submitImageChunk(payload)
          v++
      }
      val payload = generatePayload(imageId, v, "", true)
      submitImageChunk(payload)
  }
  
  private fun submitImageChunk(payload: JSONObject): Flow&lt;Unit&gt; {
      val request = requestBuilder.buildUploadImageChunkRequest(payload)
      return client.execute(request)
          .serviceFlow({ }, { response -&gt;
              throw MyHttpError(response)
          })
  }
  </code></pre>
  
  <p>And I make use of the following utility functions</p>
  
  <pre><code>// Extension function to handle Flows and their activation
  internal fun MyHttpClient.execute(request: MyHttpRequest): Flow&lt;MyHttpResponse&gt; {
      return flow {
          val deferred = CompletableDeferred&lt;MyHttpResponse&gt;()
          executeHttp(request, object : MyHttpListener {
              override fun onSuccess(response: MyHttpResponse) {
                  deferred.complete(response)
              }
  
              override fun onFailure(response: MyHttpResponse) {
                  deferred.completeExceptionally(MyHttpError(response))
              }
          })
          emit(deferred.await())
      }
  }
  
  // Extension function to catch exceptions AND to check if the response body is null
  internal fun &lt;T&gt; Flow&lt;MyHttpResponse&gt;.serviceFlow(
      onSuccess: (response: MyHttpResponse) -&gt; T,
      onError: (response: MyHttpResponse) -&gt; Unit
  ) = flatMapConcat { response -&gt;
      flowOf(response)
          .map { res -&gt;
              res.body?.let { it -&gt;
                  onSuccess(res)
              } ?: throw MyParseError("MyHttpResponse has a null body")
          }
          .catchException&lt;JSONException, T&gt; { e -&gt;
              throw MyParseError("Parsing exception $e")
          }
  }.catchException&lt;MyHttpError, T&gt; { e -&gt;
      onError(e.response)
  }
  
  // Function leveraging OkHttpClient to make a HTTPRequest
  internal fun executeHttp { ... }
  </code></pre>
  
  <p>The problem I think is due to the fact that the function <code>submitImage</code> returns after launching all the sub-Flows for the upload of the image but it does not wait for all of them to complete.
  I am not sure what construct Kotlin coroutines has for a use-case like this, can anybody help me out?</p>