Added reactive solutions

Author: svtk

src/contributors/Contributors.kt

@@ -17,7 +17,9 @@ enum class Variant {
     CONCURRENT,       // Request5Concurrent
     NOT_CANCELLABLE,  // Request6NotCancellable
     PROGRESS,         // Request6Progress
-    CHANNELS          // Request7Channels
+    CHANNELS,         // Request7Channels
+    RX,               // Request8Rx
+    RX_PROGRESS       // Request9RxProgress
 }
 
 interface Contributors: CoroutineScope {
@@ -108,6 +110,20 @@ interface Contributors: CoroutineScope {
                     }
                 }.setUpCancellation()
             }
+            RX -> {  // Using RxJava
+                loadContributorsReactive(service, req) { users, completed ->
+                    SwingUtilities.invokeLater {
+                        updateResults(users, startTime, completed)
+                    }
+                }
+            }
+            RX_PROGRESS -> {  // Using RxJava and showing progress
+                loadContributorsReactiveProgress(service, req) { users, completed ->
+                    SwingUtilities.invokeLater {
+                        updateResults(users, startTime, completed)
+                    }
+                }
+            }
         }
     }
 

src/contributors/GitHubService.kt

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

src/tasks/Request8Rx.kt

@@ -0,0 +1,36 @@
+package tasks
+
+import contributors.*
+import io.reactivex.Observable
+import io.reactivex.Single
+import io.reactivex.rxkotlin.toObservable
+import io.reactivex.rxkotlin.zipWith
+import io.reactivex.schedulers.Schedulers
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+
+fun loadContributorsReactive(
+    service: GitHubService,
+    req: RequestData,
+    callback: (List<User>, completed: Boolean) -> Unit
+) {
+    val repos: Observable<Repo> = service
+        .getOrgReposRx(req.org)
+        .subscribeOn(Schedulers.io())
+        .doOnNext { response -> logRepos(req, response) }
+        .flatMapIterable { it.bodyList() }
+
+    val allUsers: Single<List<User>> = repos
+        .flatMap { repo ->
+            service.getRepoContributorsRx(req.org, repo.name)
+                .subscribeOn(Schedulers.io())
+                .doOnNext { response -> logUsers(repo, response) }
+        }
+        .flatMapIterable { it.bodyList() }
+        .toList()
+
+    allUsers
+        .doOnSuccess { callback(it.aggregate(), true) }
+        .subscribe()
+}

src/tasks/Request9RxProgress.kt

@@ -0,0 +1,35 @@
+package tasks
+
+import contributors.*
+import io.reactivex.Observable
+import io.reactivex.rxkotlin.toObservable
+import io.reactivex.rxkotlin.zipWith
+import io.reactivex.schedulers.Schedulers
+
+fun loadContributorsReactiveProgress(
+    service: GitHubService,
+    req: RequestData,
+    callback: (List<User>, completed: Boolean) -> Unit
+) {
+    val repos: Observable<Repo> = service
+        .getOrgReposRx(req.org)
+        .subscribeOn(Schedulers.io())
+        .doOnNext { response -> logRepos(req, response) }
+        .flatMapIterable { response -> response.bodyList() }
+
+    val repoUsers: Observable<List<User>> = repos
+        .flatMap { repo ->
+            service.getRepoContributorsRx(req.org, repo.name)
+                .subscribeOn(Schedulers.io())
+                .doOnNext { response -> logUsers(repo, response) }
+                .map { response -> response.bodyList() }
+        }
+    repoUsers
+        .reduce(listOf<User>()) { allUsers, users ->
+            (allUsers + users).aggregate().also {
+                callback(it, false)
+            }
+        }
+        .doOnSuccess { callback(it, true) }
+        .subscribe()
+}

test/contributors/MockGithubService.kt

@@ -1,5 +1,6 @@
 package contributors
 
+import io.reactivex.Observable
 import kotlinx.coroutines.delay
 import retrofit2.Call
 import retrofit2.Response
@@ -24,4 +25,12 @@ object MockGithubService : GitHubService {
         delay(testRepo.delay)
         return Response.success(testRepo.users)
     }
+
+    override fun getOrgReposRx(org: String): Observable<Response<List<Repo>>> {
+        TODO()
+    }
+
+    override fun getRepoContributorsRx(owner: String, repo: String): Observable<Response<List<User>>> {
+        TODO()
+    }
 }