SlackBackfillSpout extends BaseRichSpout { @Override public void nextTuple() { Optional<Interval> optionalInterval = getRunInterval(); if (!optionalInterval.isPresent()) { try { LOG.info("Waiting for a few more minutes to go by. Granularity is {}", granularityMins); Thread.sleep(granularityMins * 60 * 1000); } catch (InterruptedException e) { LOG.warn("Interrupted while sleeping..."); } return; } Interval runInterval = optionalInterval.get(); LOG.info("Running with the following interval {}", runInterval); Map<String, User> users = slackDao.getUsers(); Map<String, Room> rooms = slackDao.getRooms(); backfillRooms(users, rooms, runInterval); dbDao.setLastMessagePullTime(runInterval.getEnd()); }  @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context,
                     SpoutOutputCollector collector); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer fields); @Override void close(); static final String SPOUT_ID; static final String BACKFILL_SLACK_MESSAGE_FIELD_STR; }
@Test public void testNextTuple() { chatConfig.granularityMins = 0; stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); IChatAlyticsDAO dbDao = mock(IChatAlyticsDAO.class); IChatApiDAO slackDao = mock(IChatApiDAO.class); underTest.open(chatConfig, slackDao, dbDao, context, collector); Map<String, User> users = ImmutableMap.of("u1", new User("u1", "email", false, false, false, "name", "mention_name", null, DateTime.now(), DateTime.now(), null, null, null, null)); Map<String, Room> rooms = ImmutableMap.of("r1", new Room("r1", "room", null, DateTime.now(), DateTime.now(), null, false, false, null, null)); when(dbDao.getLastMessagePullTime()).thenReturn(new DateTime(0, DateTimeZone.UTC)); when(slackDao.getUsers()).thenReturn(users); when(slackDao.getRooms()).thenReturn(rooms); underTest.nextTuple(); verify(slackDao).getUsers(); verify(slackDao).getRooms(); verify(dbDao).setLastMessagePullTime(any(DateTime.class)); verify(dbDao).getLastMessagePullTime(); verifyNoMoreInteractions(dbDao); }