Hi, I need share operator. The implementations seem to be quite complicated so I've written this. Do you have any idea why it shouldn't work correctly?

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*
import java.util.concurrent.atomic.*

fun <T> Flow<T>.share(): Flow<T> {
  val channel = ConflatedBroadcastChannel<T>()
  val counter = AtomicInteger()
  var job: Job? = null

  return channel
    .asFlow()
    .onStart {
      if (counter.incrementAndGet() == 1) {
        job = GlobalScope.launch {
          this@share
            .catch {
              channel.close(it)
            }
            .collect {
              channel.offer(it)
            }
        }
      }
    }
    .onCompletion {
      if (counter.decrementAndGet() == 0) {
        job?.cancelAndJoin()
        job = null
      }
    }
}

You have two pieces of state here:

counter

and

job

and there's a race in how you check/update them. E.g. you can end up in an invalid state with

counter > 0 && job == null

Thanks!

Also there is an issue with Conflated channel - it seems unexpected that

share()

operator will conflate emission by default. You probably need to make it

fun <T> Flow<T>.share(capacity: Int = CONFLATED/BUFFERED): Flow<T> { 
    val channel = BroadcastChannel<T>(capacity)
.....

Isn't rendevouz more clean option? share() operator will do the sharing but not replay?

there is no rendevouz broadcast channel - only conflated and buffered. Closest to rendevous is buffered with capacity = 1

Why not? It will replay the last element if there is an open subscription, won't it?

with Conflated will, with Buffered won't - Conflated: https://pl.kotl.in/tOI0C_NCA Buffered: https://pl.kotl.in/NtvlYT9gV You can check ArrayBroadcastChannel (it's under the hood when you choose BUFFERED) docs:

* Note, that elements that are sent to this channel while there are no subscribers are immediately lost.

👆 it applies to the second subscriber as well - if first subscriber consumed element, it lost