SlackMessageSpout 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); LOG.info("Loaded config..."); WebSocketContainer webSocketContainer = JdkContainerProvider.getWebSocketContainer(); IChatApiDAO slackDao = ChatAPIFactory.getChatApiDao(config); SlackConfig slackConfig = (SlackConfig) config.computeConfig.chatConfig; open(slackConfig, slackDao, webSocketContainer, context, collector); } SlackMessageSpout(); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context,
                     SpoutOutputCollector collector); @OnMessage void onMessageEvent(Message message, Session session); @OnError void onError(Throwable t); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer fields); @Override void close(); static final String SPOUT_ID; static final String SLACK_MESSAGE_FIELD_STR; }
@Test(expected = NotConnectedException.class) public void testOpen_withAuthException() throws Exception { stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); try { underTest.open(stormConf, mockContext, mockCollector); fail(); } catch (Exception e) { throw (Exception) e.getCause(); } }
@Test(expected = NotConnectedException.class) public void testOpen_withExceptionAndRetries() throws Exception { chatConfig.sourceConnectionSleepIntervalMs = 1000; chatConfig.sourceConnectionMaxMs = 1000; stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); try { underTest.open(stormConf, mockContext, mockCollector); fail(); } catch (Exception e) { throw (Exception) e.getCause(); } }