@Test public void testDodgyMessageProcessing() throws InterruptedException { DummyDatafeedApi df = new DummyDatafeedApi() { public V4EventList v4DatafeedIdReadGet(String id, String sessionToken, String keyManagerToken, Integer limit) { 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 -> { if (v.getId().contains("5")) { throw new IllegalArgumentException("Couldn't process: "+v.getId()); } else { events.add(v); } }, e -> exceptions.add(e), true); Thread.sleep(200); ssh.stop(); Assert.assertEquals(869, df.currentId); Assert.assertTrue(events.size() > 20); Assert.assertFalse(exceptions.isEmpty()); }
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; }