{"action":"created","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3020","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3020/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3020/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3020/events","html_url":"https://github.com/ReactiveX/RxJava/issues/3020","id":87564465,"number":3020,"title":"Custom Operators and onBackpressureBuffer","user":{"login":"davidmoten","id":318187,"avatar_url":"https://avatars.githubusercontent.com/u/318187?v=3","gravatar_id":"","url":"https://api.github.com/users/davidmoten","html_url":"https://github.com/davidmoten","followers_url":"https://api.github.com/users/davidmoten/followers","following_url":"https://api.github.com/users/davidmoten/following{/other_user}","gists_url":"https://api.github.com/users/davidmoten/gists{/gist_id}","starred_url":"https://api.github.com/users/davidmoten/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davidmoten/subscriptions","organizations_url":"https://api.github.com/users/davidmoten/orgs","repos_url":"https://api.github.com/users/davidmoten/repos","events_url":"https://api.github.com/users/davidmoten/events{/privacy}","received_events_url":"https://api.github.com/users/davidmoten/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":3,"created_at":"2015-06-12T01:14:38Z","updated_at":"2015-06-12T12:01:42Z","closed_at":null,"body":"A very common use case for me is  to need an `Operator` that takes a stream and for every input makes 0..* emissions. \r\n\r\nI can get a backpressure compliant Operator by:\r\n\r\n* requesting one from upstream when no emissions are made from an arrival so stream doesn't stall\r\n* buffering emissions using the new `QueuedProducer` \r\n\r\nThe second action is doable but is a bit invasive to implement and I would normally check the RxJava codebase for an Operator that used it and make sure I did something similar. \r\n\r\nInstead of using `QueuedProducer` I'd like a simpler solution that uses composition and `onBackpressureBuffer` looks like the one but has one significant defect in that it always requests `Long.MAX_VALUE` of upstream instead of requesting only what it needs.\r\n\r\n Is there an idiom I'm overlooking? Can we modify `OperatorOnBackpressureBuffer` to support this use case or would a new `Operator` be the go (that uses `QueuedProducer` internally).\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/111468732","html_url":"https://github.com/ReactiveX/RxJava/issues/3020#issuecomment-111468732","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3020","id":111468732,"user":{"login":"davidmoten","id":318187,"avatar_url":"https://avatars.githubusercontent.com/u/318187?v=3","gravatar_id":"","url":"https://api.github.com/users/davidmoten","html_url":"https://github.com/davidmoten","followers_url":"https://api.github.com/users/davidmoten/followers","following_url":"https://api.github.com/users/davidmoten/following{/other_user}","gists_url":"https://api.github.com/users/davidmoten/gists{/gist_id}","starred_url":"https://api.github.com/users/davidmoten/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davidmoten/subscriptions","organizations_url":"https://api.github.com/users/davidmoten/orgs","repos_url":"https://api.github.com/users/davidmoten/repos","events_url":"https://api.github.com/users/davidmoten/events{/privacy}","received_events_url":"https://api.github.com/users/davidmoten/received_events","type":"User","site_admin":false},"created_at":"2015-06-12T12:01:42Z","updated_at":"2015-06-12T12:01:42Z","body":"I'll mention a concrete example that is similar to what I encounter in my work. I have a stream of timestamped position+speed reports from a ship and from that stream I want to emit only those reports where the vessel was travelling slowly but not stopped for at least an hour. "}}