More idiomatic user aggregation

src/contributors/GitHubService.kt

@@ -27,6 +27,17 @@ interface GitHubService {
         @Path("owner") owner: String,
         @Path("repo") repo: String
     ): Call<List<User>>
+
+    @GET("orgs/{org}/repos?per_page=100")
+    suspend fun getOrgRepos(
+        @Path("org") org: String
+    ): Response<List<Repo>>
+
+    @GET("repos/{owner}/{repo}/contributors?per_page=100")
+    suspend fun getRepoContributors(
+        @Path("owner") owner: String,
+        @Path("repo") repo: String
+    ): Response<List<User>>
 }
 
 @JsonIgnoreProperties(ignoreUnknown = true)

src/tasks/Aggregation.kt

@@ -15,4 +15,6 @@ TODO: Write aggregation code.
  You can use 'Navigate | Test' menu action (note the shortcut) to navigate to the test.
 */
 fun List<User>.aggregate(): List<User> =
-    this
+    groupBy({ it.login }, { it.contributions })
+        .map { (login, contributionsList) -> User(login, contributionsList.sum()) }
+        .sortedByDescending { it.contributions }

src/tasks/Request2Background.kt

@@ -7,6 +7,6 @@ import kotlin.concurrent.thread
 
 fun loadContributorsBackground(service: GitHubService, req: RequestData, updateResults: (List<User>) -> Unit) {
     thread {
-        loadContributorsBlocking(service, req)
+        updateResults(loadContributorsBlocking(service, req))
     }
 }

src/tasks/Request3Callbacks.kt

@@ -11,16 +11,18 @@ fun loadContributorsCallbacks(service: GitHubService, req: RequestData, updateRe
     service.getOrgReposCall(req.org).onResponse { responseRepos ->
         logRepos(req, responseRepos)
         val repos = responseRepos.bodyList()
-        val allUsers = mutableListOf<User>()
+        val allUsers = Collections.synchronizedList(mutableListOf<User>())
+        val numberOfProcessed = AtomicInteger(0)
         for (repo in repos) {
             service.getRepoContributorsCall(req.org, repo.name).onResponse { responseUsers ->
                 logUsers(repo, responseUsers)
                 val users = responseUsers.bodyList()
                 allUsers += users
+                if (numberOfProcessed.incrementAndGet() == repos.size) {
+                    updateResults(allUsers.aggregate())
+                }
             }
         }
-        // TODO: Why this code doesn't work? How to fix that?
-        updateResults(allUsers.aggregate())
     }
 }
 

src/tasks/Request4Suspend.kt

@@ -3,5 +3,14 @@ package tasks
 import contributors.*
 
 suspend fun loadContributorsSuspend(service: GitHubService, req: RequestData): List<User> {
-    TODO()
+    val repos = service
+        .getOrgRepos(req.org)
+        .also { logRepos(req, it) }
+        .bodyList()
+
+    return repos.flatMap { repo ->
+        service.getRepoContributors(req.org, repo.name)
+            .also { logUsers(repo, it) }
+            .bodyList()
+    }.aggregate()
 }

src/tasks/Request5Concurrent.kt

@@ -4,5 +4,17 @@ import contributors.*
 import kotlinx.coroutines.*
 
 suspend fun loadContributorsConcurrent(service: GitHubService, req: RequestData): List<User> = coroutineScope {
-    TODO()
+    val repos = service
+        .getOrgRepos(req.org)
+        .also { logRepos(req, it) }
+        .bodyList()
+
+    val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
+        async {
+            service.getRepoContributors(req.org, repo.name)
+                .also { logUsers(repo, it) }
+                .bodyList()
+        }
+    }
+    deferreds.awaitAll().flatten().aggregate()
 }

src/tasks/Request5NotCancellable.kt

