Unit testing an observable service with Ratpack.

gistfile1.groovy

import ratpack.exec.ExecControl
import ratpack.exec.ExecController
import ratpack.launch.LaunchConfigBuilder
import spock.lang.AutoCleanup
import spock.lang.Specification
import spock.util.concurrent.BlockingVariable

import static ratpack.rx.RxRatpack.observeEach

class ExecControlSpec extends Specification {

  static class Service {

    def generators = [] // filled with biz logic closures which return Iterable<String>
    ExecControl execControl

    public Service(ExecControl execControl) {
      this.execControl = execControl
    }

    rx.Observable<String> doWork(String string) {
      rx.Observable.merge(generators.collect({ generator ->
        observeEach(execControl.blocking {
          generator(string)
        })
      }))
    }
  }

  @AutoCleanup ExecController execController = LaunchConfigBuilder.noBaseDir().build().execController

  def "test service"() {
    def service = new Service(execController.control)
    service.generators << { String it -> it.split(",").toList() }
    service.generators << { String it -> it.split(",").toList()*.toUpperCase() }
    def result = new BlockingVariable<List<String>>(5)

    when:
    execController.start {
      service.doWork("a,b").toList().subscribe {
        result.set(it)
      }
    }

    then:
    result.get().sort() == ["A", "B", "a", "b"]
  }
}