{"action":"created","issue":{"url":"https://api.github.com/repos/ratpack/ratpack/issues/682","labels_url":"https://api.github.com/repos/ratpack/ratpack/issues/682/labels{/name}","comments_url":"https://api.github.com/repos/ratpack/ratpack/issues/682/comments","events_url":"https://api.github.com/repos/ratpack/ratpack/issues/682/events","html_url":"https://github.com/ratpack/ratpack/issues/682","id":80914910,"number":682,"title":"Unexpected exception when uses publisher with async subscription process.","user":{"login":"niwinz","id":843689,"avatar_url":"https://avatars.githubusercontent.com/u/843689?v=3","gravatar_id":"","url":"https://api.github.com/users/niwinz","html_url":"https://github.com/niwinz","followers_url":"https://api.github.com/users/niwinz/followers","following_url":"https://api.github.com/users/niwinz/following{/other_user}","gists_url":"https://api.github.com/users/niwinz/gists{/gist_id}","starred_url":"https://api.github.com/users/niwinz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/niwinz/subscriptions","organizations_url":"https://api.github.com/users/niwinz/orgs","repos_url":"https://api.github.com/users/niwinz/repos","events_url":"https://api.github.com/users/niwinz/events{/privacy}","received_events_url":"https://api.github.com/users/niwinz/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2015-05-26T11:12:00Z","updated_at":"2015-05-27T04:01:38Z","closed_at":null,"body":"Hi!\r\n\r\nI found an unexpected behavior of ratpack when I send a response using reactive streams publisher with asynchronous subscribe process. \r\n\r\nWhen I use publishers that executes the .onSubscribe syncronously (in current calling thread) everything works as expected, but if a publisher executes the .onSubscribe method asynchronously (in a separated thread pool) it raises a unexpected exception and return unexpected response.\r\n\r\nI have taken the example implementaton of AsyncIteratorPublisher from reactive streams examples repository and I have made an example project for the demostration of the issue. (I'm not very familiar with spock and in general testing in java and testing guidelines of ratpack so for me is more easy\r\ngive a little project for reproduce the error, sorry if this is not a good way to contribute).\r\n\r\nThis is a link to the project: https://github.com/niwinz/ratpack-playground/tree/stream-async-subscribe\r\n\r\nJust run `./gradlew run` and access to `http://localhost:5050/`\r\n\r\nIn the console should be printed exception like that one:\r\n\r\n```\r\n[main] INFO ratpack.server.RatpackServer - Ratpack started for http://localhost:5050\r\n[ratpack-compute-1-2] WARN ratpack.server.internal.NettyHandlerAdapter - No response sent for GET request to / (last handler: my.app.Main$MyHandler)\r\n[ratpack-compute-1-2] WARN ratpack.server.internal.DefaultResponseTransmitter - attempt at double transmission for: /\r\nratpack.util.internal.InternalRatpackError: Unexpected internal error: \r\n        at ratpack.server.internal.DefaultResponseTransmitter.pre(DefaultResponseTransmitter.java:93)\r\n        at ratpack.server.internal.DefaultResponseTransmitter.transmit(DefaultResponseTransmitter.java:104)\r\n        at ratpack.server.internal.DefaultResponseTransmitter.transmit(DefaultResponseTransmitter.java:100)\r\n        at ratpack.http.internal.DefaultResponse.commit(DefaultResponse.java:319)\r\n        at ratpack.http.internal.DefaultResponse.send(DefaultResponse.java:266)\r\n        at ratpack.http.internal.DefaultResponse.send(DefaultResponse.java:242)\r\n        at ratpack.server.internal.NettyHandlerAdapter.lambda$channelRead0$178(NettyHandlerAdapter.java:150)\r\n        at ratpack.server.internal.NettyHandlerAdapter$$Lambda$36/1372844603.execute(Unknown Source)\r\n        at ratpack.exec.internal.ExecutionBacking.done(ExecutionBacking.java:210)\r\n        at ratpack.exec.internal.ExecutionBacking.drain(ExecutionBacking.java:170)\r\n        at ratpack.exec.internal.ExecutionBacking.<init>(ExecutionBacking.java:69)\r\n        at ratpack.exec.internal.DefaultExecControl$1.start(DefaultExecControl.java:118)\r\n        at ratpack.handling.internal.DefaultContext.start(DefaultContext.java:148)\r\n        at ratpack.server.internal.NettyHandlerAdapter.channelRead0(NettyHandlerAdapter.java:126)\r\n        at ratpack.server.internal.NettyHandlerAdapter.channelRead0(NettyHandlerAdapter.java:51)\r\n        at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)\r\n        at ratpack.server.internal.NettyHandlerAdapter.channelRead(NettyHandlerAdapter.java:88)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at ratpack.server.internal.NettyRatpackServer$ReloadHandler.delegate(NettyRatpackServer.java:387)\r\n        at ratpack.server.internal.NettyRatpackServer$ReloadHandler.channelRead0(NettyRatpackServer.java:365)\r\n        at ratpack.server.internal.NettyRatpackServer$ReloadHandler.channelRead0(NettyRatpackServer.java:320)\r\n        at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)\r\n        at io.netty.handler.codec.MessageToMessageCannelRead(MessageToMessageCodec.java:111)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)\r\n        at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)\r\n        at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)\r\n        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:187)\r\n        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:947)\r\n        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:127)\r\n        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:510)\r\n        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:467)\r\n        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:381)\r\n        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:353)\r\n        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:703)\r\n        at ratpack.exec.internal.DefaultExecController$ExecControllerBindingThreadFactory.lambda$newThread$63(DefaultExecController.java:82)\r\n        at ratpack.exec.internal.DefaultExecController$ExecControllerBindingThreadFactory$$Lambda$30/146370526.run(Unknown Source)\r\n    io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)\r\n        at java.lang.Thread.run(Thread.java:745)\r\n```\r\n\r\nAnd in the browser, a 500 response will be returned with content of the stream. \r\n\r\nThat behavior is using ratpack 0.9.15.\r\n\r\nWith 0.9.16, the printed exception is:\r\n```\r\nratpack.util.internal.InternalRatpackError: Unexpected internal error: \r\n        at ratpack.server.internal.DefaultResponseTransmitter.pre(DefaultResponseTransmitter.java:93)\r\n        at ratpack.server.internal.DefaultResponseTransmitter.access$200(DefaultResponseTransmitter.java:44)\r\n        at ratpack.server.internal.DefaultResponseTransmitter$1.onSubscribe(DefaultResponseTransmitter.java:178)\r\n        at ratpack.stream.internal.MapPublisher$1.onSubscribe(MapPublisher.java:47)\r\n        at my.app.AsyncIterablePublisher$SubscriptionImpl.doSubscribe(AsyncIterablePublisher.java:124)\r\n        at my.app.AsyncIterablePublisher$SubscriptionImpl.run(AsyncIterablePublisher.java:215)\r\n        at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)\r\n        at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)\r\n        at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)\r\n        at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1689)\r\n        at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)\r\n```\r\n\r\nAnd the browser does not returns any response.\r\n"},"comment":{"url":"https://api.github.com/repos/ratpack/ratpack/issues/comments/105746973","html_url":"https://github.com/ratpack/ratpack/issues/682#issuecomment-105746973","issue_url":"https://api.github.com/repos/ratpack/ratpack/issues/682","id":105746973,"user":{"login":"alkemist","id":17825,"avatar_url":"https://avatars.githubusercontent.com/u/17825?v=3","gravatar_id":"","url":"https://api.github.com/users/alkemist","html_url":"https://github.com/alkemist","followers_url":"https://api.github.com/users/alkemist/followers","following_url":"https://api.github.com/users/alkemist/following{/other_user}","gists_url":"https://api.github.com/users/alkemist/gists{/gist_id}","starred_url":"https://api.github.com/users/alkemist/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alkemist/subscriptions","organizations_url":"https://api.github.com/users/alkemist/orgs","repos_url":"https://api.github.com/users/alkemist/repos","events_url":"https://api.github.com/users/alkemist/events{/privacy}","received_events_url":"https://api.github.com/users/alkemist/received_events","type":"User","site_admin":false},"created_at":"2015-05-27T04:01:38Z","updated_at":"2015-05-27T04:01:38Z","body":"Reopening and renaming to be about better diagnostics when trying to use an async stream without binding it to the execution."}}