我已经使用下面的流程结构超过5次了。因为大多数部分都是相同的。每个作业都有一个单独的use_case。每个useCases返回Flow<Resource<MyType>>。我如何在job1，job2...函数使用的公共函数中合并相同的部分？

Job1

private var job1Job: Job? = null
    private fun job1(param: Param) {
        job1Job?.cancel()
        job1Job = useCases1(param)
            .onEach { resource->
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = resource,
                        )
                    }
            }
            .catch {throwable->
                var currentTryCount = 
                var maxTryCount = 
                if ( throwable is BleReTryableError && currentTryCount < maxTryCount ) {
                    currentTryCount++
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Loading("loading message"),
                            currentTryCount = currentTryCount
                        )
                    }
                    throw BleReTryableError(throwable.message ?: "")
                } else {
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Error("error message"))
                    }
                }
            }
            .retry {
                delay(2500)
                it is BleReTryableError
            }
            .launchIn(viewModelScope)
    }

Job2

private var job2Job: Job? = null
    private fun job2(param: Param) {
        job2Job?.cancel()
        job2Job= useCases2(param)
            .onEach { resource->
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = resource,
                        )
                    }
            }
            .catch {throwable->
                var currentTryCount = 
                var maxTryCount = 
                if ( throwable is BleReTryableError && currentTryCount < maxTryCount ) {
                    currentTryCount++
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Loading("loading message"),
                            currentTryCount = currentTryCount
                        )
                    }
                    throw BleReTryableError(throwable.message ?: "")
                } else {
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Error("error message"))
                    }
                }
            }
            .retry {
                delay(2500)
                it is BleReTryableError
            }
            .launchIn(viewModelScope)
    }

Job3

private var job3Job: Job? = null
    private fun job3(param: Param) {
        job3Job?.cancel()
        job3Job= useCases3(param)
            .onEach { resource->
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = resource,
                        )
                    }
            }
            .catch {throwable->
                var currentTryCount = 
                var maxTryCount = 
                if ( throwable is BleReTryableError && currentTryCount < maxTryCount ) {
                    currentTryCount++
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Loading("loading message"),
                            currentTryCount = currentTryCount
                        )
                    }
                    throw BleReTryableError(throwable.message ?: "")
                } else {
                    myStateFlow.update {
                        it.copy(isConsumed = false,
                            resource = Resource.Error("error message"))
                    }
                }
            }
            .retry {
                delay(2500)
                it is BleReTryableError
            }
            .launchIn(viewModelScope)
    }

Job4、Job5 ....

或者，它足以使一个功能，如下面的一个？

private fun myRun(flow: Flow<Resource<MyType>>) {

}

扩展方法也可以