{"action":"created","comment":{"url":"https://api.github.com/repos/tumblr/colossus/pulls/comments/84817719","pull_request_review_id":5570729,"id":84817719,"diff_hunk":"@@ -82,13 +83,17 @@ object WebsocketExample {\n           case \"MANY\" => {","path":"colossus-examples/src/main/scala/colossus-examples/WebsocketExample.scala","position":18,"original_position":18,"commit_id":"21ba2fe539f338457e4f0f0f4aa76d0c690b9265","original_commit_id":"21ba2fe539f338457e4f0f0f4aa76d0c690b9265","user":{"login":"DanSimon","id":287961,"avatar_url":"https://avatars.githubusercontent.com/u/287961?v=3","gravatar_id":"","url":"https://api.github.com/users/DanSimon","html_url":"https://github.com/DanSimon","followers_url":"https://api.github.com/users/DanSimon/followers","following_url":"https://api.github.com/users/DanSimon/following{/other_user}","gists_url":"https://api.github.com/users/DanSimon/gists{/gist_id}","starred_url":"https://api.github.com/users/DanSimon/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/DanSimon/subscriptions","organizations_url":"https://api.github.com/users/DanSimon/orgs","repos_url":"https://api.github.com/users/DanSimon/repos","events_url":"https://api.github.com/users/DanSimon/events{/privacy}","received_events_url":"https://api.github.com/users/DanSimon/received_events","type":"User","site_admin":false},"body":"Yes, every connection gets a unique proxy actor.  Also in this case \"MANY\" will block only until the output buffer fills up (which I believe defaults to 50).  When it comes to blocking, it's true that while the output buffer is being filled, nothing else can occur.  In this situation it doesn't matter since it happens very quickly.  I'd imagine if generating the messages to send required significant CPU usage, it would be done on a different thread.\r\n\r\nI have thought about ways to have more control over the \"greediness\" of pipes and situations like this.  One potential is to create a \"deferred\" wrapper for a pipe which will force the return of a \"Full\" result after a certain number of items have been pulled and fire off the returned signal at the beginning of the next event loop iteration (using a CallbackExecutor).  This would let you, for example, force the \"MANY\" situation to only send one item per event loop iteration.\r\n\r\nI think in most cases when it comes to dealing with a \"Full\" vs \"Ok\" result, you just want to wait until the underlying buffer can accept more items and continue from there.  In fact, the code here can be cleaned up a lot by just wrapping this in an iterator and forwarding it into the output buffer directly using `Source.into`.  I may update or add to this example to show that.","created_at":"2016-10-25T01:42:52Z","updated_at":"2016-10-25T01:42:52Z","html_url":"https://github.com/tumblr/colossus/pull/460#discussion_r84817719","pull_request_url":"https://api.github.com/repos/tumblr/colossus/pulls/460","_links":{"self":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/comments/84817719"},"html":{"href":"https://github.com/tumblr/colossus/pull/460#discussion_r84817719"},"pull_request":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/460"}}},"pull_request":{"url":"https://api.github.com/repos/tumblr/colossus/pulls/460","id":90683014,"html_url":"https://github.com/tumblr/colossus/pull/460","diff_url":"https://github.com/tumblr/colossus/pull/460.diff","patch_url":"https://github.com/tumblr/colossus/pull/460.patch","issue_url":"https://api.github.com/repos/tumblr/colossus/issues/460","number":460,"state":"open","locked":false,"title":"Replace all connection handler backpressure logic with pipes","user":{"login":"DanSimon","id":287961,"avatar_url":"https://avatars.githubusercontent.com/u/287961?v=3","gravatar_id":"","url":"https://api.github.com/users/DanSimon","html_url":"https://github.com/DanSimon","followers_url":"https://api.github.com/users/DanSimon/followers","following_url":"https://api.github.com/users/DanSimon/following{/other_user}","gists_url":"https://api.github.com/users/DanSimon/gists{/gist_id}","starred_url":"https://api.github.com/users/DanSimon/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/DanSimon/subscriptions","organizations_url":"https://api.github.com/users/DanSimon/orgs","repos_url":"https://api.github.com/users/DanSimon/repos","events_url":"https://api.github.com/users/DanSimon/events{/privacy}","received_events_url":"https://api.github.com/users/DanSimon/received_events","type":"User","site_admin":false},"body":"Once in a blue moon you really end up killing two birds with one stone.  In a nutshell, this PR replaces all of the connection handler buffering and backpressure logic with the new reincarnation of pipes.  The end result is much cleaner code and the generalization of a pattern found in several places (buffering and backpressure) into a single solution.\r\n\r\n### Preface\r\n\r\n`Pipe[I,O]` was originally created for streaming services: the idea being that a request or response could be streamed in/out without having to load the entire thing in memory before using it.  `Pipe` was the representation of the streaming data, where pieces of the message could be \"pushed\" into and \"pulled\" out of.  Originally created in version 0.6.0, `Pipe`has been heavily reworked and dramatically improved in the upcoming 0.9.0 release.  It has essentially transformed into a buffer with a monad-like interface that can efficiently handle backpressure.\r\n\r\nMeanwhile, buffering and backpressure has been a central concern in the controller and service layers of the connection handler architecture.  To recap, the `Controller` layer is the part of a connection handler responsible for reading in raw data and decoding it into message objects like Http requests, and vise versa for encoding messages into raw data.  The `Service` layer takes the incoming request messages and processes them into response messages, dealing with all the logic to make sure responses are sent back in the right order.\r\n\r\nUp until now buffering and backpressure between these layers have been handled somewhat clumsily.  For instance, `Controller` has a `push`method that would return `true`or `false` depending on whether the pushed message was successfully buffered or not.  It was then up to the service layer to deal with the result manually.  This also led to `Controller`having a lot of exposed methods like `pauseReads()` and `pendingBufferSize` that led to a lot of weird logic, and probably a lot of weird edge cases.\r\n\r\n### Using Pipes in Controller and ServiceServer/ServiceClient\r\n\r\nWell now all that is gone.  Instead now pipes are used as the interface between layers.  So now, for example, the service layer can simply access an `outgoing` `Sink` from the controller and push responses there.  Likewise, the controller uses an `incoming` `Sink` to push decoded messages.\r\n\r\nOn the client side, the biggest change is `ServiceClient` now maintains its own buffer for pending requests.  Previously it used the controller's buffer to do this, but again that led to a lot of logic bleedthrough.  But now with pipes the client can simply route it's pending pipe into the controller's outgoing pipe.\r\n\r\n### Additional Changes to Pipes\r\n\r\nA few changes were made to pipes to get this all working.\r\n\r\nThe biggest change is to `Source.pullWhile`.  Instead of taking a function of type `T => Boolean`, it now takes a `T => PullAction`.  This gives the looping function greater control of the iteration process.  Most importantly, it can now decide to pull an item after receiving it by returning `PullAction.PullContinue`, which was very important when feeding a `Source` into a `Sink`.  \r\n\r\nA new method `pullUntilNull` has been created which is similar to `pullWhile` but is intended to be used to pull out any items currently buffered and not wait for more items afterwards.  It is mainly used by `OutputController` to pull out items to write to the output buffer.\r\n\r\nAnother new method `Sink.valve`was created which essentially creates a new sink that splits into two sinks.  An item is only pushed if it can be pushed to both sub-sinks.  This is mainly used by `ServiceClient` to properly handle the `sentBufferSize` to limit the number of concurrently in-flight requests.  \r\n\r\n### Performance\r\n\r\nAfter some basic benchmarking, there appears to be no change in performance for a basic service.  I haven't done extensive benchmarking on the client yet, but initial results don't show anything drastic.\r\n\r\n### What's Next\r\n\r\nThis PR clearly the first of several, as there's definitely some cleanup work to do, but this PR is already large enough as is.","created_at":"2016-10-24T20:08:31Z","updated_at":"2016-10-25T01:42:52Z","closed_at":null,"merged_at":null,"merge_commit_sha":"54d25e2fcaae137b78b4f1b28859f79a20a3e366","assignee":null,"assignees":[],"milestone":{"url":"https://api.github.com/repos/tumblr/colossus/milestones/28","html_url":"https://github.com/tumblr/colossus/milestones/0.9.0","labels_url":"https://api.github.com/repos/tumblr/colossus/milestones/28/labels","id":1758393,"number":28,"title":"0.9.0","description":"","creator":{"login":"DanSimon","id":287961,"avatar_url":"https://avatars.githubusercontent.com/u/287961?v=3","gravatar_id":"","url":"https://api.github.com/users/DanSimon","html_url":"https://github.com/DanSimon","followers_url":"https://api.github.com/users/DanSimon/followers","following_url":"https://api.github.com/users/DanSimon/following{/other_user}","gists_url":"https://api.github.com/users/DanSimon/gists{/gist_id}","starred_url":"https://api.github.com/users/DanSimon/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/DanSimon/subscriptions","organizations_url":"https://api.github.com/users/DanSimon/orgs","repos_url":"https://api.github.com/users/DanSimon/repos","events_url":"https://api.github.com/users/DanSimon/events{/privacy}","received_events_url":"https://api.github.com/users/DanSimon/received_events","type":"User","site_admin":false},"open_issues":10,"closed_issues":13,"state":"open","created_at":"2016-05-10T17:04:26Z","updated_at":"2016-10-24T20:08:32Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/tumblr/colossus/pulls/460/commits","review_comments_url":"https://api.github.com/repos/tumblr/colossus/pulls/460/comments","review_comment_url":"https://api.github.com/repos/tumblr/colossus/pulls/comments{/number}","comments_url":"https://api.github.com/repos/tumblr/colossus/issues/460/comments","statuses_url":"https://api.github.com/repos/tumblr/colossus/statuses/21ba2fe539f338457e4f0f0f4aa76d0c690b9265","head":{"label":"tumblr:pipe-the-world","ref":"pipe-the-world","sha":"21ba2fe539f338457e4f0f0f4aa76d0c690b9265","user":{"login":"tumblr","id":366151,"avatar_url":"https://avatars.githubusercontent.com/u/366151?v=3","gravatar_id":"","url":"https://api.github.com/users/tumblr","html_url":"https://github.com/tumblr","followers_url":"https://api.github.com/users/tumblr/followers","following_url":"https://api.github.com/users/tumblr/following{/other_user}","gists_url":"https://api.github.com/users/tumblr/gists{/gist_id}","starred_url":"https://api.github.com/users/tumblr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tumblr/subscriptions","organizations_url":"https://api.github.com/users/tumblr/orgs","repos_url":"https://api.github.com/users/tumblr/repos","events_url":"https://api.github.com/users/tumblr/events{/privacy}","received_events_url":"https://api.github.com/users/tumblr/received_events","type":"Organization","site_admin":false},"repo":{"id":26492586,"name":"colossus","full_name":"tumblr/colossus","owner":{"login":"tumblr","id":366151,"avatar_url":"https://avatars.githubusercontent.com/u/366151?v=3","gravatar_id":"","url":"https://api.github.com/users/tumblr","html_url":"https://github.com/tumblr","followers_url":"https://api.github.com/users/tumblr/followers","following_url":"https://api.github.com/users/tumblr/following{/other_user}","gists_url":"https://api.github.com/users/tumblr/gists{/gist_id}","starred_url":"https://api.github.com/users/tumblr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tumblr/subscriptions","organizations_url":"https://api.github.com/users/tumblr/orgs","repos_url":"https://api.github.com/users/tumblr/repos","events_url":"https://api.github.com/users/tumblr/events{/privacy}","received_events_url":"https://api.github.com/users/tumblr/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/tumblr/colossus","description":"I/O and Microservice library for Scala","fork":false,"url":"https://api.github.com/repos/tumblr/colossus","forks_url":"https://api.github.com/repos/tumblr/colossus/forks","keys_url":"https://api.github.com/repos/tumblr/colossus/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tumblr/colossus/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tumblr/colossus/teams","hooks_url":"https://api.github.com/repos/tumblr/colossus/hooks","issue_events_url":"https://api.github.com/repos/tumblr/colossus/issues/events{/number}","events_url":"https://api.github.com/repos/tumblr/colossus/events","assignees_url":"https://api.github.com/repos/tumblr/colossus/assignees{/user}","branches_url":"https://api.github.com/repos/tumblr/colossus/branches{/branch}","tags_url":"https://api.github.com/repos/tumblr/colossus/tags","blobs_url":"https://api.github.com/repos/tumblr/colossus/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tumblr/colossus/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tumblr/colossus/git/refs{/sha}","trees_url":"https://api.github.com/repos/tumblr/colossus/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tumblr/colossus/statuses/{sha}","languages_url":"https://api.github.com/repos/tumblr/colossus/languages","stargazers_url":"https://api.github.com/repos/tumblr/colossus/stargazers","contributors_url":"https://api.github.com/repos/tumblr/colossus/contributors","subscribers_url":"https://api.github.com/repos/tumblr/colossus/subscribers","subscription_url":"https://api.github.com/repos/tumblr/colossus/subscription","commits_url":"https://api.github.com/repos/tumblr/colossus/commits{/sha}","git_commits_url":"https://api.github.com/repos/tumblr/colossus/git/commits{/sha}","comments_url":"https://api.github.com/repos/tumblr/colossus/comments{/number}","issue_comment_url":"https://api.github.com/repos/tumblr/colossus/issues/comments{/number}","contents_url":"https://api.github.com/repos/tumblr/colossus/contents/{+path}","compare_url":"https://api.github.com/repos/tumblr/colossus/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tumblr/colossus/merges","archive_url":"https://api.github.com/repos/tumblr/colossus/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tumblr/colossus/downloads","issues_url":"https://api.github.com/repos/tumblr/colossus/issues{/number}","pulls_url":"https://api.github.com/repos/tumblr/colossus/pulls{/number}","milestones_url":"https://api.github.com/repos/tumblr/colossus/milestones{/number}","notifications_url":"https://api.github.com/repos/tumblr/colossus/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tumblr/colossus/labels{/name}","releases_url":"https://api.github.com/repos/tumblr/colossus/releases{/id}","deployments_url":"https://api.github.com/repos/tumblr/colossus/deployments","created_at":"2014-11-11T16:06:06Z","updated_at":"2016-10-23T16:12:03Z","pushed_at":"2016-10-24T20:08:32Z","git_url":"git://github.com/tumblr/colossus.git","ssh_url":"git@github.com:tumblr/colossus.git","clone_url":"https://github.com/tumblr/colossus.git","svn_url":"https://github.com/tumblr/colossus","homepage":"","size":7362,"stargazers_count":836,"watchers_count":836,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":72,"mirror_url":null,"open_issues_count":25,"forks":72,"open_issues":25,"watchers":836,"default_branch":"master"}},"base":{"label":"tumblr:develop","ref":"develop","sha":"7f4b81946688baa360baf43a0863ed154c3469f7","user":{"login":"tumblr","id":366151,"avatar_url":"https://avatars.githubusercontent.com/u/366151?v=3","gravatar_id":"","url":"https://api.github.com/users/tumblr","html_url":"https://github.com/tumblr","followers_url":"https://api.github.com/users/tumblr/followers","following_url":"https://api.github.com/users/tumblr/following{/other_user}","gists_url":"https://api.github.com/users/tumblr/gists{/gist_id}","starred_url":"https://api.github.com/users/tumblr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tumblr/subscriptions","organizations_url":"https://api.github.com/users/tumblr/orgs","repos_url":"https://api.github.com/users/tumblr/repos","events_url":"https://api.github.com/users/tumblr/events{/privacy}","received_events_url":"https://api.github.com/users/tumblr/received_events","type":"Organization","site_admin":false},"repo":{"id":26492586,"name":"colossus","full_name":"tumblr/colossus","owner":{"login":"tumblr","id":366151,"avatar_url":"https://avatars.githubusercontent.com/u/366151?v=3","gravatar_id":"","url":"https://api.github.com/users/tumblr","html_url":"https://github.com/tumblr","followers_url":"https://api.github.com/users/tumblr/followers","following_url":"https://api.github.com/users/tumblr/following{/other_user}","gists_url":"https://api.github.com/users/tumblr/gists{/gist_id}","starred_url":"https://api.github.com/users/tumblr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tumblr/subscriptions","organizations_url":"https://api.github.com/users/tumblr/orgs","repos_url":"https://api.github.com/users/tumblr/repos","events_url":"https://api.github.com/users/tumblr/events{/privacy}","received_events_url":"https://api.github.com/users/tumblr/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/tumblr/colossus","description":"I/O and Microservice library for Scala","fork":false,"url":"https://api.github.com/repos/tumblr/colossus","forks_url":"https://api.github.com/repos/tumblr/colossus/forks","keys_url":"https://api.github.com/repos/tumblr/colossus/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tumblr/colossus/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tumblr/colossus/teams","hooks_url":"https://api.github.com/repos/tumblr/colossus/hooks","issue_events_url":"https://api.github.com/repos/tumblr/colossus/issues/events{/number}","events_url":"https://api.github.com/repos/tumblr/colossus/events","assignees_url":"https://api.github.com/repos/tumblr/colossus/assignees{/user}","branches_url":"https://api.github.com/repos/tumblr/colossus/branches{/branch}","tags_url":"https://api.github.com/repos/tumblr/colossus/tags","blobs_url":"https://api.github.com/repos/tumblr/colossus/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tumblr/colossus/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tumblr/colossus/git/refs{/sha}","trees_url":"https://api.github.com/repos/tumblr/colossus/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tumblr/colossus/statuses/{sha}","languages_url":"https://api.github.com/repos/tumblr/colossus/languages","stargazers_url":"https://api.github.com/repos/tumblr/colossus/stargazers","contributors_url":"https://api.github.com/repos/tumblr/colossus/contributors","subscribers_url":"https://api.github.com/repos/tumblr/colossus/subscribers","subscription_url":"https://api.github.com/repos/tumblr/colossus/subscription","commits_url":"https://api.github.com/repos/tumblr/colossus/commits{/sha}","git_commits_url":"https://api.github.com/repos/tumblr/colossus/git/commits{/sha}","comments_url":"https://api.github.com/repos/tumblr/colossus/comments{/number}","issue_comment_url":"https://api.github.com/repos/tumblr/colossus/issues/comments{/number}","contents_url":"https://api.github.com/repos/tumblr/colossus/contents/{+path}","compare_url":"https://api.github.com/repos/tumblr/colossus/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tumblr/colossus/merges","archive_url":"https://api.github.com/repos/tumblr/colossus/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tumblr/colossus/downloads","issues_url":"https://api.github.com/repos/tumblr/colossus/issues{/number}","pulls_url":"https://api.github.com/repos/tumblr/colossus/pulls{/number}","milestones_url":"https://api.github.com/repos/tumblr/colossus/milestones{/number}","notifications_url":"https://api.github.com/repos/tumblr/colossus/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tumblr/colossus/labels{/name}","releases_url":"https://api.github.com/repos/tumblr/colossus/releases{/id}","deployments_url":"https://api.github.com/repos/tumblr/colossus/deployments","created_at":"2014-11-11T16:06:06Z","updated_at":"2016-10-23T16:12:03Z","pushed_at":"2016-10-24T20:08:32Z","git_url":"git://github.com/tumblr/colossus.git","ssh_url":"git@github.com:tumblr/colossus.git","clone_url":"https://github.com/tumblr/colossus.git","svn_url":"https://github.com/tumblr/colossus","homepage":"","size":7362,"stargazers_count":836,"watchers_count":836,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":72,"mirror_url":null,"open_issues_count":25,"forks":72,"open_issues":25,"watchers":836,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/460"},"html":{"href":"https://github.com/tumblr/colossus/pull/460"},"issue":{"href":"https://api.github.com/repos/tumblr/colossus/issues/460"},"comments":{"href":"https://api.github.com/repos/tumblr/colossus/issues/460/comments"},"review_comments":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/460/comments"},"review_comment":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/tumblr/colossus/pulls/460/commits"},"statuses":{"href":"https://api.github.com/repos/tumblr/colossus/statuses/21ba2fe539f338457e4f0f0f4aa76d0c690b9265"}}}}
{"actor":{"display_login":"DanSimon"}}