Here we have taken a example where firebase is taken as a service we are expecting a callback. In place of firebase it can be a normal API call or any other third party API call

Definition.kt

// DEFINITION: -> Creates a CompletableDeferred in an active state. It is optionally a child of a parent job.
@Suppress("FunctionName")
public fun <T> CompletableDeferred(parent: Job? = null): CompletableDeferred<T> = CompletableDeferredImpl(parent)

LoginService.kt

class LoginService @Inject constructor(
    private val service: FirebaseAuthRepository,
    private var  log: LoggerRepository
) {

    companion object {
        const val RESULT_FAILURE = "firebase returned failure for the register user request"
    }

    fun registerUser(
        input: RegistrationInput
    ): Flow<State<User>> {

        // Using the API to wait
        val resultDeferred = CompletableDeferred<State<User>>()

        val result = service.getFirebaseAuth()
            .createUserWithEmailAndPassword(input.email, input.password)
            .addOnCompleteListener {
                if (it.isSuccessful) {
                    it.result.user?.let { firebaseUser ->
                        val user = User(
                            firebaseUser.uid, input.firstName, input.lastName, input.email
                        )
                        // Send callback as complete 
                        resultDeferred.complete(State.success(user))
                    }
                } else {
                    // Send callback as failure 
                    resultDeferred.complete(State.failed(it.exception?.message.toString()))
                }
            }

        return flow {
            try {
                // Here we set await for the thread
                emit(resultDeferred.await())
            } catch (e: Exception) {
                log.e(FEATURE_LOGIN, e.stackTrace.toString())
            }
        }
    }

}