concurent_vertx_consumer.groovy

def executor = new ThreadPoolExecutor(
        cores, cores,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<Runnable>(1000))

DefaultFiberNode addFiber(OperationDefinition fiberDefinition) {
    RxHelper.toObservable(vertx.eventBus().consumer(fiberDefinition.address()))
        .flatMap({ data ->
            Observable.just(data)
                .subscribeOn(Schedulers.from(executor))
                .doOnNext({
                    def msg = it as Message
                    try {
                        fiberDefinition.handler().handle(new VertxOperationContext(msg, this))
                    } catch (Exception e) {
                        msg.fail(100, e.message)
                    }
                })
        }, cores).subscribe()
    this
}