{"action":"created","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3964","repository_url":"https://api.github.com/repos/ReactiveX/RxJava","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3964/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3964/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3964/events","html_url":"https://github.com/ReactiveX/RxJava/pull/3964","id":156846449,"number":3964,"title":"1.x: observeOn + immediate scheduler to be a request rebatcher","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/ReactiveX/RxJava/labels/Enhancement","name":"Enhancement","color":"84b6eb"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2016-05-25T20:46:37Z","updated_at":"2016-05-26T12:48:18Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/3964","html_url":"https://github.com/ReactiveX/RxJava/pull/3964","diff_url":"https://github.com/ReactiveX/RxJava/pull/3964.diff","patch_url":"https://github.com/ReactiveX/RxJava/pull/3964.patch"},"body":"This PR removes the `immediate()` scheduler \"optimization\" from `observeOn` and treats it as a common scheduler. Since `observeOn` has a stable request pattern, this turns it into a rebatching operator. No matter what the downstream requests, the upstream will requests of the specified size (with 25% as low water mark; i.e., replenishment after 75%). Since `immediate` is synchronous, this will run the drain loop, non-reentrant, on the caller thread.\r\n\r\nI found this mode of operation very handy in my Reactive-RPC prototype and a simple streaming echo RPC call; it prevents going unbounded and bloating the message sender threads:\r\n\r\n```java\r\n// remote\r\npublic Observable<Integer> echo(RpcStreamContext<?> ctx, Observable<Integer> in) {\r\n   return in.observeOn(Schedulers.immediate(), 16);\r\n}\r\n\r\n// client\r\napi.echo(Observable.range(1, 100_000)).observeOn(Schedulers.immediate(), 32)\r\n.subscribe(System.out::println, Throwable::printStackTrace);\r\n```\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/221861072","html_url":"https://github.com/ReactiveX/RxJava/pull/3964#issuecomment-221861072","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3964","id":221861072,"user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"created_at":"2016-05-26T12:48:18Z","updated_at":"2016-05-26T12:48:18Z","body":"That could be a follow-up PR if this gets accepted."}}
{"actor":{"display_login":"akarnokd"}}