StockQuoteServer extends RxNettyEventServer<Quote> { @Override protected Observable<Quote> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return stockQuoteEventStreamClient .readServerSideEvents() .map(Quote::fromJson); } StockQuoteServer(int port,
                            EventStreamClient stockQuoteEventStreamClient,
                            EventStreamClient forexEventStreamClient,
                            Scheduler scheduler);   }
@Test @Ignore public void should_filter_quotes_for_requested_stock() { Observable<String> quoteSource = hot("--(ga)--", of("g", new Quote("GOOGL", 705.8673).toJson(), "a", new Quote("APPLE", 98.18).toJson())); Observable<String> forexSource = hot("--f--", of("f", new Quote("EUR/USD", 1).toJson())); StockQuoteServer stockQuoteServer = createServer(quoteSource, forexSource); HttpRequest request = createRequest("code", "GOOGL"); expectObservable(stockQuoteServer.getEvents(request).map(q -> q.code)) .toBe("--v--", of("v", "GOOGL")); }
@Test @Ignore public void should_generate_one_quote_in_euro_for_one_quote_in_dollar() { Observable<String> quoteSource = hot("--g----", of("g", new Quote("GOOGL", 1300).toJson())); Observable<String> forexSource = hot("--f-x--", of("f", new Quote("EUR/USD", 1.3).toJson(), "x", new Quote("EUR/USD", 1.4).toJson())); StockQuoteServer stockQuoteServer = createServer(quoteSource, forexSource); HttpRequest request = createRequest("code", "GOOGL"); expectObservable(stockQuoteServer.getEvents(request).map(q -> q.quote)) .toBe("--v----", of("v", 1000d)); }
@Test @Ignore public void should_generate_quotes_in_euro_using_latest_known_foreign_exchange_rate() { Observable<String> quoteSource = hot("----g-", of("g", new Quote("GOOGL", 1300).toJson())); Observable<String> forexSource = hot("-f-x--", of("f", new Quote("EUR/USD", 1.2).toJson(), "x", new Quote("EUR/USD", 1.3).toJson())); StockQuoteServer stockQuoteServer = createServer(quoteSource, forexSource); HttpRequest request = createRequest("code", "GOOGL"); expectObservable(stockQuoteServer.getEvents(request).map(q -> q.quote)) .toBe("----v-", of("v", 1000d)); }
@Test @Ignore public void should_unsubscribe_to_forex_stream_when_unscribing_to_quote() { Observable<String> quoteSource = hot("----g---", of("g", new Quote("GOOGL", 1300).toJson())); HotObservable<String> forexSource = hot("---f----", of("f", new Quote("EUR/USD", 1.2).toJson())); Observable<String> stopSource = hot("------s-"); StockQuoteServer stockQuoteServer = createServer(quoteSource, forexSource); HttpRequest request = createRequest("code", "GOOGL"); stockQuoteServer.getEvents(request).takeUntil(stopSource).subscribe(); expectSubscriptions(forexSource.getSubscriptions()) .toBe("^-----!-"); }
@Test @Ignore public void should_send_an_error_when_no_forex_data_after_five_seconds() { Observable<String> quoteSource = hot("-g-----", of("g", new Quote("GOOGL", 1300).toJson())); Observable<String> forexSource = hot("-------"); StockQuoteServer stockQuoteServer = createServer(quoteSource, forexSource); HttpRequest request = createRequest("code", "GOOGL"); expectObservable(stockQuoteServer.getEvents(request)) .toBe("------#"); }