feat: error handling for common crawler (APP-106)

src/main/kotlin/app/Logger.kt

@@ -44,7 +44,7 @@ object Logger {
     /**
      * Log error message with exception info.
      */
-    fun error(message: String, e: Exception) {
+    fun error(message: String, e: Throwable) {
         if (LEVEL >= ERROR) {
             println("[e] $message: $e")
         }

src/main/kotlin/app/hashers/CommitHasher.kt

@@ -61,12 +61,11 @@ class CommitHasher(private val localRepo: LocalRepo,
     }
 
     // Hash added and missing server commits and send them to server.
-    fun updateFromObservable(observable: Observable<Commit>) {
+    fun updateFromObservable(observable: Observable<Commit>,
+                             onError: (Throwable) -> Unit) {
         val lastKnownCommit = serverRepo.commits.lastOrNull()
         val knownCommits = serverRepo.commits.toHashSet()
 
-        val throwables = mutableListOf<Throwable>()
-
         observable
             .takeWhile { new ->  // Hash until last known commit.
                 new.rehash != lastKnownCommit?.rehash
@@ -94,9 +93,6 @@ class CommitHasher(private val localRepo: LocalRepo,
             .buffer(20, TimeUnit.SECONDS)  // Group ready commits by time.
             .subscribe({ commitsBundle ->  // OnNext.
                 postCommitsToServer(commitsBundle)  // Send ready commits.
-            }, { e ->  // OnError.
-                Logger.error("Hashing error: " + e.message)
-                throwables.add(e)  // TODO(anatoly): Top-class handling errors.
-            })
+            }, onError)
     }
 }

src/main/kotlin/app/hashers/FactHasher.kt

@@ -29,12 +29,11 @@ class FactHasher(private val localRepo: LocalRepo,
         }
     }
 
-    fun updateFromObservable(observable: Observable<Commit>) {
+    fun updateFromObservable(observable: Observable<Commit>,
+                             onError: (Throwable) -> Unit) {
         val factsDayWeek = hashMapOf<Author, Array<Int>>()
         val factsDayTime = hashMapOf<Author, Array<Int>>()
 
-        val throwables = mutableListOf<Throwable>()
-
         // TODO(anatoly): Filter hashing by email as in CommitHasher.
         observable
             .subscribe({ commit ->  // OnNext.
@@ -51,9 +50,7 @@ class FactHasher(private val localRepo: LocalRepo,
                 factDayTime[dateTime.hour] += 1
                 factsDayWeek[author] = factDayWeek
                 factsDayTime[author] = factDayTime
-            }, { e ->  // OnError.
-                throwables.add(e)  // TODO(anatoly): Top-class handling errors.
-            }, {  // OnComplete.
+            }, onError, {  // OnComplete.
                 try {
                     val facts = mutableListOf<Fact>()
                     factsDayTime.map { (author, list) ->
@@ -76,7 +73,7 @@ class FactHasher(private val localRepo: LocalRepo,
                     }
                     postFactsToServer(facts)
                 } catch (e: Throwable) {
-                    throwables.add(e)
+                    onError(e)
                 }
             })
     }

src/main/kotlin/app/hashers/RepoHasher.kt

@@ -6,9 +6,9 @@ package app.hashers
 import app.Logger
 import app.api.Api
 import app.config.Configurator
-import app.model.Author
 import app.model.LocalRepo
 import app.model.Repo
+import app.utils.HashingException
 import app.utils.RepoHelper
 import org.apache.commons.codec.digest.DigestUtils
 import org.eclipse.jgit.api.Git
@@ -26,7 +26,9 @@ class RepoHasher(private val localRepo: LocalRepo, private val api: Api,
         if (!RepoHelper.isValidRepo(localRepo.path)) {
             throw IllegalArgumentException("Invalid repo $localRepo")
         }
+    }
 
+    fun update() {
         println("Hashing $localRepo...")
         val git = loadGit(localRepo.path)
         try {
@@ -46,26 +48,44 @@ class RepoHasher(private val localRepo: LocalRepo, private val api: Api,
             // Get repo setup (commits, emails to hash) from server.
             getRepoFromServer()
 
+            // Common error handling for subscribers.
+            // Exceptions can't be thrown out of reactive chain.
+            val errors = mutableListOf<Throwable>()
+            val onError: (Throwable) -> Unit = {
+                e -> errors.add(e)
+                Logger.error("Error while hashing:", e)
+            }
+
             // Hash by all plugins.
             val observable = CommitCrawler.getObservable(git, serverRepo)
                                              .publish()
             CommitHasher(localRepo, serverRepo, api, rehashes)
-                .updateFromObservable(observable)
+                .updateFromObservable(observable, onError)
             FactHasher(localRepo, serverRepo, api)
-                .updateFromObservable(observable)
+                .updateFromObservable(observable, onError)
             // Start and synchronously wait until all subscribers complete.
             observable.connect()
 
             // TODO(anatoly): CodeLongevity hash from observable.
-            CodeLongevity(localRepo, serverRepo, api, git).update()
+            try {
+                CodeLongevity(localRepo, serverRepo, api, git).update()
+            }
+            catch (e: Throwable) {
+                onError(e)
+            }
 
             // Confirm hashing completion.
             postRepoToServer()
+
+            if (errors.isNotEmpty()) {
+                throw HashingException(errors)
+            }
+
+            println("Hashing $localRepo successfully finished.")
         }
         finally {
             closeGit(git)
         }
-        println("Hashing $localRepo successfully finished.")
     }
 
     private fun loadGit(path: String): Git {
@@ -104,29 +124,25 @@ class RepoHasher(private val localRepo: LocalRepo, private val api: Api,
     }
 
     private fun fetchRehashesAndAuthors(git: Git):
-        Pair<LinkedList<String>, HashMap<String, Author>> {
+        Pair<LinkedList<String>, HashSet<String>> {
         val head: RevCommit = RevWalk(git.repository)
             .parseCommit(git.repository.resolve(RepoHelper.MASTER_BRANCH))
 
         val revWalk = RevWalk(git.repository)
         revWalk.markStart(head)
 
         val commitsRehashes = LinkedList<String>()
-        val contributors = hashMapOf<String, Author>()
+        val authors = hashSetOf<String>()
 
         var commit: RevCommit? = revWalk.next()
         while (commit != null) {
             commitsRehashes.add(DigestUtils.sha256Hex(commit.name))
-            if (!contributors.containsKey(commit.authorIdent.emailAddress)) {
-                val author = Author(commit.authorIdent.name,
-                                    commit.authorIdent.emailAddress)
-                contributors.put(commit.authorIdent.emailAddress, author)
-            }
+            authors.add(commit.authorIdent.emailAddress)
             commit.disposeBody()
             commit = revWalk.next()
         }
         revWalk.dispose()
 
-        return Pair(commitsRehashes, contributors)
+        return Pair(commitsRehashes, authors)
     }
 }

src/main/kotlin/app/ui/UpdateRepoState.kt

@@ -7,6 +7,7 @@ import app.hashers.RepoHasher
 import app.Logger
 import app.api.Api
 import app.config.Configurator
+import app.utils.HashingException
 import app.utils.RequestException
 
 /**
@@ -20,9 +21,12 @@ class UpdateRepoState constructor(private val context: Context,
         println("Hashing your git repositories.")
         for (repo in configurator.getLocalRepos()) {
             try {
-                RepoHasher(repo, api, configurator)
-            } catch (e: RequestException) {
-                Logger.error("Network error while hashing $repo", e)
+                RepoHasher(repo, api, configurator).update()
+            } catch (e: HashingException) {
+                Logger.error("During hashing ${e.errors.size} errors occurred:")
+                e.errors.forEach { error ->
+                    Logger.error("", error)
+                }
             } catch (e: Exception) {
                 Logger.error("Error while hashing $repo", e)
             }

src/main/kotlin/app/utils/HashingException.kt

@@ -0,0 +1,6 @@
+// Copyright 2017 Sourcerer Inc. All Rights Reserved.
+// Author: Anatoly Kislov (anatoly@sourcerer.io)
+
+package app.utils
+
+class HashingException(val errors: List<Throwable>) : Exception()

src/test/kotlin/test/tests/hashers/CommitHasherTest.kt

@@ -63,10 +63,15 @@ class CommitHasherTest : Spek({
     given("repo with initial commit and no history") {
         repo.commits = listOf()
 
+        val errors = mutableListOf<Throwable>()
         val mockApi = MockApi(mockRepo = repo)
         val observable = CommitCrawler.getObservable(gitHasher, repo)
         CommitHasher(localRepo, repo, mockApi, repo.commits.map {it.rehash})
-            .updateFromObservable(observable)
+            .updateFromObservable(observable, { e -> errors.add(e) })
+
+        it ("has no errors") {
+            assertEquals(0, errors.size)
+        }
 
         it("send added commits") {
             assertEquals(1, mockApi.receivedAddedCommits.size)
@@ -80,10 +85,15 @@ class CommitHasherTest : Spek({
     given("repo with initial commit") {
         repo.commits = listOf(getLastCommit(git))
 
+        val errors = mutableListOf<Throwable>()
         val mockApi = MockApi(mockRepo = repo)
         val observable = CommitCrawler.getObservable(gitHasher, repo)
         CommitHasher(localRepo, repo, mockApi, repo.commits.map {it.rehash})
-            .updateFromObservable(observable)
+            .updateFromObservable(observable, { e -> errors.add(e) })
+
+        it ("has no errors") {
+            assertEquals(0, errors.size)
+        }
 
         it("doesn't send added commits") {
             assertEquals(0, mockApi.receivedAddedCommits.size)
@@ -97,13 +107,17 @@ class CommitHasherTest : Spek({
     given("happy path: added one commit") {
         repo.commits = listOf(getLastCommit(git))
 
+        val errors = mutableListOf<Throwable>()
         val mockApi = MockApi(mockRepo = repo)
-
         val revCommit = git.commit().setMessage("Second commit.").call()
         val addedCommit = Commit(revCommit)
         val observable = CommitCrawler.getObservable(gitHasher, repo)
         CommitHasher(localRepo, repo, mockApi, repo.commits.map {it.rehash})
-            .updateFromObservable(observable)
+            .updateFromObservable(observable, { e -> errors.add(e) })
+
+        it ("has no errors") {
+            assertEquals(0, errors.size)
+        }
 
         it("doesn't send deleted commits") {
             assertEquals(0, mockApi.receivedDeletedCommits.size)
@@ -121,6 +135,7 @@ class CommitHasherTest : Spek({
     given("happy path: added a few new commits") {
         repo.commits = listOf(getLastCommit(git))
 
+        val errors = mutableListOf<Throwable>()
         val mockApi = MockApi(mockRepo = repo)
 
         val otherAuthorsNames = listOf("a", "b", "a")
@@ -139,7 +154,11 @@ class CommitHasherTest : Spek({
         }
         val observable = CommitCrawler.getObservable(gitHasher, repo)
         CommitHasher(localRepo, repo, mockApi, repo.commits.map {it.rehash})
-            .updateFromObservable(observable)
+            .updateFromObservable(observable, { e -> errors.add(e) })
+
+        it ("has no errors") {
+            assertEquals(0, errors.size)
+        }
 
         it("posts five commits as added") {
             assertEquals(5, mockApi.receivedAddedCommits.size)
@@ -185,6 +204,7 @@ class CommitHasherTest : Spek({
     given("lost server") {
         repo.commits = listOf(getLastCommit(git))
 
+        val errors = mutableListOf<Throwable>()
         val mockApi = MockApi(mockRepo = repo)
 
         // Add some commits.
@@ -201,7 +221,11 @@ class CommitHasherTest : Spek({
 
         val observable = CommitCrawler.getObservable(gitHasher, repo)
         CommitHasher(localRepo, repo, mockApi, repo.commits.map {it.rehash})
-            .updateFromObservable(observable)
+            .updateFromObservable(observable, { e -> errors.add(e) })
+
+        it ("has no errors") {
+            assertEquals(0, errors.size)
+        }
 
         it("adds posts one commit as added and received commit is lost one") {
             assertEquals(1, mockApi.receivedAddedCommits.size)

src/test/kotlin/test/tests/hashers/FactHasherTest.kt

@@ -55,10 +55,12 @@ class FactHasherTest : Spek({
                 date = createDate(year = 2017, month = 1, day = 1,  // Sunday.
                     hour = 13, minute = 0, seconds = 0))
 
+            val errors = mutableListOf<Throwable>()
             val observable = CommitCrawler.getObservable(testRepo.git, repo)
             FactHasher(localRepo, repo, mockApi)
-                .updateFromObservable(observable)
+                .updateFromObservable(observable, { e -> errors.add(e) })
 
+            assertEquals(0, errors.size)
             assertEquals(2, facts.size)
             assertTrue(facts.contains(Fact(repo, FactCodes.COMMITS_DAY_TIME, 13,
                 1.0, author1)))
@@ -79,10 +81,12 @@ class FactHasherTest : Spek({
                 date = createDate(year=2017, month = 1, day = 2,  // Monday.
                     hour = 13, minute = 0, seconds = 0))
 
+            val errors = mutableListOf<Throwable>()
             val observable = CommitCrawler.getObservable(testRepo.git, repo)
             FactHasher(localRepo, repo, mockApi)
-                .updateFromObservable(observable)
+                .updateFromObservable(observable, { e -> errors.add(e) })
 
+            assertEquals(0, errors.size)
             assertEquals(5, facts.size)
             assertTrue(facts.contains(Fact(repo, FactCodes.COMMITS_DAY_TIME, 18,
                 1.0, author2)))