{"action":"created","comment":{"url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/comments/73119291","id":73119291,"diff_hunk":"@@ -0,0 +1,244 @@\n+/*\n+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>\n+ */\n+package akka.stream.contrib\n+\n+import akka.stream._\n+import akka.stream.scaladsl._\n+import akka.stream.stage._\n+import scala.util.{ Try, Success, Failure }\n+import scala.concurrent.Future\n+import scala.collection.immutable\n+\n+/**\n+ * This object defines a factory methods for retry operaions.\n+ */\n+object Retry {\n+\n+  /**\n+   * Factory for a retry flow. Given a flow which recieves a tuple of \"input\" & \"state\",\n+   * and emits a tuples of a Try of \"output\" & \"state\", and given a function that given a state\n+   * can produce a new tuple of \"input-state\" to pass again through the flow, this function will\n+   * yield a new flow, which wraps the given flow, and retries failed (i.e. elements passed through\n+   * the flow and come out with `Failure[O]` as the first part of the tuple) elements with the new\n+   * \"input-state\" tuple recieved from the `retryWith` function, and outputs a `Try[O]`.\n+   * To not continue retrying (and getting a `Failure[O]` as output), the `retryWith` function\n+   * should return a `None`\n+   *\n+   * IMPORTANT CAVEAT:\n+   * The given flow must not change the number of elements passing through it (i.e. it should output\n+   * exactly one element for every recieved element). Ignoring this, will have an unpredicted result.\n+   *\n+   * @param flow the flow to retry\n+   * @param retryWith if output was failure, we can optionaly recover from it,\n+   *        and retry with a new pair of input & new state we get from this function.\n+   * @tparam I input elements type\n+   * @tparam O output elements type\n+   * @tparam S state to create a new (input,state): (I,S) to retry with\n+   * @tparam M materialized value type\n+   * @return `Graph[FlowShape[(I,S),Try[O]],M]` instance\n+   */\n+  def apply[I, O, S, M](flow: Graph[FlowShape[(I, S), (Try[O], S)], M])(retryWith: S => Option[(I, S)]): Graph[FlowShape[(I, S), (Try[O], S)], M] = {\n+    GraphDSL.create(flow) { implicit b => origFlow =>\n+      import GraphDSL.Implicits._\n+\n+      val retry = b.add(new RetryCoordinator[I, S, O](retryWith))\n+\n+      retry.out2 ~> origFlow ~> retry.in2\n+\n+      FlowShape(retry.in1, retry.out1)\n+    }\n+  }\n+\n+  /**\n+   * Factory for multiple retries flow. similar to the simple retry, but this will allow to\n+   * break down a \"heavy\" element which failed into multiple \"thin\" elements, that may succeed\n+   * indevidually. Since it's very easy to inflate elements in retry cycle, there's also a\n+   * limit parameter, that will limit the amount of generated elements by the `retryWith` function,\n+   * and will fail the stage if that limit is exceeded.\n+   * Passing `Some(Nil)` is valid, and will result in filtering out the failure quietly, without\n+   * emitting a failed `Try` element.\n+   *\n+   * IMPORTANT CAVEAT:\n+   * The given flow must not change the number of elements passing through it (i.e. it should output\n+   * exactly one element for every recieved element). Ignoring this, will have an unpredicted result.\n+   *\n+   * @param limit since every retry can generate more elements,\n+   *        the inner queue can get too big. if the limit is reached,\n+   *        the stage will fail.\n+   * @param flow the flow to retry\n+   * @param retryWith if output was failure, we can optionaly recover from it,\n+   *        and retry with a sequence of input & new state pairs we get from this function.\n+   * @tparam I input elements type\n+   * @tparam O output elements type\n+   * @tparam S state to create a new (input,state): (I,S) to retry with\n+   * @tparam M materialized value type\n+   * @return `Graph[FlowShape[(I,S),Try[O]],M]` instance\n+   */\n+  def concat[I, O, S, M](limit: Long, flow: Graph[FlowShape[(I, S), (Try[O], S)], M])(retryWith: S => Option[immutable.Iterable[(I, S)]]): Graph[FlowShape[(I, S), (Try[O], S)], M] = {\n+    GraphDSL.create(flow) { implicit b => origFlow =>\n+      import GraphDSL.Implicits._\n+\n+      val retry = b.add(new RetryConcatCoordinator[I, S, O](limit, retryWith))\n+\n+      retry.out2 ~> origFlow ~> retry.in2\n+\n+      FlowShape(retry.in1, retry.out1)\n+    }\n+  }\n+\n+  private[akka] class RetryCoordinator[I, S, O](retryWith: S => Option[(I, S)]) extends GraphStage[BidiShape[(I, S), (Try[O], S), (Try[O], S), (I, S)]] {\n+    val in1 = Inlet[(I, S)](\"RertryCoordinator.in\")","path":"src/main/scala/akka/stream/contrib/Retry.scala","position":91,"original_position":91,"commit_id":"2d45e159f514ec5b4fd6925759c55f3343703047","original_commit_id":"2d45e159f514ec5b4fd6925759c55f3343703047","user":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?v=3","gravatar_id":"","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"body":"I recommend using more descriptive names for these ports to so the code reads nicer.","created_at":"2016-08-02T09:02:59Z","updated_at":"2016-08-02T09:02:59Z","html_url":"https://github.com/akka/akka-stream-contrib/pull/26#discussion_r73119291","pull_request_url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26","_links":{"self":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/comments/73119291"},"html":{"href":"https://github.com/akka/akka-stream-contrib/pull/26#discussion_r73119291"},"pull_request":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26"}}},"pull_request":{"url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26","id":74401424,"html_url":"https://github.com/akka/akka-stream-contrib/pull/26","diff_url":"https://github.com/akka/akka-stream-contrib/pull/26.diff","patch_url":"https://github.com/akka/akka-stream-contrib/pull/26.patch","issue_url":"https://api.github.com/repos/akka/akka-stream-contrib/issues/26","number":26,"state":"open","locked":false,"title":"Retry flows API","user":{"login":"hochgi","id":881075,"avatar_url":"https://avatars.githubusercontent.com/u/881075?v=3","gravatar_id":"","url":"https://api.github.com/users/hochgi","html_url":"https://github.com/hochgi","followers_url":"https://api.github.com/users/hochgi/followers","following_url":"https://api.github.com/users/hochgi/following{/other_user}","gists_url":"https://api.github.com/users/hochgi/gists{/gist_id}","starred_url":"https://api.github.com/users/hochgi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hochgi/subscriptions","organizations_url":"https://api.github.com/users/hochgi/orgs","repos_url":"https://api.github.com/users/hochgi/repos","events_url":"https://api.github.com/users/hochgi/events{/privacy}","received_events_url":"https://api.github.com/users/hochgi/received_events","type":"User","site_admin":false},"body":"PR for https://github.com/akka/akka-stream-contrib/issues/25\r\n","created_at":"2016-06-20T07:47:03Z","updated_at":"2016-08-02T09:03:00Z","closed_at":null,"merged_at":null,"merge_commit_sha":"9cfe1070b9a82661fd384ddba4ae7dd9308f44fb","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26/commits","review_comments_url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26/comments","review_comment_url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/comments{/number}","comments_url":"https://api.github.com/repos/akka/akka-stream-contrib/issues/26/comments","statuses_url":"https://api.github.com/repos/akka/akka-stream-contrib/statuses/2d45e159f514ec5b4fd6925759c55f3343703047","head":{"label":"hochgi:25-hochgi-retry","ref":"25-hochgi-retry","sha":"2d45e159f514ec5b4fd6925759c55f3343703047","user":{"login":"hochgi","id":881075,"avatar_url":"https://avatars.githubusercontent.com/u/881075?v=3","gravatar_id":"","url":"https://api.github.com/users/hochgi","html_url":"https://github.com/hochgi","followers_url":"https://api.github.com/users/hochgi/followers","following_url":"https://api.github.com/users/hochgi/following{/other_user}","gists_url":"https://api.github.com/users/hochgi/gists{/gist_id}","starred_url":"https://api.github.com/users/hochgi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hochgi/subscriptions","organizations_url":"https://api.github.com/users/hochgi/orgs","repos_url":"https://api.github.com/users/hochgi/repos","events_url":"https://api.github.com/users/hochgi/events{/privacy}","received_events_url":"https://api.github.com/users/hochgi/received_events","type":"User","site_admin":false},"repo":{"id":60439872,"name":"akka-stream-contrib","full_name":"hochgi/akka-stream-contrib","owner":{"login":"hochgi","id":881075,"avatar_url":"https://avatars.githubusercontent.com/u/881075?v=3","gravatar_id":"","url":"https://api.github.com/users/hochgi","html_url":"https://github.com/hochgi","followers_url":"https://api.github.com/users/hochgi/followers","following_url":"https://api.github.com/users/hochgi/following{/other_user}","gists_url":"https://api.github.com/users/hochgi/gists{/gist_id}","starred_url":"https://api.github.com/users/hochgi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hochgi/subscriptions","organizations_url":"https://api.github.com/users/hochgi/orgs","repos_url":"https://api.github.com/users/hochgi/repos","events_url":"https://api.github.com/users/hochgi/events{/privacy}","received_events_url":"https://api.github.com/users/hochgi/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/hochgi/akka-stream-contrib","description":"Add-ons to Akka Stream","fork":true,"url":"https://api.github.com/repos/hochgi/akka-stream-contrib","forks_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/forks","keys_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/keys{/key_id}","collaborators_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/teams","hooks_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/hooks","issue_events_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/issues/events{/number}","events_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/events","assignees_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/assignees{/user}","branches_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/branches{/branch}","tags_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/tags","blobs_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/git/refs{/sha}","trees_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/git/trees{/sha}","statuses_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/statuses/{sha}","languages_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/languages","stargazers_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/stargazers","contributors_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/contributors","subscribers_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/subscribers","subscription_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/subscription","commits_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/commits{/sha}","git_commits_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/git/commits{/sha}","comments_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/comments{/number}","issue_comment_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/issues/comments{/number}","contents_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/contents/{+path}","compare_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/compare/{base}...{head}","merges_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/merges","archive_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/downloads","issues_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/issues{/number}","pulls_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/pulls{/number}","milestones_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/milestones{/number}","notifications_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/labels{/name}","releases_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/releases{/id}","deployments_url":"https://api.github.com/repos/hochgi/akka-stream-contrib/deployments","created_at":"2016-06-05T02:37:05Z","updated_at":"2016-06-05T02:37:06Z","pushed_at":"2016-08-01T14:28:46Z","git_url":"git://github.com/hochgi/akka-stream-contrib.git","ssh_url":"git@github.com:hochgi/akka-stream-contrib.git","clone_url":"https://github.com/hochgi/akka-stream-contrib.git","svn_url":"https://github.com/hochgi/akka-stream-contrib","homepage":null,"size":54,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"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":"akka:master","ref":"master","sha":"812d9e7f1c52bc141d675b88101a26d7596c626c","user":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"repo":{"id":58454852,"name":"akka-stream-contrib","full_name":"akka/akka-stream-contrib","owner":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/akka/akka-stream-contrib","description":"Add-ons to Akka Stream","fork":false,"url":"https://api.github.com/repos/akka/akka-stream-contrib","forks_url":"https://api.github.com/repos/akka/akka-stream-contrib/forks","keys_url":"https://api.github.com/repos/akka/akka-stream-contrib/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akka/akka-stream-contrib/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akka/akka-stream-contrib/teams","hooks_url":"https://api.github.com/repos/akka/akka-stream-contrib/hooks","issue_events_url":"https://api.github.com/repos/akka/akka-stream-contrib/issues/events{/number}","events_url":"https://api.github.com/repos/akka/akka-stream-contrib/events","assignees_url":"https://api.github.com/repos/akka/akka-stream-contrib/assignees{/user}","branches_url":"https://api.github.com/repos/akka/akka-stream-contrib/branches{/branch}","tags_url":"https://api.github.com/repos/akka/akka-stream-contrib/tags","blobs_url":"https://api.github.com/repos/akka/akka-stream-contrib/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akka/akka-stream-contrib/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akka/akka-stream-contrib/git/refs{/sha}","trees_url":"https://api.github.com/repos/akka/akka-stream-contrib/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akka/akka-stream-contrib/statuses/{sha}","languages_url":"https://api.github.com/repos/akka/akka-stream-contrib/languages","stargazers_url":"https://api.github.com/repos/akka/akka-stream-contrib/stargazers","contributors_url":"https://api.github.com/repos/akka/akka-stream-contrib/contributors","subscribers_url":"https://api.github.com/repos/akka/akka-stream-contrib/subscribers","subscription_url":"https://api.github.com/repos/akka/akka-stream-contrib/subscription","commits_url":"https://api.github.com/repos/akka/akka-stream-contrib/commits{/sha}","git_commits_url":"https://api.github.com/repos/akka/akka-stream-contrib/git/commits{/sha}","comments_url":"https://api.github.com/repos/akka/akka-stream-contrib/comments{/number}","issue_comment_url":"https://api.github.com/repos/akka/akka-stream-contrib/issues/comments{/number}","contents_url":"https://api.github.com/repos/akka/akka-stream-contrib/contents/{+path}","compare_url":"https://api.github.com/repos/akka/akka-stream-contrib/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akka/akka-stream-contrib/merges","archive_url":"https://api.github.com/repos/akka/akka-stream-contrib/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akka/akka-stream-contrib/downloads","issues_url":"https://api.github.com/repos/akka/akka-stream-contrib/issues{/number}","pulls_url":"https://api.github.com/repos/akka/akka-stream-contrib/pulls{/number}","milestones_url":"https://api.github.com/repos/akka/akka-stream-contrib/milestones{/number}","notifications_url":"https://api.github.com/repos/akka/akka-stream-contrib/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akka/akka-stream-contrib/labels{/name}","releases_url":"https://api.github.com/repos/akka/akka-stream-contrib/releases{/id}","deployments_url":"https://api.github.com/repos/akka/akka-stream-contrib/deployments","created_at":"2016-05-10T11:13:49Z","updated_at":"2016-07-13T20:06:17Z","pushed_at":"2016-08-01T14:28:46Z","git_url":"git://github.com/akka/akka-stream-contrib.git","ssh_url":"git@github.com:akka/akka-stream-contrib.git","clone_url":"https://github.com/akka/akka-stream-contrib.git","svn_url":"https://github.com/akka/akka-stream-contrib","homepage":null,"size":61,"stargazers_count":33,"watchers_count":33,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":9,"mirror_url":null,"open_issues_count":19,"forks":9,"open_issues":19,"watchers":33,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26"},"html":{"href":"https://github.com/akka/akka-stream-contrib/pull/26"},"issue":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/issues/26"},"comments":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/issues/26/comments"},"review_comments":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26/comments"},"review_comment":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/pulls/26/commits"},"statuses":{"href":"https://api.github.com/repos/akka/akka-stream-contrib/statuses/2d45e159f514ec5b4fd6925759c55f3343703047"}}}}
{"actor":{"display_login":"drewhk"}}