StockServer extends RxNettyEventServer<Stock> { @Override protected Observable<Stock> getEvents(HttpRequest request) { return Observable.never(); } StockServer(int port,
                       RequestReplyClient stockClient,
                       EventStreamClient quoteEventStreamClient,
                       Scheduler scheduler);   }
@Test @Ignore public void should_send_a_stock_message_when_receiving_a_quote() { Observable<String> quoteSource = hot("--q--", of("q", new Quote("GOOGL", 705.8673).toJson())); StockServer server = createServer(quoteSource); expectObservable(server.getEvents(null).map(s -> s.companyName).concatWith(Observable.never())) .toBe("--s--", of("s", "Alphabet Inc")); }
@Test @Ignore public void should_send_a_stock_message_only_once_when_receiving_two_quotes_for_the_same_stock() { Observable<String> quoteSource = hot("--f-s-t--", of("f", new Quote("GOOGL", 705.8673).toJson(), "s", new Quote("GOOGL", 705.8912).toJson(), "t", new Quote("IBM", 106.344).toJson() )); StockServer server = createServer(quoteSource); expectObservable(server.getEvents(null).map(s -> s.companyName).concatWith(Observable.never())) .toBe("--g---i--", of("g", "Alphabet Inc", "i", "International Business Machines Corp.")); }
@Test @Ignore public void should_stop_stream_after_10_seconds() { Observable<String> quoteSource = Observable.never(); StockServer server = createServer(quoteSource); expectObservable(server.getEvents(null)).toBe("----------|"); }