@@ -5,5 +5,19 @@ import kotlinx.coroutines.*
 import kotlin.coroutines.coroutineContext
 
 suspend fun loadContributorsNotCancellable(service: GitHubService, req: RequestData): List<User> {
-    TODO()
+    val repos = service
+        .getOrgRepos(req.org)
+        .also { logRepos(req, it) }
+        .bodyList()
+
+    val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
+        GlobalScope.async(Dispatchers.Default) {
+            log("starting loading for ${repo.name}")
+            delay(3000)
+            service.getRepoContributors(req.org, repo.name)
+                .also { logUsers(repo, it) }
+                .bodyList()
+        }
+    }
+    return deferreds.awaitAll().flatten().aggregate()
 }

src/tasks/Request6Progress.kt

@@ -7,5 +7,18 @@ suspend fun loadContributorsProgress(
     req: RequestData,
     updateResults: suspend (List<User>, completed: Boolean) -> Unit
 ) {
-    TODO()
+    val repos = service
+        .getOrgRepos(req.org)
+        .also { logRepos(req, it) }
+        .bodyList()
+
+    var allUsers = emptyList<User>()
+    for ((index, repo) in repos.withIndex()) {
+        val users = service.getRepoContributors(req.org, repo.name)
+            .also { logUsers(repo, it) }
+            .bodyList()
+
+        allUsers = (allUsers + users).aggregate()
+        updateResults(allUsers, index == repos.lastIndex)
+    }
 }

src/tasks/Request7Channels.kt

@@ -9,8 +9,25 @@ suspend fun loadContributorsChannels(
     service: GitHubService,
     req: RequestData,
     updateResults: suspend (List<User>, completed: Boolean) -> Unit
-) {
-    coroutineScope {
-        TODO()
+) = coroutineScope {
+    val repos = service
+        .getOrgRepos(req.org)
+        .also { logRepos(req, it) }
+        .bodyList()
+
+    val channel = Channel<List<User>>()
+    for (repo in repos) {
+        launch {
+            val users = service.getRepoContributors(req.org, repo.name)
+                .also { logUsers(repo, it) }
+                .bodyList()
+            channel.send(users)
+        }
+    }
+    var allUsers = emptyList<User>()
+    repeat(repos.size) {
+        val users = channel.receive()
+        allUsers = (allUsers + users).aggregate()
+        updateResults(allUsers, it == repos.lastIndex)
     }
 }

test/contributors/MockGithubService.kt

@@ -14,9 +14,6 @@ object MockGithubService : GitHubService {
         return Calls.response(reposMap.getValue(repo).users)
     }
 
-/*
-    // Uncomment the following implementations after adding these methods to GitHubService:
-
     override suspend fun getOrgRepos(org: String): Response<List<Repo>> {
         delay(reposDelay)
         return Response.success(repos)
@@ -27,5 +24,4 @@ object MockGithubService : GitHubService {
         delay(testRepo.delay)
         return Response.success(testRepo.users)
     }
-*/
 }

test/tasks/Request4SuspendKtTest.kt

@@ -3,29 +3,23 @@ package tasks
 import contributors.MockGithubService
 import contributors.expectedResults
 import contributors.testRequestData
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
 import org.junit.Test
 
