{"action":"created","comment":{"url":"https://api.github.com/repos/lagom/lagom/pulls/comments/132133276","pull_request_review_id":55180046,"id":132133276,"diff_hunk":"@@ -28,44 +29,58 @@ import scala.concurrent.{ ExecutionContext, Future }\n  * Internal API\n  */\n private[cassandra] abstract class CassandraReadSideHandler[Event <: AggregateEvent[Event], Handler](\n-  session: CassandraSession, handlers: Map[Class[_ <: Event], Handler], dispatcher: String\n+  session:    CassandraSession,\n+  handlers:   Map[Class[_ <: Event], Handler],\n+  dispatcher: String\n )(implicit ec: ExecutionContext) extends ReadSideHandler[Event] {\n \n   private val log = LoggerFactory.getLogger(this.getClass)\n \n   protected def invoke(handler: Handler, event: Event, offset: Offset): CompletionStage[JList[BoundStatement]]\n \n   override def handle(): Flow[Pair[Event, Offset], Done, _] = {\n-    akka.stream.scaladsl.Flow[Pair[Event, Offset]].mapAsync(parallelism = 1) { pair =>\n-      handlers.get(pair.first.getClass.asInstanceOf[Class[Event]]) match {\n-        case Some(handler) =>\n-          for {\n-            statements <- invoke(handler, pair.first, pair.second).toScala\n-            done <- statements.size match {\n-              case 0 => Future.successful(Done.getInstance())\n-              case 1 => session.executeWrite(statements.get(0)).toScala\n-              case _ =>\n-                val batch = new BatchStatement\n-                val iter = statements.iterator()\n-                while (iter.hasNext)\n-                  batch.add(iter.next)\n-                session.executeWriteBatch(batch).toScala\n-            }\n-          } yield done\n-        case None =>\n-          if (log.isDebugEnabled)\n-            log.debug(\"Unhandled event [{}]\", pair.first.getClass.getName)\n-          Future.successful(Done.getInstance())\n+\n+    def executeStatements(statements: JList[BoundStatement]): Future[Done] =\n+      statements.size match {\n+        case 0 => Future.successful(Done.getInstance())\n+        case 1 => session.executeWrite(statements.get(0)).toScala\n+        case _ =>\n+          val batch = new BatchStatement\n+          val iter = statements.iterator()\n+          while (iter.hasNext)\n+            batch.add(iter.next)\n+          session.executeWriteBatch(batch).toScala\n       }\n-    }.withAttributes(ActorAttributes.dispatcher(dispatcher)).asJava\n+\n+    akka.stream.scaladsl.Flow[Pair[Event, Offset]]\n+      .mapAsync(parallelism = 1) { pair =>\n+\n+        val handler =\n+          handlers.getOrElse(\n+            // lookup handler\n+            pair.first.getClass.asInstanceOf[Class[Event]],","path":"persistence-cassandra/javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/cassandra/CassandraReadSideHandler.scala","position":63,"original_position":63,"commit_id":"8398c9e0c61dfebd57be27573a42fbafb5dc5713","original_commit_id":"8398c9e0c61dfebd57be27573a42fbafb5dc5713","user":{"login":"rcavalcanti","id":502982,"avatar_url":"https://avatars0.githubusercontent.com/u/502982?v=4","gravatar_id":"","url":"https://api.github.com/users/rcavalcanti","html_url":"https://github.com/rcavalcanti","followers_url":"https://api.github.com/users/rcavalcanti/followers","following_url":"https://api.github.com/users/rcavalcanti/following{/other_user}","gists_url":"https://api.github.com/users/rcavalcanti/gists{/gist_id}","starred_url":"https://api.github.com/users/rcavalcanti/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rcavalcanti/subscriptions","organizations_url":"https://api.github.com/users/rcavalcanti/orgs","repos_url":"https://api.github.com/users/rcavalcanti/repos","events_url":"https://api.github.com/users/rcavalcanti/events{/privacy}","received_events_url":"https://api.github.com/users/rcavalcanti/received_events","type":"User","site_admin":false},"body":"yep, extracting is a great idea","created_at":"2017-08-09T09:28:21Z","updated_at":"2017-08-09T09:28:21Z","html_url":"https://github.com/lagom/lagom/pull/915#discussion_r132133276","pull_request_url":"https://api.github.com/repos/lagom/lagom/pulls/915","_links":{"self":{"href":"https://api.github.com/repos/lagom/lagom/pulls/comments/132133276"},"html":{"href":"https://github.com/lagom/lagom/pull/915#discussion_r132133276"},"pull_request":{"href":"https://api.github.com/repos/lagom/lagom/pulls/915"}}},"pull_request":{"url":"https://api.github.com/repos/lagom/lagom/pulls/915","id":134156866,"html_url":"https://github.com/lagom/lagom/pull/915","diff_url":"https://github.com/lagom/lagom/pull/915.diff","patch_url":"https://github.com/lagom/lagom/pull/915.patch","issue_url":"https://api.github.com/repos/lagom/lagom/issues/915","number":915,"state":"open","locked":false,"title":"Saves offset for unhandled events","user":{"login":"rcavalcanti","id":502982,"avatar_url":"https://avatars0.githubusercontent.com/u/502982?v=4","gravatar_id":"","url":"https://api.github.com/users/rcavalcanti","html_url":"https://github.com/rcavalcanti","followers_url":"https://api.github.com/users/rcavalcanti/followers","following_url":"https://api.github.com/users/rcavalcanti/following{/other_user}","gists_url":"https://api.github.com/users/rcavalcanti/gists{/gist_id}","starred_url":"https://api.github.com/users/rcavalcanti/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rcavalcanti/subscriptions","organizations_url":"https://api.github.com/users/rcavalcanti/orgs","repos_url":"https://api.github.com/users/rcavalcanti/repos","events_url":"https://api.github.com/users/rcavalcanti/events{/privacy}","received_events_url":"https://api.github.com/users/rcavalcanti/received_events","type":"User","site_admin":false},"body":"Fixes #904 ","created_at":"2017-08-04T12:32:12Z","updated_at":"2017-08-09T09:28:21Z","closed_at":null,"merged_at":null,"merge_commit_sha":"c139634a38494aef15a778b54aa2ea815b2a4a0a","assignee":null,"assignees":[],"requested_reviewers":[{"login":"ignasi35","id":762126,"avatar_url":"https://avatars2.githubusercontent.com/u/762126?v=4","gravatar_id":"","url":"https://api.github.com/users/ignasi35","html_url":"https://github.com/ignasi35","followers_url":"https://api.github.com/users/ignasi35/followers","following_url":"https://api.github.com/users/ignasi35/following{/other_user}","gists_url":"https://api.github.com/users/ignasi35/gists{/gist_id}","starred_url":"https://api.github.com/users/ignasi35/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ignasi35/subscriptions","organizations_url":"https://api.github.com/users/ignasi35/orgs","repos_url":"https://api.github.com/users/ignasi35/repos","events_url":"https://api.github.com/users/ignasi35/events{/privacy}","received_events_url":"https://api.github.com/users/ignasi35/received_events","type":"User","site_admin":false}],"milestone":null,"commits_url":"https://api.github.com/repos/lagom/lagom/pulls/915/commits","review_comments_url":"https://api.github.com/repos/lagom/lagom/pulls/915/comments","review_comment_url":"https://api.github.com/repos/lagom/lagom/pulls/comments{/number}","comments_url":"https://api.github.com/repos/lagom/lagom/issues/915/comments","statuses_url":"https://api.github.com/repos/lagom/lagom/statuses/8398c9e0c61dfebd57be27573a42fbafb5dc5713","head":{"label":"rcavalcanti:read-side-offset-persistence","ref":"read-side-offset-persistence","sha":"8398c9e0c61dfebd57be27573a42fbafb5dc5713","user":{"login":"rcavalcanti","id":502982,"avatar_url":"https://avatars0.githubusercontent.com/u/502982?v=4","gravatar_id":"","url":"https://api.github.com/users/rcavalcanti","html_url":"https://github.com/rcavalcanti","followers_url":"https://api.github.com/users/rcavalcanti/followers","following_url":"https://api.github.com/users/rcavalcanti/following{/other_user}","gists_url":"https://api.github.com/users/rcavalcanti/gists{/gist_id}","starred_url":"https://api.github.com/users/rcavalcanti/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rcavalcanti/subscriptions","organizations_url":"https://api.github.com/users/rcavalcanti/orgs","repos_url":"https://api.github.com/users/rcavalcanti/repos","events_url":"https://api.github.com/users/rcavalcanti/events{/privacy}","received_events_url":"https://api.github.com/users/rcavalcanti/received_events","type":"User","site_admin":false},"repo":{"id":54745780,"name":"lagom","full_name":"rcavalcanti/lagom","owner":{"login":"rcavalcanti","id":502982,"avatar_url":"https://avatars0.githubusercontent.com/u/502982?v=4","gravatar_id":"","url":"https://api.github.com/users/rcavalcanti","html_url":"https://github.com/rcavalcanti","followers_url":"https://api.github.com/users/rcavalcanti/followers","following_url":"https://api.github.com/users/rcavalcanti/following{/other_user}","gists_url":"https://api.github.com/users/rcavalcanti/gists{/gist_id}","starred_url":"https://api.github.com/users/rcavalcanti/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rcavalcanti/subscriptions","organizations_url":"https://api.github.com/users/rcavalcanti/orgs","repos_url":"https://api.github.com/users/rcavalcanti/repos","events_url":"https://api.github.com/users/rcavalcanti/events{/privacy}","received_events_url":"https://api.github.com/users/rcavalcanti/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/rcavalcanti/lagom","description":"Reactive Microservices for the JVM","fork":true,"url":"https://api.github.com/repos/rcavalcanti/lagom","forks_url":"https://api.github.com/repos/rcavalcanti/lagom/forks","keys_url":"https://api.github.com/repos/rcavalcanti/lagom/keys{/key_id}","collaborators_url":"https://api.github.com/repos/rcavalcanti/lagom/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/rcavalcanti/lagom/teams","hooks_url":"https://api.github.com/repos/rcavalcanti/lagom/hooks","issue_events_url":"https://api.github.com/repos/rcavalcanti/lagom/issues/events{/number}","events_url":"https://api.github.com/repos/rcavalcanti/lagom/events","assignees_url":"https://api.github.com/repos/rcavalcanti/lagom/assignees{/user}","branches_url":"https://api.github.com/repos/rcavalcanti/lagom/branches{/branch}","tags_url":"https://api.github.com/repos/rcavalcanti/lagom/tags","blobs_url":"https://api.github.com/repos/rcavalcanti/lagom/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/rcavalcanti/lagom/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/rcavalcanti/lagom/git/refs{/sha}","trees_url":"https://api.github.com/repos/rcavalcanti/lagom/git/trees{/sha}","statuses_url":"https://api.github.com/repos/rcavalcanti/lagom/statuses/{sha}","languages_url":"https://api.github.com/repos/rcavalcanti/lagom/languages","stargazers_url":"https://api.github.com/repos/rcavalcanti/lagom/stargazers","contributors_url":"https://api.github.com/repos/rcavalcanti/lagom/contributors","subscribers_url":"https://api.github.com/repos/rcavalcanti/lagom/subscribers","subscription_url":"https://api.github.com/repos/rcavalcanti/lagom/subscription","commits_url":"https://api.github.com/repos/rcavalcanti/lagom/commits{/sha}","git_commits_url":"https://api.github.com/repos/rcavalcanti/lagom/git/commits{/sha}","comments_url":"https://api.github.com/repos/rcavalcanti/lagom/comments{/number}","issue_comment_url":"https://api.github.com/repos/rcavalcanti/lagom/issues/comments{/number}","contents_url":"https://api.github.com/repos/rcavalcanti/lagom/contents/{+path}","compare_url":"https://api.github.com/repos/rcavalcanti/lagom/compare/{base}...{head}","merges_url":"https://api.github.com/repos/rcavalcanti/lagom/merges","archive_url":"https://api.github.com/repos/rcavalcanti/lagom/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/rcavalcanti/lagom/downloads","issues_url":"https://api.github.com/repos/rcavalcanti/lagom/issues{/number}","pulls_url":"https://api.github.com/repos/rcavalcanti/lagom/pulls{/number}","milestones_url":"https://api.github.com/repos/rcavalcanti/lagom/milestones{/number}","notifications_url":"https://api.github.com/repos/rcavalcanti/lagom/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/rcavalcanti/lagom/labels{/name}","releases_url":"https://api.github.com/repos/rcavalcanti/lagom/releases{/id}","deployments_url":"https://api.github.com/repos/rcavalcanti/lagom/deployments","created_at":"2016-03-25T20:29:03Z","updated_at":"2016-03-25T20:29:06Z","pushed_at":"2017-08-08T16:13:16Z","git_url":"git://github.com/rcavalcanti/lagom.git","ssh_url":"git@github.com:rcavalcanti/lagom.git","clone_url":"https://github.com/rcavalcanti/lagom.git","svn_url":"https://github.com/rcavalcanti/lagom","homepage":"https://lightbend.com/lagom","size":6350,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"lagom:master","ref":"master","sha":"6cfa96cede049456b10fa76e20a0fb9d1cda2c65","user":{"login":"lagom","id":17168851,"avatar_url":"https://avatars1.githubusercontent.com/u/17168851?v=4","gravatar_id":"","url":"https://api.github.com/users/lagom","html_url":"https://github.com/lagom","followers_url":"https://api.github.com/users/lagom/followers","following_url":"https://api.github.com/users/lagom/following{/other_user}","gists_url":"https://api.github.com/users/lagom/gists{/gist_id}","starred_url":"https://api.github.com/users/lagom/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lagom/subscriptions","organizations_url":"https://api.github.com/users/lagom/orgs","repos_url":"https://api.github.com/users/lagom/repos","events_url":"https://api.github.com/users/lagom/events{/privacy}","received_events_url":"https://api.github.com/users/lagom/received_events","type":"Organization","site_admin":false},"repo":{"id":52320967,"name":"lagom","full_name":"lagom/lagom","owner":{"login":"lagom","id":17168851,"avatar_url":"https://avatars1.githubusercontent.com/u/17168851?v=4","gravatar_id":"","url":"https://api.github.com/users/lagom","html_url":"https://github.com/lagom","followers_url":"https://api.github.com/users/lagom/followers","following_url":"https://api.github.com/users/lagom/following{/other_user}","gists_url":"https://api.github.com/users/lagom/gists{/gist_id}","starred_url":"https://api.github.com/users/lagom/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lagom/subscriptions","organizations_url":"https://api.github.com/users/lagom/orgs","repos_url":"https://api.github.com/users/lagom/repos","events_url":"https://api.github.com/users/lagom/events{/privacy}","received_events_url":"https://api.github.com/users/lagom/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/lagom/lagom","description":"Reactive Microservices for the JVM","fork":false,"url":"https://api.github.com/repos/lagom/lagom","forks_url":"https://api.github.com/repos/lagom/lagom/forks","keys_url":"https://api.github.com/repos/lagom/lagom/keys{/key_id}","collaborators_url":"https://api.github.com/repos/lagom/lagom/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/lagom/lagom/teams","hooks_url":"https://api.github.com/repos/lagom/lagom/hooks","issue_events_url":"https://api.github.com/repos/lagom/lagom/issues/events{/number}","events_url":"https://api.github.com/repos/lagom/lagom/events","assignees_url":"https://api.github.com/repos/lagom/lagom/assignees{/user}","branches_url":"https://api.github.com/repos/lagom/lagom/branches{/branch}","tags_url":"https://api.github.com/repos/lagom/lagom/tags","blobs_url":"https://api.github.com/repos/lagom/lagom/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/lagom/lagom/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/lagom/lagom/git/refs{/sha}","trees_url":"https://api.github.com/repos/lagom/lagom/git/trees{/sha}","statuses_url":"https://api.github.com/repos/lagom/lagom/statuses/{sha}","languages_url":"https://api.github.com/repos/lagom/lagom/languages","stargazers_url":"https://api.github.com/repos/lagom/lagom/stargazers","contributors_url":"https://api.github.com/repos/lagom/lagom/contributors","subscribers_url":"https://api.github.com/repos/lagom/lagom/subscribers","subscription_url":"https://api.github.com/repos/lagom/lagom/subscription","commits_url":"https://api.github.com/repos/lagom/lagom/commits{/sha}","git_commits_url":"https://api.github.com/repos/lagom/lagom/git/commits{/sha}","comments_url":"https://api.github.com/repos/lagom/lagom/comments{/number}","issue_comment_url":"https://api.github.com/repos/lagom/lagom/issues/comments{/number}","contents_url":"https://api.github.com/repos/lagom/lagom/contents/{+path}","compare_url":"https://api.github.com/repos/lagom/lagom/compare/{base}...{head}","merges_url":"https://api.github.com/repos/lagom/lagom/merges","archive_url":"https://api.github.com/repos/lagom/lagom/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/lagom/lagom/downloads","issues_url":"https://api.github.com/repos/lagom/lagom/issues{/number}","pulls_url":"https://api.github.com/repos/lagom/lagom/pulls{/number}","milestones_url":"https://api.github.com/repos/lagom/lagom/milestones{/number}","notifications_url":"https://api.github.com/repos/lagom/lagom/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/lagom/lagom/labels{/name}","releases_url":"https://api.github.com/repos/lagom/lagom/releases{/id}","deployments_url":"https://api.github.com/repos/lagom/lagom/deployments","created_at":"2016-02-23T01:51:06Z","updated_at":"2017-08-09T06:05:54Z","pushed_at":"2017-08-09T01:16:21Z","git_url":"git://github.com/lagom/lagom.git","ssh_url":"git@github.com:lagom/lagom.git","clone_url":"https://github.com/lagom/lagom.git","svn_url":"https://github.com/lagom/lagom","homepage":"https://www.lagomframework.com","size":6301,"stargazers_count":1461,"watchers_count":1461,"language":"Scala","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":291,"mirror_url":null,"open_issues_count":182,"forks":291,"open_issues":182,"watchers":1461,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/lagom/lagom/pulls/915"},"html":{"href":"https://github.com/lagom/lagom/pull/915"},"issue":{"href":"https://api.github.com/repos/lagom/lagom/issues/915"},"comments":{"href":"https://api.github.com/repos/lagom/lagom/issues/915/comments"},"review_comments":{"href":"https://api.github.com/repos/lagom/lagom/pulls/915/comments"},"review_comment":{"href":"https://api.github.com/repos/lagom/lagom/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/lagom/lagom/pulls/915/commits"},"statuses":{"href":"https://api.github.com/repos/lagom/lagom/statuses/8398c9e0c61dfebd57be27573a42fbafb5dc5713"}}}}
{"actor":{"display_login":"rcavalcanti"}}