{"action":"created","comment":{"url":"https://api.github.com/repos/akka/akka/pulls/comments/40774746","id":40774746,"diff_hunk":"@@ -120,35 +120,81 @@ private[http] object Websocket {\n       }\n     }\n \n-    /**\n-     * Distributes output from the FrameHandler into bypass and userFlow.\n-     */\n-    object BypassRouter\n-      extends FlexiRoute[Either[BypassEvent, MessagePart], FanOutShape2[Either[BypassEvent, MessagePart], BypassEvent, MessagePart]](new FanOutShape2(\"bypassRouter\"), Attributes.name(\"bypassRouter\")) {\n-      def createRouteLogic(s: FanOutShape2[Either[BypassEvent, MessagePart], BypassEvent, MessagePart]): RouteLogic[Either[BypassEvent, MessagePart]] =\n-        new RouteLogic[Either[BypassEvent, MessagePart]] {\n-          def initialState: State[_] = State(DemandFromAny(s)) { (ctx, out, ev) ⇒\n-            ev match {\n-              case Left(_) ⇒\n-                State(DemandFrom(s.out0)) { (ctx, _, ev) ⇒ // FIXME: #17004\n-                  ctx.emit(s.out0)(ev.left.get)\n-                  initialState\n-                }\n-              case Right(_) ⇒\n-                State(DemandFrom(s.out1)) { (ctx, _, ev) ⇒\n-                  ctx.emit(s.out1)(ev.right.get)\n-                  initialState\n-                }\n+    def bypassRouter = new GraphStage[FanOutShape2[Output, BypassEvent, MessagePart]] {\n+      private val in = Inlet[Output](\"in\")\n+      private val bypass = Outlet[BypassEvent](\"bypass-out\")\n+      private val user = Outlet[MessagePart](\"message-out\")\n+\n+      def shape = new FanOutShape2(in, bypass, user)\n+      def createLogic = new GraphStageLogic {\n+\n+        setHandler(in, new InHandler {\n+          override def onPush(): Unit = {\n+            grab(in) match {\n+              case b: BypassEvent with MessagePart ⇒ emit(bypass, b, emit(user, b, pull(in)))\n+              case b: BypassEvent                  ⇒ emit(bypass, b, pull(in))\n+              case m: MessagePart                  ⇒ emit(user, m, pull(in))\n             }\n           }\n+        })\n+\n+        setHandler(bypass, EagerTerminateOutput)\n+        setHandler(user, IgnoreTerminateOutput)\n+\n+        override def preStart(): Unit = {\n+          super.preStart()\n+          pull(in)\n+        }\n+\n+        /*\n+         * Everything below to be pushed into GraphStageLogic for reuse\n+         */\n+\n+        private object EagerTerminateOutput extends OutHandler {\n+          override def onPull(): Unit = ()\n+        }\n+\n+        private object IgnoreTerminateOutput extends OutHandler {\n+          override def onPull(): Unit = ()\n+          override def onDownstreamFinish(): Unit = ()\n+        }\n+\n+        private def emit[T](out: Outlet[T], elem: T, andThen: ⇒ Unit): Unit = {\n+          if (isAvailable(out)) {\n+            push(out, elem)\n+            andThen\n+          } else {\n+            setHandler(out, new Emitting(out, elem, getHandler(out), () ⇒ andThen))\n+          }\n+        }\n \n-          override def initialCompletionHandling: CompletionHandling = super.initialCompletionHandling.copy(\n-            onDownstreamFinish = { (ctx, out) ⇒\n-              if (out == s.out0) ctx.finish()\n-              SameState\n-            })\n+        private def emit[T](out: Outlet[T], elem: T): Unit = {\n+          if (isAvailable(out)) {\n+            push(out, elem)\n+          } else {\n+            setHandler(out, new Emitting(out, elem, getHandler(out), DoNothing))\n+          }\n         }\n+\n+        private class Emitting[T](out: Outlet[T], elem: T, next: OutHandler, andThen: () ⇒ Unit) extends OutHandler {\n+          if (getHandler(out).isInstanceOf[Emitting[_]])","path":"akka-http-core/src/main/scala/akka/http/impl/engine/ws/Websocket.scala","position":85,"original_position":85,"commit_id":"2d5811225a7afedb6b78b84f4768de8a5e5f739b","original_commit_id":"2d5811225a7afedb6b78b84f4768de8a5e5f739b","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":"ah I see, it is checked. That will probably be an `isInstanceOf[InternalAction]` or something like that, accompanied by an `abort(port)` which allows removing the handler (or something similar).","created_at":"2015-09-30T09:32:08Z","updated_at":"2015-09-30T09:32:08Z","html_url":"https://github.com/akka/akka/pull/18605#discussion_r40774746","pull_request_url":"https://api.github.com/repos/akka/akka/pulls/18605","_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/comments/40774746"},"html":{"href":"https://github.com/akka/akka/pull/18605#discussion_r40774746"},"pull_request":{"href":"https://api.github.com/repos/akka/akka/pulls/18605"}}},"pull_request":{"url":"https://api.github.com/repos/akka/akka/pulls/18605","id":46378174,"html_url":"https://github.com/akka/akka/pull/18605","diff_url":"https://github.com/akka/akka/pull/18605.diff","patch_url":"https://github.com/akka/akka/pull/18605.patch","issue_url":"https://api.github.com/repos/akka/akka/issues/18605","number":18605,"state":"open","locked":false,"title":"=htp replace WS.bypassRouter with GraphStage","user":{"login":"rkuhn","id":470469,"avatar_url":"https://avatars.githubusercontent.com/u/470469?v=3","gravatar_id":"","url":"https://api.github.com/users/rkuhn","html_url":"https://github.com/rkuhn","followers_url":"https://api.github.com/users/rkuhn/followers","following_url":"https://api.github.com/users/rkuhn/following{/other_user}","gists_url":"https://api.github.com/users/rkuhn/gists{/gist_id}","starred_url":"https://api.github.com/users/rkuhn/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rkuhn/subscriptions","organizations_url":"https://api.github.com/users/rkuhn/orgs","repos_url":"https://api.github.com/users/rkuhn/repos","events_url":"https://api.github.com/users/rkuhn/events{/privacy}","received_events_url":"https://api.github.com/users/rkuhn/received_events","type":"User","site_admin":false},"body":"Early preview of the working principle for review before diving in for real: please check mostly what is in `Websocket.scala`.\r\n\r\n* also found a usability improvement (better error message for when ports don’t have handlers at the right time\r\n* will also add nicer errors for when calling `pull()` and friends before `preStart()` (currently just gives a NPE)\r\n* and fixed some more warnings related to https://issues.scala-lang.org/browse/SI-9490","created_at":"2015-09-30T09:18:19Z","updated_at":"2015-09-30T09:32:08Z","closed_at":null,"merged_at":null,"merge_commit_sha":"3467024ff1df1b18fba0b47e7f3a03f76ec0f1d3","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/akka/akka/pulls/18605/commits","review_comments_url":"https://api.github.com/repos/akka/akka/pulls/18605/comments","review_comment_url":"https://api.github.com/repos/akka/akka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/akka/akka/issues/18605/comments","statuses_url":"https://api.github.com/repos/akka/akka/statuses/2d5811225a7afedb6b78b84f4768de8a5e5f739b","head":{"label":"akka:wip-HTTP-graphs-∂π","ref":"wip-HTTP-graphs-∂π","sha":"2d5811225a7afedb6b78b84f4768de8a5e5f739b","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":130013,"name":"akka","full_name":"akka/akka","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","description":"Akka Project","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}","created_at":"2009-02-16T12:51:54Z","updated_at":"2015-09-30T09:22:52Z","pushed_at":"2015-09-30T09:31:21Z","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":272647,"stargazers_count":4501,"watchers_count":4501,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1334,"mirror_url":null,"open_issues_count":747,"forks":1334,"open_issues":747,"watchers":4501,"default_branch":"master"}},"base":{"label":"akka:release-2.3-dev","ref":"release-2.3-dev","sha":"d841a20c90112e8fda4b191284297366387e22c7","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":130013,"name":"akka","full_name":"akka/akka","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","description":"Akka Project","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}","created_at":"2009-02-16T12:51:54Z","updated_at":"2015-09-30T09:22:52Z","pushed_at":"2015-09-30T09:31:21Z","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":272647,"stargazers_count":4501,"watchers_count":4501,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1334,"mirror_url":null,"open_issues_count":747,"forks":1334,"open_issues":747,"watchers":4501,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/18605"},"html":{"href":"https://github.com/akka/akka/pull/18605"},"issue":{"href":"https://api.github.com/repos/akka/akka/issues/18605"},"comments":{"href":"https://api.github.com/repos/akka/akka/issues/18605/comments"},"review_comments":{"href":"https://api.github.com/repos/akka/akka/pulls/18605/comments"},"review_comment":{"href":"https://api.github.com/repos/akka/akka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/akka/akka/pulls/18605/commits"},"statuses":{"href":"https://api.github.com/repos/akka/akka/statuses/2d5811225a7afedb6b78b84f4768de8a5e5f739b"}}}}