@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("^-----!-"); }
@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);   }