title: How can I use TestScheduler with RxNetty http server
body: <p>I'm trying to create a simulation server to help make it easier to write tests for a client that operates on an event stream.</p>
  
  <p>I'm implementing the server and client with RxNetty, my question is around how I can use a TestScheduler with RxNetty to control when events are emitted.</p>
  
  <p>Here is my (simplified) server:</p>
  
  <pre><code>    final Observable&lt;Event&gt; events = Observable.just(...);
      final TestScheduler testScheduler = new TestScheduler();
      final HttpServer&lt;ByteBuf, ByteBuf&gt; server = RxNetty.createHttpServer(0, (request, response) -&gt; {
  
          // [snip] validate request...
  
          return request.getContent().flatMap(buf -&gt;
              events
                  .zipWith(Observable.interval(10, TimeUnit.MILLISECONDS, testScheduler), (e, t) -&gt; e)
                  .map(Encoder::eventToBytes)
                  .flatMap(response::writeBytesAndFlush)
          );
      });
      server.start();
  </code></pre>
  
  <p>However when I specify that <code>interval</code> should use <code>testScheduler</code> the server never sends any events. If I remove <code>testScheduler</code> or either <code>Schedulers.computation()</code> or <code>Schedulers.io()</code> the server is able to send events.</p>
  
  <p>Here is the client/usage side that is expecting to receive the events coming from the server:</p>
  
  <pre><code>    final int serverPort = server.getServerPort();
      final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", serverPort));
      final HttpClient&lt;ByteBuf, ByteBuf&gt; client = RxNetty.&lt;ByteBuf, ByteBuf&gt;newHttpClientBuilder(uri.getHost(), uri.getPort())
          .pipelineConfigurator(new HttpClientPipelineConfigurator&lt;&gt;())
          .build();
  
      final Event event = new Event("some event")
  
      final TestScheduler clientReceiveScheduler = new TestScheduler();
      final HttpClientRequest&lt;ByteBuf&gt; request = HttpClientRequest.createPost(uri.getPath())
          .withHeader("Content-Type", "application/x-protobuf")
          .withHeader("Accept", "application/x-protobuf")
          .withContent(event.toByteArray());
  
      final Observable&lt;byte[]&gt; observable = client.submit(request)
          .flatMap(AbstractHttpContentHolder::getContent)
          .map(BufUtils::bufToBytes)
          .zipWith(Observable.interval(10, TimeUnit.MILLISECONDS, clientReceiveScheduler), (b, l) -&gt; b)
          .doOnNext((b) -&gt; LOGGER.info("Received bytes: {}", Arrays.toString(b)));
  
      final TestSubscriber&lt;Event&gt; sub = new TestSubscriber&lt;&gt;();
      observable.subscribe(sub);
  
      testScheduler.advanceTimeBy(10, TimeUnit.MILLISECONDS);
      clientReceiveScheduler.advanceTimeBy(10, TimeUnit.MILLISECONDS);
      sub.getOnNextEvents()
          .stream()
          .forEach(System.out::println);
  </code></pre>
  
  <p>Here is a link to my <a href="https://github.com/BenWhitehead/mesos-rxjava/blob/scheduler-simulation/mesos-rxjava-test/src/main/java/com/mesosphere/mesos/rx/java/test/simulation/MesosSchedulerSimulation.java#L242" rel="nofollow">full server</a> and a <a href="https://github.com/BenWhitehead/mesos-rxjava/blob/scheduler-simulation/mesos-rxjava-test/src/test/java/com/mesosphere/mesos/rx/java/test/simulation/MesosSchedulerSimulationStreamTest.java#L78" rel="nofollow">test case</a> trying to use the server.</p>