Ratpack HttpClient StreamedResponse read timeout not propagating to user code so promise doesn't complete.

build.gradle

group = 'jamesl'
version = '1.0-SNAPSHOT'

buildscript {
	repositories {
		jcenter()
	}
	dependencies {
		classpath 'io.ratpack:ratpack-gradle:1.4.4'
		classpath 'com.github.jengelman.gradle.plugins:shadow:1.2.3'
	}
}

apply plugin: 'io.ratpack.ratpack-groovy'
apply plugin: 'com.github.johnrengelman.shadow'
apply plugin: 'maven'

repositories {
	jcenter()
    mavenLocal()
}

dependencies {
    compile 'io.projectreactor:reactor-core:3.0.3.RELEASE'
    compile 'org.slf4j:slf4j-simple:1.7.21'
	testCompile 'org.spockframework:spock-core:1.0-groovy-2.4'
}

run {
	systemProperty 'io.netty.leakDetectionLevel', 'PARANOID'
    systemProperty 'Log4jContextSelector', 'org.apache.logging.log4j.core.async.AsyncLoggerContextSelector'
}

shadowJar {
	classifier = "all"
}

task wrapper(type: Wrapper) {
	gradleVersion = '3.2'
}

HttpClientReadTimeoutSpec.groovy

package jamesl

import groovy.util.logging.Slf4j
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufAllocator
import io.netty.buffer.UnpooledByteBufAllocator
import ratpack.groovy.test.embed.GroovyEmbeddedApp
import ratpack.http.client.HttpClient
import ratpack.http.client.HttpClientReadTimeoutException
import ratpack.test.CloseableApplicationUnderTest
import ratpack.test.exec.ExecHarness
import reactor.core.publisher.Flux
import spock.lang.Specification

import java.nio.charset.StandardCharsets
import java.time.Duration
import java.util.function.Function

/**
 * @author jamesl
 */
@Slf4j
class HttpClientReadTimeoutSpec extends Specification {
    ByteBufAllocator allocator
    CloseableApplicationUnderTest app
    ExecHarness exec

    def setup() {
        allocator = UnpooledByteBufAllocator.DEFAULT
        app = GroovyEmbeddedApp.fromHandler { context ->
            Function<reactor.util.function.Tuple2<Long, String>, String> f = { p ->
                p.t2
            }

            Function<String, ByteBuf> f2 = { s ->
                log.info("s={}", s)
                allocator.buffer().writeBytes(s.bytes)
            }

            def words = Flux.just("hello", "james")
            def publisher = Flux.interval(Duration.ofMillis(200), Duration.ofSeconds(2))
            .zipWith(words)
            .map(f)
            .map(f2)

            context.response.sendStream(publisher)
        }

        exec = ExecHarness.harness()
    }

    def cleanup() {
        exec.close()
        app.close()
    }

    def "http client read timeout is propagated with ReceivedResponse"() {
        when:
        def result = exec.yield { e ->
            http(Duration.ofMillis(800)).get(app.address)
            .map { response ->
                response.body.text
            }
        }

        then:
        result.error
        result.throwable instanceof HttpClientReadTimeoutException
    }

    def "http client read timeout is propagated with StreamedResponse"() {
        when:
        def result = exec.yield { e ->
            http(Duration.ofMillis(800)).requestStream(app.address) { spec ->
            }
            .wiretap { r ->
                log.info("r={}", r)
            }
            .flatMap { response ->
                response.body.reduce(allocator.compositeBuffer()) { x, buffer ->
                    x.addComponent(true, buffer)
                }
                .map { x ->
                    x.toString(StandardCharsets.UTF_8)
                }
            }
        }

        then:
        result.error
        result.throwable instanceof HttpClientReadTimeoutException
    }

    /**
     *
     * @param readTimeout
     * @return
     */
    HttpClient http(Duration readTimeout) {
        HttpClient.of { spec ->
            spec.byteBufAllocator(allocator)
            spec.readTimeout(readTimeout)
        }
    }
}

Ratpack (1.4.4) HttpClient#requestStream does not propagate read timeout correctly whereas HttpClient#get does.

It appears that the ReadTimeoutException is thrown but is swallowed by the redirect handler because the promise has already been completed - ReceivedResponse completes when the full response has been received whereas StreamedResponse completes when the http response header is received.