{"action":"created","issue":{"url":"https://api.github.com/repos/akka/akka/issues/19256","labels_url":"https://api.github.com/repos/akka/akka/issues/19256/labels{/name}","comments_url":"https://api.github.com/repos/akka/akka/issues/19256/comments","events_url":"https://api.github.com/repos/akka/akka/issues/19256/events","html_url":"https://github.com/akka/akka/issues/19256","id":123412904,"number":19256,"title":"InputStreamPublisher should immediately emit data only when there's available demand","user":{"login":"fsat","id":1665611,"avatar_url":"https://avatars.githubusercontent.com/u/1665611?v=3","gravatar_id":"","url":"https://api.github.com/users/fsat","html_url":"https://github.com/fsat","followers_url":"https://api.github.com/users/fsat/followers","following_url":"https://api.github.com/users/fsat/following{/other_user}","gists_url":"https://api.github.com/users/fsat/gists{/gist_id}","starred_url":"https://api.github.com/users/fsat/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fsat/subscriptions","organizations_url":"https://api.github.com/users/fsat/orgs","repos_url":"https://api.github.com/users/fsat/repos","events_url":"https://api.github.com/users/fsat/events{/privacy}","received_events_url":"https://api.github.com/users/fsat/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/akka/akka/labels/bug","name":"bug","color":"e11d21"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":7,"created_at":"2015-12-22T05:06:28Z","updated_at":"2015-12-22T13:00:59Z","closed_at":null,"body":"I have encountered the following exception when using `StreamConverters.fromInputStream`:\r\n\r\n```\r\n2015-12-22T04:51:10Z 6135c7857c85 ERROR ActorSystemImpl ActorSystem(conductr) - Outgoing response stream error\r\njava.lang.IllegalStateException: onNext is not allowed when the stream has not requested elements, totalDemand was 0\r\n\tat akka.stream.actor.ActorPublisher$class.onNext(ActorPublisher.scala:182) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher.onNext(InputStreamPublisher.scala:30) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher.readAndEmit(InputStreamPublisher.scala:67) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher.readAndSignal(InputStreamPublisher.scala:49) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher$$anonfun$receive$1.applyOrElse(InputStreamPublisher.scala:43) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:467) ~[com.typesafe.akka.akka-actor_2.11-2.3-bin-rp-15v09p03i03.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher.akka$stream$actor$ActorPublisher$$super$aroundReceive(InputStreamPublisher.scala:30) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.actor.ActorPublisher$class.aroundReceive(ActorPublisher.scala:309) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.stream.impl.io.InputStreamPublisher.aroundReceive(InputStreamPublisher.scala:30) ~[com.typesafe.akka.akka-stream-experimental_2.11-2.0.jar:na]\r\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:520) ~[com.typesafe.akka.akka-actor_2.11-2.3-bin-rp-15v09p03i03.jar:na]\r\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:489) ~[com.typesafe.akka.akka-actor_2.11-2.3-bin-rp-15v09p03i03.jar:na]\r\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) ~[com.typesafe.akka.akka-actor_2.11-2.3-bin-rp-15v09p03i03.jar:na]\r\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220) ~[com.typesafe.akka.akka-actor_2.11-2.3-bin-rp-15v09p03i03.jar:na]\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_66-internal]\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0_66-internal]\r\n\tat java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_66-internal]\r\n```\r\n\r\nI believe `InputStreamPublisher` need to read & emit when there's a demand."},"comment":{"url":"https://api.github.com/repos/akka/akka/issues/comments/166611291","html_url":"https://github.com/akka/akka/issues/19256#issuecomment-166611291","issue_url":"https://api.github.com/repos/akka/akka/issues/19256","id":166611291,"user":{"login":"sirthias","id":135347,"avatar_url":"https://avatars.githubusercontent.com/u/135347?v=3","gravatar_id":"","url":"https://api.github.com/users/sirthias","html_url":"https://github.com/sirthias","followers_url":"https://api.github.com/users/sirthias/followers","following_url":"https://api.github.com/users/sirthias/following{/other_user}","gists_url":"https://api.github.com/users/sirthias/gists{/gist_id}","starred_url":"https://api.github.com/users/sirthias/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sirthias/subscriptions","organizations_url":"https://api.github.com/users/sirthias/orgs","repos_url":"https://api.github.com/users/sirthias/repos","events_url":"https://api.github.com/users/sirthias/events{/privacy}","received_events_url":"https://api.github.com/users/sirthias/received_events","type":"User","site_admin":false},"created_at":"2015-12-22T13:00:59Z","updated_at":"2015-12-22T13:00:59Z","body":"Ok. Yeah, I think this shouldn't be hard to dig up and fix.\r\nIt'll be interesting to see why this wasn't caught so far. Maybe we need a test serving some larger static input. I'll submit a a PR with a reproducer later today."}}