build.gradle

buildscript {
    repositories {
        maven { url "http://oss.jfrog.org/repo" }
        mavenCentral()
    }
    dependencies {
        classpath 'io.ratpack:ratpack-gradle:0.9.7'
    }
}

repositories {
    maven { url "http://oss.jfrog.org/repo" }
    mavenCentral()
    maven { url "http://repo.springsource.org/repo" } // for springloaded
}

// The “ratpack” plugin applies the “application” plugin, making it easy to create a standalone application.
// See: http://gradle.org/docs/current/userguide/application_plugin.html
apply plugin: "io.ratpack.ratpack-groovy"

// The “ratpack” plugin is IDEA aware.
// It will create a run configuration in IDEA to launch your app in your IDE, with hot reloading.
apply plugin: "idea"
idea {
    project {
        jdkName "1.7"
        languageLevel "1.7"
    }
}

configurations.all {
    exclude module: "groovy"
}

dependencies {
    // SpringLoaded enables runtime hot reloading.
    // It is not part of the app runtime and is not shipped in the distribution.
    springloaded "org.springsource.springloaded:springloaded-core:1.1.4"
    testCompile "org.spockframework:spock-core:0.7-groovy-2.0", {
        exclude module: "groovy-all"
    }
    compile 'io.ratpack:ratpack-rx:0.9.7'
    compile 'com.ning:async-http-client:1.8.13'
}

ratpack.groovy

import ratpack.http.client.HttpClient

import javax.crypto.Cipher
import javax.crypto.spec.IvParameterSpec
import javax.crypto.spec.SecretKeySpec

import static ratpack.groovy.Groovy.ratpack
import static ratpack.rx.RxRatpack.*

ratpack {
    bindings {
        initialize()
    }
    handlers {
        println "startup"
        get("multi") { HttpClient httpClient ->
            // call "single" endpoint multiple times
            def count = 0
            rx.Observable.from(1..10000).
                    groupBy {
                        it % 14 // this number of parallel streams
                    }.
                    lift(forkOnNext(context)).
                    flatMap {
                        it.flatMap {
                            observe(httpClient.get {
                                it.url.set(new URI("http://127.0.0.1:5050/single"))
                            }).map {
                                it.body.text
                            }
                        }.
                                toList()
                    }.
                    flatMap {
                        rx.Observable.from(it)
                    }.
                    subscribe({
                        println "${++count}"
                    }, {
                        error(it)
                    }, {
                        println "complete! - emitted $count times"
                        render "complete! - emitted $count times"
                    })
        }

        get("single") {
            observe(blocking({
                // simulate some blocking load
                Cipher cipher = Cipher.getInstance("AES/CBC/NoPadding", "SunJCE");
                SecretKeySpec key = new SecretKeySpec("0123456789abcdef".bytes, "AES");
                cipher.init(Cipher.ENCRYPT_MODE, key, new IvParameterSpec("AAAAAAAAAAAAAAAA".bytes));
                6400.times { cipher.update(it.byteValue()) }
                cipher.doFinal("0123456789ABCDEF".bytes).toString()
            })).subscribe({
                render "ok $it"
            })
        }
    }

}