ForexServer extends RxNettyEventServer<Double> { @Override protected Observable<Double> getEvents(HttpRequest request) { return Observable.never(); } ForexServer(int port, EventStreamClient eventStreamClient);   }
@Test @Ignore public void should_forward_forex_data() { Observable<String> forexSource = hot("--f--", of("f", new Quote("EUR/USD", 1.4).toJson())); ForexServer forexServer = create(forexSource); expectObservable(forexServer.getEvents(null).take(1)) .toBe("--(v|)", of("v", 1.4)); }
@Test @Ignore public void should_forward_only_one_forex_data() { Observable<String> forexSource = hot("--f-x-", of("f", new Quote("EUR/USD", 1.4).toJson(), "x", new Quote("EUR/USD", 1.5).toJson())); ForexServer forexServer = create(forexSource); expectObservable(forexServer.getEvents(null)) .toBe("--(v|)", of("v", 1.4)); }