Select Expression

/**
 * Waits for the result of multiple suspend operations simultaneously
 * @param builder block that configures the select clauses
 * @return result from the first clause that completes
 */
suspend fun <R> select(builder: SelectBuilder<R>.() -> Unit): R

/**
 * Unbiased version of select (fair selection)
 */
suspend fun <R> selectUnbiased(builder: SelectBuilder<R>.() -> Unit): R

/**
 * Builder for select expressions
 */
interface SelectBuilder<in R> {
    /** Configure a select clause with no parameters */
    fun SelectClause0.invoke(block: suspend () -> R)
    
    /** Configure a select clause with one parameter */
    fun <P> SelectClause1<P>.invoke(block: suspend (P) -> R)
    
    /** Configure a select clause with two parameters */
    fun <P, Q> SelectClause2<P, Q>.invoke(param: P, block: suspend (Q) -> R)
    
    /** Add timeout clause */
    fun onTimeout(timeMillis: Long, block: suspend () -> R)
}

/**
 * Select clause with no parameters
 */
interface SelectClause0

/**
 * Select clause with one parameter
 */
interface SelectClause1<out T>

/**
 * Select clause with two parameters
 */
interface SelectClause2<in P, out T>

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.*

suspend fun channelSelect() {
    val channel1 = Channel<String>()
    val channel2 = Channel<Int>()
    
    // Launch producers
    launch {
        delay(100)
        channel1.send("Hello")
    }
    
    launch {
        delay(200)
        channel2.send(42)
    }
    
    // Select between channels
    val result = select<String> {
        channel1.onReceive { value ->
            "String: $value"
        }
        channel2.onReceive { value ->
            "Number: $value"
        }
        onTimeout(500) {
            "Timeout occurred"
        }
    }
    
    println(result) // "String: Hello"
}

// Fan-in pattern with select
suspend fun fanIn(channels: List<ReceiveChannel<String>>): ReceiveChannel<String> {
    return produce {
        while (true) {
            val value = select<String?> {
                channels.forEach { channel ->
                    channel.onReceiveCatching { result ->
                        result.getOrNull()
                    }
                }
            }
            if (value != null) {
                send(value)
            } else {
                break // All channels closed
            }
        }
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.selects.*

suspend fun jobSelect() {
    val job1 = async { 
        delay(100)
        "Fast result"
    }
    
    val job2 = async {
        delay(200) 
        "Slow result"
    }
    
    // Wait for first job to complete
    val result = select<String> {
        job1.onAwait { "Job1: $it" }
        job2.onAwait { "Job2: $it" }
    }
    
    println(result) // "Job1: Fast result"
    
    // Cancel remaining job
    job2.cancel()
}

// Racing multiple async operations
suspend fun raceOperations() {
    val operations = listOf(
        async { fetchFromServer1() },
        async { fetchFromServer2() },
        async { fetchFromCache() }
    )
    
    val winner = select<String> {
        operations.forEachIndexed { index, deferred ->
            deferred.onAwait { result ->
                "Server $index: $result"
            }
        }
        onTimeout(1000) {
            "All operations timed out"
        }
    }
    
    // Cancel remaining operations
    operations.forEach { it.cancel() }
    
    return winner
}

class LoadBalancer {
    private val workers = List(3) { Channel<Work>() }
    
    suspend fun submitWork(work: Work) {
        // Send to first available worker
        select<Unit> {
            workers.forEach { worker ->
                worker.onSend(work) {
                    println("Work sent to worker")
                }
            }
        }
    }
    
    fun startWorkers() {
        workers.forEachIndexed { index, channel ->
            launch {
                for (work in channel) {
                    processWork(work, "Worker-$index")
                }
            }
        }
    }
}

suspend fun fetchWithFallback(): String {
    return select {
        // Primary operation
        async { fetchFromPrimarySource() }.onAwait { 
            "Primary: $it" 
        }
        
        // Timeout with fallback
        onTimeout(1000) {
            // Try fallback source
            select<String> {
                async { fetchFromFallbackSource() }.onAwait {
                    "Fallback: $it"
                }
                onTimeout(500) {
                    "Default value"
                }
            }
        }
    }
}

class ProducerConsumerSystem {
    private val highPriority = Channel<Task>(capacity = 10)
    private val lowPriority = Channel<Task>(capacity = 100)
    
    suspend fun processTask(): Task? {
        return select {
            // Prefer high priority tasks
            highPriority.onReceiveCatching { result ->
                result.getOrNull()
            }
            
            // Fall back to low priority if high priority is empty
            lowPriority.onReceiveCatching { result ->
                if (highPriority.isEmpty) {
                    result.getOrNull()
                } else {
                    null // Skip low priority if high priority has tasks
                }
            }
        }
    }
}

// Select - first wins, others cancelled
val result = select {
    async { operation1() }.onAwait { "Op1: $it" }
    async { operation2() }.onAwait { "Op2: $it" }
}

// async/await - wait for all
val results = listOf(
    async { operation1() },
    async { operation2() }
).awaitAll()

// Blocking receive - waits for specific channel
val value = channel.receive()

// Select receive - first available channel
val value = select {
    channel1.onReceive { "From 1: $it" }
    channel2.onReceive { "From 2: $it" }
}

suspend fun selectWithCleanup() {
    val resource1 = acquireResource()
    val resource2 = acquireResource()
    
    try {
        val result = select {
            resource1.onComplete { "Result 1: $it" }
            resource2.onComplete { "Result 2: $it" }
        }
        return result
    } finally {
        resource1.cleanup()
        resource2.cleanup()
    }
}

// Inefficient
while (true) {
    select {
        channel.onReceive { process(it) }
        onTimeout(10) { /* check condition */ }
    }
}

// Better
for (value in channel) {
    process(value)
    if (shouldStop()) break
}

// Biased - earlier clauses preferred
select {
    channel1.onReceive { /* ... */ }
    channel2.onReceive { /* ... */ }
}

// Unbiased - fair selection
selectUnbiased {
    channel1.onReceive { /* ... */ }
    channel2.onReceive { /* ... */ }
}