title: Kotlin flows as a message queue between coroutines
body: <p>I'm attempting to use Kotlin's <code>Flow</code> class as a message queue to transfer data from a producer (a camera) to a set of workers (image analyzers) running on separate coroutines.</p>
  
  <p>The producer in my case is a camera, and will run substantially faster than the workers. Back pressure should be handled by dropping data so that the image analyzers are always operating on the latest images from the camera.</p>
  
  <p>When using channels, this solution works, but seems messy and does not provide an easy way for me to translate the data between the camera and the analyzers (like <code>flow.map</code>).</p>
  
  <pre class="lang-kotlin prettyprint-override"><code>class ImageAnalyzer&lt;Result&gt; {
      fun analyze(image: Bitmap): Result {
          // perform some work on the image and return a Result. This can take a long time.
      }
  }
  
  class CameraAdapter {
  
      private val imageChannel = Channel&lt;Bitmap&gt;(capacity = Channel.RENDEZVOUS)
      private val imageReceiveMutex = Mutex()
  
      // additional code to make this camera work and listen to lifecycle events of the enclosing activity.
  
      protected fun sendImageToStream(image: CameraOutput) {
          // use channel.offer to ensure the latest images are processed
          runBlocking { imageChannel.offer(image) }
      }
  
      @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
      fun onDestroy() {
          runBlocking { imageChannel.close() }
      }
  
      /**
       * Get the stream of images from the camera.
       */
      fun getImageStream(): ReceiveChannel&lt;Bitmap&gt; = imageChannel
  }
  
  class ImageProcessor&lt;Result&gt;(workers: List&lt;ImageAnalyzer&lt;Result&gt;&gt;) {
      private val analysisResults = Channel&lt;Result&gt;(capacity = Channel.RENDEZVOUS)
      private val cancelMutex = Mutex()
  
     var finished = false // this can be set elsewhere when enough images have been analyzed
  
      fun subscribeTo(channel: ReceiveChannel&lt;Bitmap&gt;, processingCoroutineScope: CoroutineScope) {
          // omit some checks to make sure this is not already subscribed
  
          processingCoroutineScope.launch {
              val workerScope = this
              workers.forEachIndexed { index, worker -&gt;
                  launch(Dispatchers.Default) {
                      startWorker(channel, workerScope, index, worker)
                  }
              }
          }
      }
  
      private suspend fun startWorker(
          channel: ReceiveChannel&lt;Bitmap&gt;,
          workerScope: CoroutineScope,
          workerId: Int,
          worker: ImageAnalyzer
      ) {
          for (bitmap in channel) {
              analysisResults.send(worker.analyze(bitmap))
  
              cancelMutex.withLock {
                  if (finished &amp;&amp; workerScope.isActive) {
                      workerScope.cancel()
                  }
              }
          }
      }
  }
  
  class ExampleApplication : CoroutineScope {
      private val cameraAdapter: CameraAdapter = ...
      private val imageProcessor: ImageProcessor&lt;Result&gt; = ...
  
      fun analyzeCameraStream() {
          imageProcessor.subscribeTo(cameraAdapter.getImageStream())
      }
  }
  </code></pre>
  
  <p>What's the proper way to do this? I would like to use a <code>ChannelFlow</code> instead of a <code>Channel</code> to pass data between the camera and the <code>ImageProcessor</code>. This would allow me to call <code>flow.map</code> to add metadata to the images before they're sent to the analyzers. However, when doing so, each ImageAnalyzer gets a copy of the same image instead of processing different images in parallel. Is it possible to use a <code>Flow</code> as a message queue rather than a broadcaster?</p>