Flow<T>.share() operator - caches the latest value

share.kt

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

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
            }
        }
}

test_activity_main.xml

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:orientation="vertical"
    tools:context=".MainActivity">

    <TextView
        android:id="@+id/tv1"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content" />

    <TextView
        android:id="@+id/tv2"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content" />

    <Button
        android:id="@+id/btn1"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="JOB A" />

    <Button
        android:id="@+id/btn2"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="JOB B" />

</LinearLayout>

test_MainActivity.kt

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import androidx.lifecycle.lifecycleScope
import kotlinx.android.synthetic.main.activity_main.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        var jobA: Job? = null
        var jobB: Job? = null

        val flow: Flow<Int> = flow {
            var i = 0
            while (true) {
                emit(i++)
                delay(500)
            }
        }.share()

        btn1.setOnClickListener {
            if (jobA == null) {
                jobA = lifecycleScope.launch {
                    flow
                        .collect {
                            tv1.text = it.toString()
                        }
                }
            } else {
                jobA?.cancel()
                jobA = null
            }
        }

        btn2.setOnClickListener {
            if (jobB == null) {
                jobB = lifecycleScope.launch {
                    flow
                        .collect {
                            tv2.text = it.toString()
                        }
                }
            } else {
                jobB?.cancel()
                jobB = null
            }
        }
    }
}