{"action":"created","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3492","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3492/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3492/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3492/events","html_url":"https://github.com/ReactiveX/RxJava/issues/3492","id":115056106,"number":3492,"title":"Backpressure on OperatorZip doesn't work when the observable is created using toList()","user":{"login":"MaltAlex","id":10195391,"avatar_url":"https://avatars.githubusercontent.com/u/10195391?v=3","gravatar_id":"","url":"https://api.github.com/users/MaltAlex","html_url":"https://github.com/MaltAlex","followers_url":"https://api.github.com/users/MaltAlex/followers","following_url":"https://api.github.com/users/MaltAlex/following{/other_user}","gists_url":"https://api.github.com/users/MaltAlex/gists{/gist_id}","starred_url":"https://api.github.com/users/MaltAlex/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/MaltAlex/subscriptions","organizations_url":"https://api.github.com/users/MaltAlex/orgs","repos_url":"https://api.github.com/users/MaltAlex/repos","events_url":"https://api.github.com/users/MaltAlex/events{/privacy}","received_events_url":"https://api.github.com/users/MaltAlex/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/ReactiveX/RxJava/labels/Bug","name":"Bug","color":"fc2929"}],"state":"open","locked":false,"assignee":null,"milestone":{"url":"https://api.github.com/repos/ReactiveX/RxJava/milestones/11","html_url":"https://github.com/ReactiveX/RxJava/milestones/1.0.x","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/milestones/11/labels","id":817560,"number":11,"title":"1.0.x","description":"Patch fixes on 1.0.x that do not add new functionality.","creator":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=3","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"open_issues":39,"closed_issues":58,"state":"open","created_at":"2014-10-07T22:28:55Z","updated_at":"2015-11-04T14:17:23Z","due_on":null,"closed_at":null},"comments":1,"created_at":"2015-11-04T13:56:44Z","updated_at":"2015-11-04T14:23:09Z","closed_at":null,"body":"I've written an \"SortedMerge\" operator, which is used to orderly merge several ordered observables. The operator is based on [OperatorZip](https://github.com/ReactiveX/RxJava/blob/f2410f88dc28c1013f48754606e38507e493245f/src/main/java/rx/internal/operators/OperatorZip.java) except for several key differences:\r\n\r\n 1. Instead of passing an array of items into a zip function the items are compared using a comparator and only the smallest one is emitted.\r\n 2. After emissions, `requestMore` is called only on the observable from which the minimum was taken, and only its buffer's head is removed. The others are left untouched for future comparison. So no items are lost.\r\n 3. Additional type safety is added since the types of the incoming observables is known.\r\n\r\n(I can contribute the operator to the project if anyone's interested).\r\n\r\nDuring testing, I stumbled upon a use case that failed for my operator, and it turns out that the same problem exists in the original Zip.\r\n\r\nTake a look at the following observables. First, o1:\r\n\r\n\tObservable<Integer>[] osArray = new Observable[] { Observable.range(0, 10), Observable.range(0, 10) };\r\n\tObservable<Observable<Integer>> os = (Observable<Observable<Integer>>) Observable.from(osArray);\r\n\tObservable<Integer> o1 = Observable.zip(os, (a) -> 0);\r\n\r\nSecond, o2:\r\n\r\n\tObservable<Integer> o2 = Observable.zip(Observable.range(0, 10), Observable.range(0, 10), (a, b) -> 0);\r\n\r\nAccording to my understanding, both should work in exactly the same way. However, their response to requests from the subscriber is different:\r\n\r\n\tTestSubscriber<Integer> sub1 = new TestSubscriber<>(5);\r\n\tTestSubscriber<Integer> sub2 = new TestSubscriber<>(5);\r\n\t\r\n\to1.subscribe(sub1);\r\n\to2.subscribe(sub2);\r\n\t\r\n\tsub1.requestMore(5);\r\n\tsub2.requestMore(5);\r\n\t\r\n\tsub1.awaitTerminalEvent(1, TimeUnit.SECONDS);\r\n\tsub2.awaitTerminalEvent(1, TimeUnit.SECONDS);\r\n\t\r\n\tSystem.out.println(sub1.getOnNextEvents().size());\r\n\tSystem.out.println(sub2.getOnNextEvents().size());\r\n\r\nproduces this response:\r\n     5\r\n     10\r\n\r\nWhich is exactly what happened to my SortedMerge operator. After poking at the operator for a while, it turned out that the issue can be solved by changing the construction of the Zip subscriber ZipSubscriber from this:\r\n\r\n    public ZipSubscriber(Subscriber<? super R> child, Zip<R> zipper, ZipProducer<R> producer) {\r\n            super(child);\r\n            this.child = child;\r\n            this.zipper = zipper;\r\n            this.producer = producer;\r\n        }\r\n\r\nto this:\r\n\r\n    public ZipSubscriber(Subscriber<? super R> child, Zip<R> zipper, ZipProducer<R> producer) {\r\n            super();\r\n            this.child = child;\r\n            this.zipper = zipper;\r\n            this.producer = producer;\r\n        }\r\n\r\nWhich seems to make sense, but I'm not sure that I understand Rx's backpressure well enough to know whether that's indeed the right solution."},"comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/153737396","html_url":"https://github.com/ReactiveX/RxJava/issues/3492#issuecomment-153737396","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3492","id":153737396,"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":"2015-11-04T14:23:09Z","updated_at":"2015-11-04T14:23:09Z","body":"Hello and thanks for the in-depth analysis. What happens is that the child is caputed by `ZipSubscriber` and when it receives its `Producer` from `toList`, that also gets forwarded to child which overwrites the `ZipProducer` and thus no further requests get accounted. The fix is posted in #3493\r\n\r\nOrdered merging comes up from time to time on StackOverflow; you can post a PR for it but frankly, if Netflix doesn't need one urgently, it can take a long time to get it approved."}}