Any way to "join" a currently-active Continuation in Kotlin Coroutines?

Question

I feel like I'm close here.

Let's say I have a method with a suspendCancellableCoroutine: (Sorry for any syntactical/etc. errors, just trying to get an idea down)

suspend fun foo(): String {
    suspendCancellableCoroutine {
       val listener = object : Listener {
           override fun theThingHappened() {
               it.resume("The result")
           }
       }

       // Long involved process
       thirdPartyThing.doSomethingWithListener(listener)
    }
}

I call that method once, it does its thing and returns a string. But if another process calls the same method while the first CancellableContinuation is still running, is there a way to just join it and return the same result?

The reason I'm asking is that buried in my suspendCancellableCoroutine, there's a third-party method that can only run once at any given time, and attempting to run it again before the first one has completed will throw an error.

So let's say that foo() takes 10 seconds to run. It gets called once and will return the result string. But suppose that 5 seconds later another process calls it. Instead of spawning off another suspendCancellableCoroutine, can I just have the 2nd call wait for the first to finish and return that value as well?

I hope I've described this well.

Answer 1

Suppose you leave foo() alone as you have defined it and create a new public function bar() to do what you requested. I think you could do it something like this by storing a coroutine Deferred in a property. The scope here is a CoroutineScope appropriate for the current class. Maybe the current class creates and manages its own scope, or an outer class passes an appropriate scope to its constructor.

private val mutex = Mutex()
private var fooDeferred: Deferred<String>? = null

suspend fun bar(): String = mutex.withLock {
    fooDeferred ?: scope.async {
            val result = foo()
            mutex.withLock { fooDeferred = null }
            result
        }.also { fooDeferred = it }
}.await()

---

Another possible strategy using a SharedFlow with shareIn such that it restarts the flow whenever it has run out of subscribers and gets a new one:

private val fooFlow = flow { 
        emit(foo())
    }.shareIn(scope, SharingStarted.WhileSubscribed())

suspend fun bar() = fooFlow.first()

Note: I didn’t test either concept.

Answer 2

You don't necessarily need to use coroutine machinery to achieve what you want here. How would you phrase and approach this question if coroutines weren't involved? You could still have exactly the same scenario, where you want to join an existing listener if there is one, or add a new listener otherwise.

The solutions involving coroutine machinery are likely to be a bit messy. In most cases you're going to need a separate coroutine scope, as you'll be launching coroutines that are shared between multiple consumers.

I think there's a lot of value in solving this problem for the general case of callbacks with no coroutines. Then you can simply apply that solution to your existing code.

My solution involves creating a new listener which will wrap all the existing listeners. It will register itself as a listener with the third party library. When it receives the callback, it will fan-out and notify all of its own listeners.

I'm using a lock-free Trieber stack to store the list of listeners. Not only does this make the whole thing thread-safe without locking, but it also means we can atomically identify when the list goes from empty to non-empty. That makes it easy to know when to invoke the third-party library. If you think this is overkill, it would be easy enough to make a simpler implementation that uses a lock.

class SingleListenerObservable(private val delegate: Observable) : Observable {
    private class ListenerStackNode(val listener: Listener, val next: ListenerStackNode?)

    private val listeners = AtomicReference<ListenerStackNode?>()

    private val actualListener = object : Listener {
        override fun theThingHappened() {
            val head = clear()
            generateSequence(head) { it.next }
                .forEach { it.listener.theThingHappened() }
        }
    }

    override fun addListener(listener: Listener) {
        val head = push(listener)
        if (head.next == null) {
            // we are the first listener
            delegate.addListener(actualListener)
        }
    }

    /**
     * Add a listener to the stack, and return the new head of the stack
     */
    private fun push(listener: Listener): ListenerStackNode {
        var newHead: ListenerStackNode
        do {
            val oldHead = listeners.get()
            newHead = ListenerStackNode(listener, oldHead)
            val success = listeners.compareAndSet(oldHead, newHead)
        } while (!success)
        return newHead
    }

    /**
     * Clear the stack, and return the old head of the stack
     */
    private fun clear(): ListenerStackNode? {
        var oldHead: ListenerStackNode?
        do {
            oldHead = listeners.get()
            val success = listeners.compareAndSet(oldHead, null)
        } while (!success)
        return oldHead
    }
}

Using this approach, you could just wrap your third party library with this decorator class, and use suspendCancellableCoroutine to add listeners as you normally would.

Answer 3

Thank you all for the suggestions! In the time since posting this question, I arrived at an answer that seems to work for me.

Basically I do something like this. Of course this is pseudo-ish code but the gist is that I have a SharedFlow that emits Foo values. Inside the getFoo method, the first thing I do in all cases is set ourselves up to collect the flow and resume the continuation once we get a value.

Then, if the process is not already running, we call requestFoo() with a listener that emits to the flow.

Calling this method several times in succession before the process finishes results in one call to requestFoo(), but each call to the method gets a Foo returned from the method. If the process finishes and we call getFoo() again, the whole thing starts anew.

Note that I'm not really concerned just yet with cancelling an existing process if I no longer need it, etc.

private val fooFlow = MutableSharedFlow<Foo>()

private suspend fun getFoo(): Foo = suspendCancellableCoroutine { continuation ->
    lifecycleScope.launch {
        geoComplyTokenFlow.collect {
            continuation.resume(it)
        }
    }
    
    // If we're NOT already running this process, 
    // set up request listener and make the call
    if (!fooManager.isFooProcessRunning) {
        val fooProcessListener: FooProcessListener = FooProcessListener {
            // We just got a Foo result
            lifecycleScope.launch { foo ->
                fooFlow.emit(foo)
            }
        }

        fooManager.requestFoo(fooProcessListener)
    }