@Test public void testStreamDying() throws InterruptedException { DummyDatafeedApi df = new DummyDatafeedApi() { int readFrom = 0; public V4EventList v4DatafeedIdReadGet(String id, String sessionToken, String keyManagerToken, Integer limit) { if (readFrom == currentId) { throw new BadRequestException("broken"); } else { readFrom = currentId; V4EventList out = new V4EventList(); out.add(createEvent()); out.add(createEvent()); out.add(createEvent()); out.add(createEvent()); out.add(createEvent()); return out; } } }; SymphonyStreamHandler ssh = new SymphonyStreamHandler(df, v -> events.add(v), e -> exceptions.add(e), true); Thread.sleep(4000); ssh.stop(); Assert.assertTrue(df.currentId > 869); Assert.assertTrue(events.size() > 5); Assert.assertFalse(exceptions.isEmpty()); Assert.assertTrue(exceptions.get(0) instanceof BadRequestException); }
public void stop() { running = false; worker.shutdown(); runThread.interrupt(); }
SymphonyStreamHandler { public void stop() { running = false; worker.shutdown(); runThread.interrupt(); } }
SymphonyStreamHandler { public void stop() { running = false; worker.shutdown(); runThread.interrupt(); } SymphonyStreamHandler(DatafeedApi api, Consumer<V4Event> eventConsumer, Consumer<Exception> exceptionHandler, boolean start); }
SymphonyStreamHandler { public void stop() { running = false; worker.shutdown(); runThread.interrupt(); } SymphonyStreamHandler(DatafeedApi api, Consumer<V4Event> eventConsumer, Consumer<Exception> exceptionHandler, boolean start); void start(); void consumeLoop(String initialDatafeedId); void stop(); }
SymphonyStreamHandler { public void stop() { running = false; worker.shutdown(); runThread.interrupt(); } SymphonyStreamHandler(DatafeedApi api, Consumer<V4Event> eventConsumer, Consumer<Exception> exceptionHandler, boolean start); void start(); void consumeLoop(String initialDatafeedId); void stop(); static final long MIN_BACK_OFF_MS; static final long MAX_BACK_OFF_MS; }