+@UseExperimental(ExperimentalCoroutinesApi::class)
 class Request4SuspendKtTest {
     @Test
-    fun testSuspend() = runBlocking {
-        val startTime = System.currentTimeMillis()
+    fun testSuspend() = runBlockingTest {
+        val startTime = currentTime
         val result = loadContributorsSuspend(MockGithubService, testRequestData)
         Assert.assertEquals("Wrong result for 'loadContributorsSuspend'", expectedResults.users, result)
-        val totalTime = System.currentTimeMillis() - startTime
-        /*
-        // TODO: uncomment this assertion
+        val totalTime = currentTime - startTime
         Assert.assertEquals(
-            "The calls run consequently, so the total virtual time should be 4000 ms: " +
-                    "1000 for repos request plus (1000 + 1200 + 800) = 3000 for sequential contributors requests)",
+            "The calls run consequently," +
+                    "so the total virtual time should be 4000 ms: ",
             expectedResults.timeFromStart, totalTime
         )
-        */
-        Assert.assertTrue(
-            "The calls run consequently, so the total time should be around 4000 ms: " +
-                    "1000 for repos request plus (1000 + 1200 + 800) = 3000 for sequential contributors requests)",
-            totalTime in expectedResults.timeFromStart..(expectedResults.timeFromStart + 500)
-        )
     }
 }

test/tasks/Request5ConcurrentKtTest.kt

@@ -3,29 +3,23 @@ package tasks
 import contributors.MockGithubService
 import contributors.expectedConcurrentResults
 import contributors.testRequestData
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
 import org.junit.Test
 
+@UseExperimental(ExperimentalCoroutinesApi::class)
 class Request5ConcurrentKtTest {
     @Test
-    fun testConcurrent() = runBlocking {
-        val startTime = System.currentTimeMillis()
+    fun testConcurrent() = runBlockingTest {
+        val startTime = currentTime
         val result = loadContributorsConcurrent(MockGithubService, testRequestData)
         Assert.assertEquals("Wrong result for 'loadContributorsConcurrent'", expectedConcurrentResults.users, result)
-        val totalTime = System.currentTimeMillis() - startTime
-        /*
-        // TODO: uncomment this assertion
+        val totalTime = currentTime - startTime
         Assert.assertEquals(
             "The calls run concurrently, so the total virtual time should be 2200 ms: " +
                     "1000 ms for repos request plus max(1000, 1200, 800) = 1200 ms for concurrent contributors requests)",
             expectedConcurrentResults.timeFromStart, totalTime
         )
-        */
-        Assert.assertTrue(
-            "The calls run concurrently, so the total virtual time should be 2200 ms: " +
-                    "1000 ms for repos request plus max(1000, 1200, 800) = 1200 ms for concurrent contributors requests)",
-            totalTime in expectedConcurrentResults.timeFromStart..(expectedConcurrentResults.timeFromStart + 500)
-        )
     }
 }

test/tasks/Request6ProgressKtTest.kt

@@ -3,24 +3,23 @@ package tasks
 import contributors.MockGithubService
 import contributors.progressResults
 import contributors.testRequestData
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
 import org.junit.Test
 
+@UseExperimental(ExperimentalCoroutinesApi::class)
 class Request6ProgressKtTest {
     @Test
-    fun testProgress() = runBlocking {
-        val startTime = System.currentTimeMillis()
+    fun testProgress() = runBlockingTest {
+        val startTime = currentTime
         var index = 0
         loadContributorsProgress(MockGithubService, testRequestData) {
             users, _ ->
             val expected = progressResults[index++]
-            val time = System.currentTimeMillis() - startTime
-            /*
-            // TODO: uncomment this assertion
+            val time = currentTime - startTime
             Assert.assertEquals("Expected intermediate result after virtual ${expected.timeFromStart} ms:",
                 expected.timeFromStart, time)
-            */
             Assert.assertEquals("Wrong intermediate result after $time:", expected.users, users)
         }
     }

test/tasks/Request7ChannelsKtTest.kt

@@ -3,24 +3,23 @@ package tasks
 import contributors.MockGithubService
 import contributors.concurrentProgressResults
 import contributors.testRequestData
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
 import org.junit.Assert
 import org.junit.Test
 
+@UseExperimental(ExperimentalCoroutinesApi::class)
 class Request7ChannelsKtTest {
     @Test
-    fun testChannels() = runBlocking {
-        val startTime = System.currentTimeMillis()
+    fun testChannels() = runBlockingTest {
+        val startTime = currentTime
         var index = 0
         loadContributorsChannels(MockGithubService, testRequestData) {
                 users, _ ->
             val expected = concurrentProgressResults[index++]
-            val time = System.currentTimeMillis() - startTime
-            /*
-            // TODO: uncomment this assertion
+            val time = currentTime - startTime
             Assert.assertEquals("Expected intermediate result after virtual ${expected.timeFromStart} ms:",
-                expected.timeFromStart, virtualTime)
-            */
+                expected.timeFromStart, time)
             Assert.assertEquals("Wrong intermediate result after $time:", expected.users, users)
         }
     }