{"action":"created","issue":{"url":"https://api.github.com/repos/reactor/reactor/issues/500","labels_url":"https://api.github.com/repos/reactor/reactor/issues/500/labels{/name}","comments_url":"https://api.github.com/repos/reactor/reactor/issues/500/comments","events_url":"https://api.github.com/repos/reactor/reactor/issues/500/events","html_url":"https://github.com/reactor/reactor/issues/500","id":92886475,"number":500,"title":"Stream.flatMap() creates a exponentially increasing demand when defining a capacity.  ","user":{"login":"nitzanvolman","id":4519219,"avatar_url":"https://avatars.githubusercontent.com/u/4519219?v=3","gravatar_id":"","url":"https://api.github.com/users/nitzanvolman","html_url":"https://github.com/nitzanvolman","followers_url":"https://api.github.com/users/nitzanvolman/followers","following_url":"https://api.github.com/users/nitzanvolman/following{/other_user}","gists_url":"https://api.github.com/users/nitzanvolman/gists{/gist_id}","starred_url":"https://api.github.com/users/nitzanvolman/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nitzanvolman/subscriptions","organizations_url":"https://api.github.com/users/nitzanvolman/orgs","repos_url":"https://api.github.com/users/nitzanvolman/repos","events_url":"https://api.github.com/users/nitzanvolman/events{/privacy}","received_events_url":"https://api.github.com/users/nitzanvolman/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-07-03T14:30:03Z","updated_at":"2015-07-03T16:34:47Z","closed_at":null,"body":"\r\nconsider the code below:\r\nwhen commenting out  ``.capacity(batchSize)`` the ``BiConsumer`` receives exactly one call with a demand of the ``numOfItems`` - as expected. \r\nbut when running this code with ``.capacity(batchSize)`` it seams like the ``BiConsumer`` gets called several times with a demand that grows exponentially on each subsequent request. \r\nMy expectation is that the demand will always be less or equal to the ``batchSize``\r\n\r\nthis is probably related to [this issue](https://github.com/reactor/reactor/issues/491) but i recreated this on both 2.0.3 and on the latest 2.1.0 snapshot, which should contain the  [fix](https://github.com/reactor/reactor/commit/71b1a76157c75edede7d85b09cb5a308d0333b63)\r\n\r\n```java\r\npackage fizbuz;\r\n\r\nimport reactor.Environment;\r\nimport reactor.rx.Streams;\r\n\r\nimport java.util.Timer;\r\nimport java.util.TimerTask;\r\nimport java.util.concurrent.atomic.AtomicLong;\r\n\r\npublic class FizBuzDemandBug {\r\n\r\n\r\n    public static void main(String[] args) {\r\n        int numOfItems = 1024;\r\n        int batchSize = 8;\r\n        Environment.initializeIfEmpty();\r\n        final Timer timer = new Timer();\r\n        AtomicLong globalCounter = new AtomicLong();\r\n\r\n        Streams.createWith((demand, subscriber) -> {\r\n            System.out.println(\"demand is \" + demand);\r\n            if (!subscriber.isCancelled()) {\r\n                for (int i = 0; i < demand; i++) {\r\n                    long curr = globalCounter.incrementAndGet();\r\n                    if (curr % 5 == 0 && curr % 3 == 0) subscriber.onNext(\"FizBuz \\r\\n\");\r\n                    else if (curr % 3 == 0) subscriber.onNext(\"Fiz \");\r\n                    else if (curr % 5 == 0) subscriber.onNext(\"Buz \");\r\n                    else subscriber.onNext(String.valueOf(curr) + \" \");\r\n\r\n                    if (globalCounter.get() > numOfItems) {\r\n                        subscriber.onComplete();\r\n                        return;\r\n                    }\r\n                }\r\n            }\r\n        })\r\n        .flatMap((s) -> Streams.create((sub) -> {\r\n            timer.schedule(\r\n                    new TimerTask() {\r\n                        @Override\r\n                        public void run() {\r\n                            sub.onNext(s);\r\n                        }\r\n                    }, 10);\r\n        }))\r\n        .capacity(batchSize)\r\n//                .observe(System.out::print)\r\n        .consume(numOfItems)\r\n        ;\r\n\r\n\r\n    }\r\n\r\n}\r\n```\r\n\r\noutput:\r\n```\r\ndemand is 8\r\ndemand is 64\r\ndemand is 576\r\ndemand is 5184\r\n\r\n```\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/reactor/reactor/issues/comments/118388682","html_url":"https://github.com/reactor/reactor/issues/500#issuecomment-118388682","issue_url":"https://api.github.com/repos/reactor/reactor/issues/500","id":118388682,"user":{"login":"smaldini","id":76718,"avatar_url":"https://avatars.githubusercontent.com/u/76718?v=3","gravatar_id":"","url":"https://api.github.com/users/smaldini","html_url":"https://github.com/smaldini","followers_url":"https://api.github.com/users/smaldini/followers","following_url":"https://api.github.com/users/smaldini/following{/other_user}","gists_url":"https://api.github.com/users/smaldini/gists{/gist_id}","starred_url":"https://api.github.com/users/smaldini/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/smaldini/subscriptions","organizations_url":"https://api.github.com/users/smaldini/orgs","repos_url":"https://api.github.com/users/smaldini/repos","events_url":"https://api.github.com/users/smaldini/events{/privacy}","received_events_url":"https://api.github.com/users/smaldini/received_events","type":"User","site_admin":false},"created_at":"2015-07-03T16:34:47Z","updated_at":"2015-07-03T16:34:47Z","body":"I'm testing if there are any regression anyway by forcing passing demand N upstream. "}}