Kotlin Flows Java Interop Callback

user_2959479 0
Viewed 3177

I´ve been looking for a suitable solution or best practice when I want to use Kotlin Flows with ordinary callbacks. My use case is that I write a kotlin library that uses Kotlin Flow internally and i have to assume that the users will use Java for instance. So I thought that the best solution is to overload a basic callback interface to my flow method and call it in collect something like this:

class KotlinClass {

    interface Callback {
        fun onResult(result: Int)
    }

    private fun foo() = flow {
        for (i in 1..3) {
            emit(i)
        }
    }

    fun bar(callback: Callback) {
        runBlocking {
            foo().collect { callback.onResult(it) }
        }
    }

  private fun main() {
    bar(object : Callback {
        override fun onResult(result: Int) {
            TODO("Not yet implemented")
        }
    })
}

and in my Java Application i can simply use it like that:

public class JavaClass {

    public void main() {
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(new KotlinClass.Callback() {
            @Override
            public void onResult(int result) {
                // TODO("Not yet implemented")
            }
        });
    } 
}

I am not sure whats the way to go because I would like to have my Kotlin library that uses Flows usable in a good fashion for Java and Kotlin.

I came across callbackFlow but that seems to be only if I want to let´s call it flow-ify a callback-based API? Because I am quite new to Kotlin and Flows please apologise if my question is flawed in cause of missing some basic concepts of kotlin.

kotlin-flow kotlin-coroutines kotlin java
3 Answers

I would give the Java client more control over the flow. I would add a onStart and onCompletion method to your callback interface. Beside this I would use an own CoroutineScope - maybe customizable from the Java client. And I would not block the calling thread from within the Kotlin function - no runBlocking.

@InternalCoroutinesApi
class KotlinClass {
    val coroutineScope = CoroutineScope(Dispatchers.Default)

    interface FlowCallback {
        @JvmDefault
        fun onStart() = Unit

        @JvmDefault
        fun onCompletion(thr: Throwable?) = Unit
        fun onResult(result: Int)
    }

    private fun foo() = flow {
        for (i in 1..3) {
            emit(i)
        }
    }

    fun bar(flowCallback: FlowCallback) {
        coroutineScope.launch {
            foo().onStart { flowCallback.onStart() }
                .onCompletion { flowCallback.onCompletion(it) }
                .collect { flowCallback.onResult(it) }
        }
    }

    fun close() {
        coroutineScope.cancel()
    }    
}

Now the Java client is in full control how to start, collect and cancel the flow. For example you could use a latch to wait for completion, set an timeout and cancel the couroutine scope. This looks in the first place like a lot of code, but typically you will need this kind of flexibility.

public class JavaClass {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(new KotlinClass.FlowCallback() {
            @Override
            public void onCompletion(@Nullable Throwable thr) {
                latch.countDown();
            }


            @Override
            public void onResult(int result) {
                System.out.println(result);
            }
        });

        try {
            latch.await(5, TimeUnit.SECONDS);
        } finally {
            libraryClass.close();
        }
    }
}
user_91644710

You don't need to create a interface in the Kotlin code. You can define bar like that: 

 fun bar(callback: (Int) -> Unit) {
     runBlocking {
         foo().collect { callback(it) }
     }
 }

From the Java code you can call the function like that:

public class JavaClass {

    public static void main(String[] args) {
        KotlinClass libraryClass = new KotlinClass();
        libraryClass.bar(v -> { System.out.println(v); return Unit.INSTANCE; });
    } 
}
user_16788930

In case anyone wondering for a general solution. Here's our version of enhancement from @rene answer here.

  1. Accept a generic type
  2. A configurable coroutineScope
// JavaFlow.kt
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.launch

@InternalCoroutinesApi
class JavaFlow<T>(
    private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
) {

    interface OperatorCallback <T> {
        @JvmDefault
        fun onStart() = Unit

        @JvmDefault
        fun onCompletion(thr: Throwable?) = Unit
        fun onResult(result: T)
    }

    fun collect(
        flow: Flow<T>,
        operatorCallback: OperatorCallback<T>,
    ) {
        coroutineScope.launch {
            flow
                .onStart { operatorCallback.onStart() }
                .onCompletion { operatorCallback.onCompletion(it) }
                .collect { operatorCallback.onResult(it) }
        }
    }

    fun close() {
        coroutineScope.cancel()
    }
}

Java caller-side:

// code omitted...
new JavaFlow<File>().collect(
        // compressImageAsFlow is our actual kotlin flow extension
        FileUtils.compressImageAsFlow(file, activity),
        new JavaFlow.OperatorCallback<File>() {
            @Override
            public void onResult(File result) {
                // do something with the result here
                SafeSingleton.setFile(result);
            }
        }
);


// or using lambda with method references
// new JavaFlow<File>().collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );

// Change coroutineScope to Main
// new JavaFlow<File>(CoroutineScopeKt.MainScope()).collect(
//        FileUtils.compressImageAsFlow(file, activity),
//        SafeSingleton::setFile
// );

OperatorCallback.onStart and OperatorCallback.onCompletion is optional, override it as needed.

user_37630320
Related