SlackBackfillSpout extends BaseRichSpout { @Override public void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector) { String configYaml = (String) conf.get(ConfigurationConstants.CHATALYTICS_CONFIG.txt); ChatAlyticsConfig config = YamlUtils.readChatAlyticsConfigFromString(configYaml); SlackBackfillerConfig chatConfig = (SlackBackfillerConfig) config.computeConfig.chatConfig; Preconditions.checkArgument(chatConfig.granularityMins >= 0, "Granularity has to be >= 0"); open(chatConfig, SlackApiDAOFactory.getSlackApiDao(config), ChatAlyticsDAOFactory.createChatAlyticsDao(config), context, collector); }  @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 testOpen() { chatConfig.granularityMins = 1; chatConfig.startDate = DateTime.now().toString(); stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); underTest.open(stormConf, context, collector); verifyZeroInteractions(collector, context); }
@Test public void testOpen_nullStartDate() { chatConfig.granularityMins = 1; chatConfig.startDate = null; stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); underTest.open(stormConf, context, collector); verifyZeroInteractions(collector, context); }
@Test(expected = IllegalArgumentException.class) public void testOpen_invalidGranularity() { chatConfig.granularityMins = -1; stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); underTest.open(stormConf, context, collector); }