Is there any way to make coroutine channel follow a specific order rather than first-in first-out during receiving

I have some tasks which along with three step to execute, the first&third steps must run in sequentially, only the second step run in concurrently, giving the kotlin Channel followed first-in first-out order, I can't always make the step3 run in sequentially, does any imlementation can fit my purpose, or another data structure?

below is my demonstrate code:

import android.util.Log
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlin.random.Random

class SortingReceiveChannelTest(
    private val coroutineScope: CoroutineScope
) {
    private val tasks = List(30) { Task(it) }

    private val randomDelay = Random(Int.MAX_VALUE)
    private val step2Channel = Channel<Task>(2)
    private val step3Channel = Channel<Task>()

    fun start() {
        coroutineScope.launch(Dispatchers.IO) {
            for (task in tasks) {
                task.step1()
                step2Channel.send(task)
            }
        }

        coroutineScope.launch(Dispatchers.IO) {
            repeat(tasks.size) {
                val task = step2Channel.receive()
                coroutineScope.launch(Dispatchers.IO) {
                    task.step2()
                    step3Channel.send(task)
                }
            }
        }

        coroutineScope.launch(Dispatchers.IO) {
            repeat(tasks.size) {
                val task = step3Channel.receive()
                task.step3()
            }
        }
    }

    inner class Task(val index: Int) {
        fun step1() {
            Log.i(TAG, "execute step1[$index] sequential")
            // sleep to simulating a long run task
            Thread.sleep(randomDelay.nextLong(100, 200))
//            Log.i(TAG, "step1[$index] executed")
        }

        fun step2() {
            Log.i(TAG, "execute step2[$index] concurrently")
            if (index == 2 || index == 5 || index == 8 || index == 16) {
                // make specific tasks run longer to mess the step3's order
                Thread.sleep(randomDelay.nextLong(2000, 5000))
            } else {
                Thread.sleep(randomDelay.nextLong(100, 300))
            }
//            Log.i(TAG, "step2[$index] executed")
        }

        fun step3() {
            Log.i(TAG, "execute step3[$index] sequential")
            Thread.sleep(randomDelay.nextLong(100, 300))
//            Log.i(TAG, "step3[$index] executed")
        }
    }

    companion object {
        private const val TAG = "SortChannel"
    }
}

If I understand correctly:

• Steps 1, 2, and 3 must happen sequentially
• Steps 1 and 3 must be processed sequentially for each of the items of the task list in the same order they were in the source list.

One way to do this is to eliminate the intermediary Channel and instead sending both the secondary job and the task together to the final Channel:

class SortingReceiveChannelTest(
private val coroutineScope: CoroutineScope
) {
private val tasks = List(30) { Task(it) }
private val randomDelay = Random(Int.MAX_VALUE)
private val step3Channel = Channel<Pair<Job, Task>>(capacity = tasks.size)
fun start() {
coroutineScope.launch(Dispatchers.IO) {
for (task in tasks) {
task.step1()
val step2Job = coroutineScope.launch { task.step2() }
step3Channel.send(step2Job to task)
}
}
coroutineScope.launch(Dispatchers.IO) {
repeat(tasks.size) {
val (step2Job, task) = step3Channel.receive()
step2Job.join()
task.step3()
}
}
}
//...
}

inspired from Tenfour04's anwser, I tweak a bit, added Semaphore to prevent step1 run too fast as it probably claim many memory to use in product case

import kotlinx.coroutines.sync.Semaphore

class SortingReceiveChannelTest(
    private val coroutineScope: CoroutineScope
) {
    private val tasks = List(30) { Task(it) }

    private val randomDelay = Random(Int.MAX_VALUE)
    private val step3Channel = Channel<Pair<Job, Task>>(capacity = tasks.size)

    fun start() {
        val semaphore = Semaphore(5)
        coroutineScope.launch(Dispatchers.IO) {
            for (task in tasks) {
                semaphore.acquire()
                task.step1()
                step3Channel.send(launch {
                    task.step2()
                    semaphore.release()
                } to task)
            }
        }

        coroutineScope.launch(Dispatchers.IO) {
            repeat(tasks.size) {
                val (step2Job, task) = step3Channel.receive()
                step2Job.join()
                task.step3()
            }
        }
    }

    //...
}