### Question:
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; }### Answer:
@Test public void testGeneralOperation() throws InterruptedException { SymphonyStreamHandler ssh = new SymphonyStreamHandler(new DummyDatafeedApi(), v -> events.add(v), e -> exceptions.add(e), true); Thread.sleep(50); Assert.assertEquals(1, events.size()); Assert.assertEquals(0, exceptions.size()); ssh.stop(); }
@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()); }
@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); }
@Test(expected = BadRequestException.class) public void testSymphonyDeadOnStartup() { DummyDatafeedApi df = new DummyDatafeedApi() { @Override public Datafeed v4DatafeedCreatePost(String sessionToken, String keyManagerToken) { throw new BadRequestException(); } }; SymphonyStreamHandler ssh = new SymphonyStreamHandler(df, v -> events.add(v), e -> exceptions.add(e), true); ssh.stop(); }
@Test public void testBackoffExceptionSpam() throws InterruptedException { DummyDatafeedApi df = new DummyDatafeedApi() { public V4EventList v4DatafeedIdReadGet(String id, String sessionToken, String keyManagerToken, Integer limit) { throw new BadRequestException("broken"); } }; SymphonyStreamHandler ssh = new SymphonyStreamHandler(df, v -> events.add(v), e -> exceptions.add(e), true); Thread.sleep(10000); ssh.stop(); Assert.assertTrue(exceptions.size() == 3); }