how to resolve FATAL EXCEPTION: DefaultDispatcher-worker-4

Question

I have written a new usecase to communicate to api which uses Flow, I am guessing I am not handing the threading properly in the Usecase between Main thread and IO thread,

This is the error I get

-01-18 02:20:40.555 26602-26870/com.xxx.xx.staging E/AndroidRuntime: FATAL EXCEPTION: DefaultDispatcher-worker-4
    Process: com.xxx.xx.staging, PID: 26602
    java.lang.IllegalStateException: Event bus [Bus "fill_order"] accessed from non-main thread null
        at com.squareup.otto.ThreadEnforcer$2.enforce(ThreadEnforcer.java:47)
        at com.squareup.otto.Bus.post(Bus.java:317)
        at com.xxx.xx.fragments.filller.fillorder.BasefillOrderFragment.postFldeckStatusUpdateEvent(BasefillOrderFragment.java:117)
        at com.xxx.xx.fragments.filller.fillorder.fillOrderDataFragment.postFldeckStatusUpdateEvent(fillOrderDataFragment.java:1955)
        at com.xxx.xx.fragments.filller.fillorder.fillOrderDataFragment.updateView(fillOrderDataFragment.java:845)
        at com.xx.xx.fragments.filller.fillorder.fillOrderDataFragment.legacyUpdateView(fillOrderDataFragment.java:2317)
        at com.xxx.xx.clean.fillorder.presenter.BasefillDataPresenter.onStartfilllingSuccess(BasefillDataPresenter.kt:460)
        at com.xxx.xx.clean.fillorder.presenter.BasefillDataPresenter.handleStartfilllingClicked(BasefillDataPresenter.kt:315)
        at com.xxx.xx.clean.fillorder.presenter.BasefillDataPresenter.access$handleStartfilllingClicked(BasefillDataPresenter.kt:49)

The error is at handleStartfilllingClicked(view, it) in . collect

I am calling startfilllingUseCaseFlow usecase which might be the issue

@FlowPreview
fun initFlowSubscription(view: View) {
    launch {
        view.startfilllingObservableFlow
            .conflate()
            .catch {
                onStartfilllingError(view)
            }
            .flatMapMerge {
                if (!hasOpenInopIncidents()) {
                    equipmentProvider.get()?.let {
                        startfilllingUseCaseFlow(StartfilllingUseCaseFlow.Params(it))
                    }!!
                } else {
                    val incidentOpenResponse = GenericResponse(false)
                    incidentOpenResponse.error = OPEN_INCIDENTS
                    flowOf(incidentOpenResponse)
                }
            }
            .collect {
                handleStartfilllingClicked(view, it) // ERROR IS HERE
            }
    }
}


private fun handleStartfilllingClicked(view: View, response: GenericResponse) {
    if (response.success == false && response.error == OPEN_INCIDENTS) {
        view.showCannotProceedInopIncidentDialog()
        view.hideLoader(false)
        return
    }

    onStartfilllingSuccess(view) // Error is here
}

StartfilllingUseCaseFlow

class StartfilllingUseCaseFlow @Inject constructor(
    private val currentOrderStorage: CurrentOrderStorage,
    private val fillOrderRepository: fillOrderRepository,
    private val app: App
): FlowUseCase<StartfilllingUseCaseFlow.Params, GenericResponse>() {

    override suspend fun run(params: Params): Flow<GenericResponse> {
        val startTime = DateTime()
        val action = TimestampedAction(
            app.session.user.id, null, startTime
        )
        return flowOf(fillOrderRepository.startfilllingSuspend(
            currentOrderStorage.fillOrder!!.id,
            action
        )).onEach { onSuccess(startTime, params.equipment) }
            .catch { e -> e.message?.let { onError(it) } }
            .flowOn(Dispatchers.IO)
    }

    private fun onSuccess(startTime: DateTime, equipment: Equipment) {
        if (currentOrderStorage.getfillOrder() == null) return
        currentOrderStorage.getfillOrder()!!.setStatus(fillOrderData.STATUS_fillLING)
        equipment.times.start = startTime
        app.saveState()
    }

    private fun onError(errorMessage: String) {
        Timber.e(errorMessage, "Error calling started fillling! %s", errorMessage)
    }

    data class Params(val equipment: Equipment)

}

I am guessing I am not handing IO and Main thread properly here

abstract class FlowUseCase<in Params, out T>() {

    abstract suspend fun run(params: Params): Flow<T>

    suspend operator fun invoke(params: Params): Flow<T> = run(params).flowOn(Dispatchers.IO)
}

Could you suggest where I am gettig it wrong

Thanks R

Answer 1

Coroutines sometimes consume unhandled exceptions (this seems to be more prevalent when using async/await). Anyway, add a CoroutineExceptionHandler in these cases.

CoroutineScope(IO + coroutineExceptionHandler).launch {
 //perform background task
}

val coroutineExceptionHandler = CoroutineExceptionHandler{_, throwable ->
    Log.d("coroutineExceptionHandler", "yes this happened")
    throwable.printStackTrace()
}

Answer 2

You are trying to update the view in Coroutines default thread. All views updates must be in the MainThread. try:

 fun initFlowSubscription(view: View) {
      launch(Dispatchers.Main) {
      //enter code here
      }
    }

This might give another error because you are doing too much process in the main thread. To avoid that you. could use "async" and update your view after:

Exemple:

 fun initFlowSubscription(view: View) {
      launch(Dispatchers.Main) {
         val asyncValue = async(Dispatchers.IO) {
            //Do yours suspend fun
        }
        val value = asyncValue.await()
      }
    }

This example should yours fine and avoid stopping the users UI