Solution: DiscNewsRepository
You need to wrap the blocking function with withContext, which uses the appropriate dispatcher. In other cases, this dispatcher might have been Dispatchers.IO, but in this case we expect this function to be used intensively. You want to support 200 parallel reads, so you need to use a dispatcher with a custom thread limit. The best choice is to use Dispatchers.IO with the limitedParallelism parameter set to 200.
class DiscNewsRepository(
private val discReader: DiscReader
) : NewsRepository {
private val dispatcher = Dispatchers.IO
.limitedParallelism(200)
override suspend fun getNews(
newsId: String
): News = withContext(dispatcher) {
val (title, content) = discReader.read("user/$newsId")
News(title, content)
}
}
Example solution in playground
import kotlinx.coroutines.*
import org.junit.Test
import kotlin.system.measureTimeMillis
import kotlin.test.assertEquals
class DiscNewsRepository(
private val discReader: DiscReader
) : NewsRepository {
private val dispatcher = Dispatchers.IO
.limitedParallelism(200)
override suspend fun getNews(
newsId: String
): News = withContext(dispatcher) {
val (title, content) = discReader.read("user/$newsId")
News(title, content)
}
}
interface DiscReader {
fun read(key: String): List<String>
}
interface NewsRepository {
suspend fun getNews(newsId: String): News
}
data class News(val title: String, val content: String)
@Suppress("FunctionName")
class DiscNewsRepositoryTests {
@Test
fun `should read data from disc using DiscReader`() = runBlocking {
val repo = DiscNewsRepository(OneSecDiscReader(listOf("Some title", "Some content")))
val res = repo.getNews("SomeUserId")
assertEquals("Some title", res.title)
assertEquals("Some content", res.content)
}
class ImmediateDiscReader(val map: Map<String, List<String>>) : DiscReader {
override fun read(key: String): List<String> = map[key] ?: error("Element not found")
}
@Test
fun `should be prepared for many reads at the same time`() = runBlocking<Unit> {
val repo = DiscNewsRepository(OneSecDiscReader(listOf("Some title", "Some content")))
val time = measureTimeMillis {
coroutineScope {
repeat(10) { id ->
launch {
repo.getNews("SomeUserId$id")
}
}
}
}
assert(time < 2000) { "Should take less than 2000, took $time" }
}
@Test(timeout = 2000)
fun `should be prepared for 200 parallel reads`() = runBlocking<Unit> {
val repo = DiscNewsRepository(OneSecDiscReader(listOf("Some title", "Some content")))
val time = measureTimeMillis {
coroutineScope {
repeat(200) { id ->
launch {
repo.getNews("SomeUserId$id")
}
}
}
}
assert(time < 1900) { "Should take less than 2000, took $time" }
}
@Test(timeout = 3000)
fun `should not allow more than 200 parallel reads`() = runBlocking<Unit> {
val repo = DiscNewsRepository(OneSecDiscReader(listOf("Some title", "Some content")))
val time = measureTimeMillis {
coroutineScope {
repeat(201) { id ->
launch {
repo.getNews("SomeUserId$id")
}
}
}
}
assert(time > 2000) { "Should take longer than 2000, took $time" }
}
class OneSecDiscReader(private val response: List<String>) : DiscReader {
override fun read(key: String): List<String> {
Thread.sleep(1000)
return response
}
}
}
