{"action":"created","comment":{"url":"https://api.github.com/repos/akka/akka/pulls/comments/183800597","pull_request_review_id":114868303,"id":183800597,"diff_hunk":"@@ -0,0 +1,126 @@\n+/**\n+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>\n+ */\n+\n+package akka.stream.typed.scaladsl\n+\n+import akka.NotUsed\n+import akka.actor.typed.ActorRef\n+import akka.pattern.AskTimeoutException\n+import akka.stream._\n+import akka.stream.scaladsl._\n+import akka.util.Timeout\n+\n+import scala.annotation.implicitNotFound\n+import scala.reflect.ClassTag\n+\n+/**\n+ * Collection of Flows aimed at integrating with typed Actors.\n+ */\n+object ActorFlow {\n+\n+  // TODO would be nice to provide Implicits to allow .ask() directly on Flow/Source\n+\n+  /**\n+   * Use the `ask` pattern to send a request-reply message to the target `ref` actor.\n+   * If any of the asks times out it will fail the stream with a [[java.util.concurrent.TimeoutException]].\n+   *\n+   * Do not forget to include the expected response type in the method call, like so:\n+   *\n+   * {{{\n+   * flow.via(ActorFlow.ask[String, Asking, Reply](ref)((el, replyTo) => Asking(el, replyTo)))\n+   *\n+   * // or even:\n+   * flow.via(ActorFlow.ask[String, Asking, Reply](ref)(Asking(_, _)))\n+   * }}}\n+   *\n+   * otherwise `Nothing` will be assumed, which is most likely not what you want.\n+   *\n+   * Defaults to parallelism of 2 messages in flight, since while one ask message may be being worked on, the second one\n+   * still be in the mailbox, so defaulting to sending the second one a bit earlier than when first ask has replied maintains\n+   * a slightly healthier throughput.\n+   *\n+   * The stage fails with an [[akka.stream.WatchedActorTerminatedException]] if the target actor is terminated,\n+   * or with an [[java.util.concurrent.TimeoutException]] in case the ask exceeds the timeout passed in.\n+   *\n+   * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.\n+   *\n+   * '''Emits when''' the futures (in submission order) created by the ask pattern internally are completed\n+   *\n+   * '''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures\n+   *\n+   * '''Completes when''' upstream completes and all futures have been completed and all elements have been emitted\n+   *\n+   * '''Fails when''' the passed in actor terminates, or a timeout is exceeded in any of the asks performed\n+   *\n+   * '''Cancels when''' downstream cancels\n+   *\n+   * @tparam I Incoming element type of the Flow\n+   * @tparam Q Question message type that is spoken by the target actor\n+   * @tparam O answer type that the Actor is expected to reply with, it will become the Output type of this Flow\n+   */\n+  @implicitNotFound(\"Missing an implicit akka.util.Timeout for the ask() stage\")\n+  def ask[I, Q, O](ref: ActorRef[Q])(makeMessage: (I, ActorRef[O]) ⇒ Q)(implicit timeout: Timeout): Flow[I, O, NotUsed] =\n+    ask(parallelism = 2)(ref)(makeMessage)(timeout)\n+\n+  /**\n+   * Use the `ask` pattern to send a request-reply message to the target `ref` actor.\n+   * If any of the asks times out it will fail the stream with a [[java.util.concurrent.TimeoutException]].\n+   *\n+   * Do not forget to include the expected response type in the method call, like so:\n+   *\n+   * {{{\n+   * flow.via(ActorFlow.ask[String, Asking, Reply](parallelism = 4)(ref, (el, replyTo) => Asking(el, replyTo)))\n+   *\n+   * // or even:\n+   * flow.via(ActorFlow.ask[String, Asking, Reply](parallelism = 4)(ref, Asking(_, _)))\n+   * }}}\n+   *\n+   * otherwise `Nothing` will be assumed, which is most likely not what you want.\n+   *\n+   * The stage fails with an [[akka.stream.WatchedActorTerminatedException]] if the target actor is terminated,\n+   * or with an [[java.util.concurrent.TimeoutException]] in case the ask exceeds the timeout passed in.\n+   *\n+   * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.\n+   *\n+   * '''Emits when''' the futures (in submission order) created by the ask pattern internally are completed\n+   *\n+   * '''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures\n+   *\n+   * '''Completes when''' upstream completes and all futures have been completed and all elements have been emitted\n+   *\n+   * '''Fails when''' the passed in actor terminates, or a timeout is exceeded in any of the asks performed\n+   *\n+   * '''Cancels when''' downstream cancels\n+   *\n+   * @tparam I Incoming element type of the Flow\n+   * @tparam Q Question message type that is spoken by the target actor\n+   * @tparam O answer type that the Actor is expected to reply with, it will become the Output type of this Flow\n+   */\n+  @implicitNotFound(\"Missing an implicit akka.util.Timeout for the ask() stage\")\n+  def ask[I, Q, O](parallelism: Int)(ref: ActorRef[Q])(makeMessage: (I, ActorRef[O]) ⇒ Q)(implicit timeout: Timeout): Flow[I, O, NotUsed] = {\n+    import akka.actor.typed.scaladsl.adapter._\n+    val untypedRef = ref.toUntyped\n+\n+    val askFlow = Flow[I]\n+      .watch(untypedRef)\n+      .mapAsync(parallelism) { el ⇒\n+        akka.pattern.extended.ask(untypedRef, (replyTo: akka.actor.ActorRef) ⇒ makeMessage(el, replyTo))\n+      }\n+      .map(_.asInstanceOf[O])\n+      .mapError {","path":"akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorFlow.scala","position":111,"original_position":111,"commit_id":"cda43759f6978fd06f32dd53fc3daf83ba32a051","original_commit_id":"cda43759f6978fd06f32dd53fc3daf83ba32a051","user":{"login":"ktoso","id":120979,"avatar_url":"https://avatars1.githubusercontent.com/u/120979?v=4","gravatar_id":"","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"body":"done 👍 ","created_at":"2018-04-24T16:35:05Z","updated_at":"2018-04-24T16:35:06Z","html_url":"https://github.com/akka/akka/pull/24892#discussion_r183800597","pull_request_url":"https://api.github.com/repos/akka/akka/pulls/24892","author_association":"OWNER","_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/comments/183800597"},"html":{"href":"https://github.com/akka/akka/pull/24892#discussion_r183800597"},"pull_request":{"href":"https://api.github.com/repos/akka/akka/pulls/24892"}},"in_reply_to_id":183791110},"pull_request":{"url":"https://api.github.com/repos/akka/akka/pulls/24892","id":181231396,"html_url":"https://github.com/akka/akka/pull/24892","diff_url":"https://github.com/akka/akka/pull/24892.diff","patch_url":"https://github.com/akka/akka/pull/24892.patch","issue_url":"https://api.github.com/repos/akka/akka/issues/24892","number":24892,"state":"open","locked":false,"title":"Akka Typed ask() for Akka Streams","user":{"login":"ktoso","id":120979,"avatar_url":"https://avatars1.githubusercontent.com/u/120979?v=4","gravatar_id":"","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"body":"I really wanted to avoid forcing users to provide an implicit scheduler, so it has to be a custom graph stage like this.\r\n\r\nWould we be OK with this or do we really want to avoid the mini-custom impl?\r\nNotice that it does not reimplement the mapAsync stuff but relies on it following it; this changes the asks-in-flight slightly but perhaps could be good enough?\r\n\r\nI needed some form of ask in streams for akka streams and typed for the coursera course, so will likely put this as \"please use this one, assume it works\" hidden so people can use it in the course perhaps.\r\n\r\nThough would be best to decide how to address it here and get it merged of course.\r\nPlease voice opinions on how to implement it.","created_at":"2018-04-12T14:55:54Z","updated_at":"2018-04-24T16:35:06Z","closed_at":null,"merged_at":null,"merge_commit_sha":"cb948622354a2a2a2954cc1d8db25cbf8a73927d","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[{"id":37797347,"url":"https://api.github.com/repos/akka/akka/labels/tested","name":"tested","color":"d7e102","default":false}],"milestone":null,"commits_url":"https://api.github.com/repos/akka/akka/pulls/24892/commits","review_comments_url":"https://api.github.com/repos/akka/akka/pulls/24892/comments","review_comment_url":"https://api.github.com/repos/akka/akka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/akka/akka/issues/24892/comments","statuses_url":"https://api.github.com/repos/akka/akka/statuses/cda43759f6978fd06f32dd53fc3daf83ba32a051","head":{"label":"ktoso:wip-akka-typed-ask-flow","ref":"wip-akka-typed-ask-flow","sha":"cda43759f6978fd06f32dd53fc3daf83ba32a051","user":{"login":"ktoso","id":120979,"avatar_url":"https://avatars1.githubusercontent.com/u/120979?v=4","gravatar_id":"","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"repo":{"id":4462204,"name":"akka","full_name":"ktoso/akka","owner":{"login":"ktoso","id":120979,"avatar_url":"https://avatars1.githubusercontent.com/u/120979?v=4","gravatar_id":"","url":"https://api.github.com/users/ktoso","html_url":"https://github.com/ktoso","followers_url":"https://api.github.com/users/ktoso/followers","following_url":"https://api.github.com/users/ktoso/following{/other_user}","gists_url":"https://api.github.com/users/ktoso/gists{/gist_id}","starred_url":"https://api.github.com/users/ktoso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ktoso/subscriptions","organizations_url":"https://api.github.com/users/ktoso/orgs","repos_url":"https://api.github.com/users/ktoso/repos","events_url":"https://api.github.com/users/ktoso/events{/privacy}","received_events_url":"https://api.github.com/users/ktoso/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/ktoso/akka","description":"Akka Project","fork":true,"url":"https://api.github.com/repos/ktoso/akka","forks_url":"https://api.github.com/repos/ktoso/akka/forks","keys_url":"https://api.github.com/repos/ktoso/akka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ktoso/akka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ktoso/akka/teams","hooks_url":"https://api.github.com/repos/ktoso/akka/hooks","issue_events_url":"https://api.github.com/repos/ktoso/akka/issues/events{/number}","events_url":"https://api.github.com/repos/ktoso/akka/events","assignees_url":"https://api.github.com/repos/ktoso/akka/assignees{/user}","branches_url":"https://api.github.com/repos/ktoso/akka/branches{/branch}","tags_url":"https://api.github.com/repos/ktoso/akka/tags","blobs_url":"https://api.github.com/repos/ktoso/akka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ktoso/akka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ktoso/akka/git/refs{/sha}","trees_url":"https://api.github.com/repos/ktoso/akka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ktoso/akka/statuses/{sha}","languages_url":"https://api.github.com/repos/ktoso/akka/languages","stargazers_url":"https://api.github.com/repos/ktoso/akka/stargazers","contributors_url":"https://api.github.com/repos/ktoso/akka/contributors","subscribers_url":"https://api.github.com/repos/ktoso/akka/subscribers","subscription_url":"https://api.github.com/repos/ktoso/akka/subscription","commits_url":"https://api.github.com/repos/ktoso/akka/commits{/sha}","git_commits_url":"https://api.github.com/repos/ktoso/akka/git/commits{/sha}","comments_url":"https://api.github.com/repos/ktoso/akka/comments{/number}","issue_comment_url":"https://api.github.com/repos/ktoso/akka/issues/comments{/number}","contents_url":"https://api.github.com/repos/ktoso/akka/contents/{+path}","compare_url":"https://api.github.com/repos/ktoso/akka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ktoso/akka/merges","archive_url":"https://api.github.com/repos/ktoso/akka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ktoso/akka/downloads","issues_url":"https://api.github.com/repos/ktoso/akka/issues{/number}","pulls_url":"https://api.github.com/repos/ktoso/akka/pulls{/number}","milestones_url":"https://api.github.com/repos/ktoso/akka/milestones{/number}","notifications_url":"https://api.github.com/repos/ktoso/akka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ktoso/akka/labels{/name}","releases_url":"https://api.github.com/repos/ktoso/akka/releases{/id}","deployments_url":"https://api.github.com/repos/ktoso/akka/deployments","created_at":"2012-05-27T14:05:17Z","updated_at":"2016-10-24T14:59:20Z","pushed_at":"2018-04-24T16:02:36Z","git_url":"git://github.com/ktoso/akka.git","ssh_url":"git@github.com:ktoso/akka.git","clone_url":"https://github.com/ktoso/akka.git","svn_url":"https://github.com/ktoso/akka","homepage":"http://akka.io","size":216616,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":0,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"akka:master","ref":"master","sha":"b6a27deaeaa2efd4a21c4748fc2c7de06402bb21","user":{"login":"akka","id":1496237,"avatar_url":"https://avatars2.githubusercontent.com/u/1496237?v=4","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":130013,"name":"akka","full_name":"akka/akka","owner":{"login":"akka","id":1496237,"avatar_url":"https://avatars2.githubusercontent.com/u/1496237?v=4","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","description":"Build highly concurrent, distributed, and resilient message-driven applications on the JVM","fork":false,"url":"https://api.github.com/repos/akka/akka","forks_url":"https://api.github.com/repos/akka/akka/forks","keys_url":"https://api.github.com/repos/akka/akka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akka/akka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akka/akka/teams","hooks_url":"https://api.github.com/repos/akka/akka/hooks","issue_events_url":"https://api.github.com/repos/akka/akka/issues/events{/number}","events_url":"https://api.github.com/repos/akka/akka/events","assignees_url":"https://api.github.com/repos/akka/akka/assignees{/user}","branches_url":"https://api.github.com/repos/akka/akka/branches{/branch}","tags_url":"https://api.github.com/repos/akka/akka/tags","blobs_url":"https://api.github.com/repos/akka/akka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akka/akka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akka/akka/git/refs{/sha}","trees_url":"https://api.github.com/repos/akka/akka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akka/akka/statuses/{sha}","languages_url":"https://api.github.com/repos/akka/akka/languages","stargazers_url":"https://api.github.com/repos/akka/akka/stargazers","contributors_url":"https://api.github.com/repos/akka/akka/contributors","subscribers_url":"https://api.github.com/repos/akka/akka/subscribers","subscription_url":"https://api.github.com/repos/akka/akka/subscription","commits_url":"https://api.github.com/repos/akka/akka/commits{/sha}","git_commits_url":"https://api.github.com/repos/akka/akka/git/commits{/sha}","comments_url":"https://api.github.com/repos/akka/akka/comments{/number}","issue_comment_url":"https://api.github.com/repos/akka/akka/issues/comments{/number}","contents_url":"https://api.github.com/repos/akka/akka/contents/{+path}","compare_url":"https://api.github.com/repos/akka/akka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akka/akka/merges","archive_url":"https://api.github.com/repos/akka/akka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akka/akka/downloads","issues_url":"https://api.github.com/repos/akka/akka/issues{/number}","pulls_url":"https://api.github.com/repos/akka/akka/pulls{/number}","milestones_url":"https://api.github.com/repos/akka/akka/milestones{/number}","notifications_url":"https://api.github.com/repos/akka/akka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akka/akka/labels{/name}","releases_url":"https://api.github.com/repos/akka/akka/releases{/id}","deployments_url":"https://api.github.com/repos/akka/akka/deployments","created_at":"2009-02-16T12:51:54Z","updated_at":"2018-04-24T16:12:33Z","pushed_at":"2018-04-24T16:02:33Z","git_url":"git://github.com/akka/akka.git","ssh_url":"git@github.com:akka/akka.git","clone_url":"https://github.com/akka/akka.git","svn_url":"https://github.com/akka/akka","homepage":"http://akka.io","size":210622,"stargazers_count":8467,"watchers_count":8467,"language":"Scala","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2694,"mirror_url":null,"archived":false,"open_issues_count":719,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":2694,"open_issues":719,"watchers":8467,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/24892"},"html":{"href":"https://github.com/akka/akka/pull/24892"},"issue":{"href":"https://api.github.com/repos/akka/akka/issues/24892"},"comments":{"href":"https://api.github.com/repos/akka/akka/issues/24892/comments"},"review_comments":{"href":"https://api.github.com/repos/akka/akka/pulls/24892/comments"},"review_comment":{"href":"https://api.github.com/repos/akka/akka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/akka/akka/pulls/24892/commits"},"statuses":{"href":"https://api.github.com/repos/akka/akka/statuses/cda43759f6978fd06f32dd53fc3daf83ba32a051"}},"author_association":"OWNER"}}
{"actor":{"display_login":"ktoso"}}