Reactor TraderService Example

TraderServerExample.kt

package reactor.quickstart

import org.slf4j.LoggerFactory
import java.util.concurrent.CountDownLatch
import kotlin.properties.Delegates
import reactor.core.Environment
import reactor.core.spec.Reactors
import reactor.event.selector.Selectors
import reactor.event.Event
import java.util.concurrent.TimeUnit

/**
 * Created by IntelliJ IDEA.
 * @author Mario Arias
 * Date: 17/04/14
 * Time: 23:07
 */

private val log = LoggerFactory.getLogger("Quickstart")!!
private var latch: CountDownLatch by Delegates.notNull()
private val totalTrades = 10000000

fun main(args: Array<String>) {
    val env = Environment()
    val server = TradeServer()

    val reactor = Reactors.reactor()!!
            .env(env)!!
            .dispatcher("ringBuffer")!!
            .get()!!

    val topic = "trade.execute"

    reactor.on(Selectors.`object`(topic)) {(tradeEvent: Event<Trade>?) ->
        server.execute(tradeEvent!!.getData())
        latch.countDown()
    }

    timed {
        (0..totalTrades).forEach {
            val t = server.nextTrade()!!
            reactor.notify(topic, t.eventWrap())
        }
    }

    server.stop()

}

private inline fun timed(body: () -> Unit) {
    log.info("Starting throughput test with {} trades...", totalTrades)
    latch = CountDownLatch(totalTrades)
    val startTime = System.currentTimeMillis()
    body()
    latch.await(30, TimeUnit.SECONDS)
    val endTime = System.currentTimeMillis()
    val elapsed = (endTime - startTime) * 1.0
    val throughput = totalTrades / (elapsed / 1000)
    log.info("Executed {} trades/sec in {}ms", throughput.toInt(), elapsed.toInt())
}

fun <T> T.eventWrap(): Event<T> {
    return Event.wrap(this)!!
}