@Test @Ignore public void should_generate_at_most_one_event_per_sec() { TestSubscriber<Vwap> testSubscriber = new TestSubscriber<>(); HttpRequest request = createRequest("code", "GOOGL"); vwapServer.getEvents(request).subscribe(testSubscriber); for (int i = 1; i < 11; i++) { tradeSourceSubject.onNext(new Trade("GOOGL", 10, 7000).toJson(), i * 50); } scheduler.advanceTimeBy(2, TimeUnit.SECONDS); List<Vwap> events = testSubscriber.getOnNextEvents(); assertThat(events).hasSize(1); Vwap vwap = events.get(0); assertThat(vwap.code).isEqualTo("GOOGL"); assertThat(vwap.vwap).isEqualTo(700); assertThat(vwap.volume).isEqualTo(100); }
@Override protected Observable<Vwap> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return Observable.never(); }
VwapServer extends RxNettyEventServer<Vwap> { @Override protected Observable<Vwap> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return Observable.never(); } }
VwapServer extends RxNettyEventServer<Vwap> { @Override protected Observable<Vwap> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return Observable.never(); } VwapServer(int port, EventStreamClient tradeEventStreamClient, Scheduler scheduler); }
VwapServer extends RxNettyEventServer<Vwap> { @Override protected Observable<Vwap> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return Observable.never(); } VwapServer(int port, EventStreamClient tradeEventStreamClient, Scheduler scheduler);  }
VwapServer extends RxNettyEventServer<Vwap> { @Override protected Observable<Vwap> getEvents(HttpRequest request) { String stockCode = request.getParameter("code"); return Observable.never(); } VwapServer(int port, EventStreamClient tradeEventStreamClient, Scheduler scheduler);   }