@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("------#"); }
@Override protected Observable<Quote> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return stockQuoteEventStreamClient .readServerSideEvents() .map(Quote::fromJson); }
StockQuoteServer extends RxNettyEventServer<Quote> { @Override protected Observable<Quote> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return stockQuoteEventStreamClient .readServerSideEvents() .map(Quote::fromJson); } }
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); }
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);  }
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);   }