@Test(expected = IllegalArgumentException.class) public void testExecuteEmptyConf() throws Exception { bolt.prepare(new HashMap(), null, null); }
@Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { if (!stormConf.containsKey(CATALOG_ROOT_URL)) { throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL); } Map<String, Object> notificationConf = null; if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) { notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY); } else { notificationConf = Collections.emptyMap(); } NotificationStore notificationStore = null; try { if (!StringUtils.isEmpty(notificationStoreClazz)) { Class<?> clazz = Class.forName(notificationStoreClazz); notificationStore = (NotificationStore) clazz.newInstance(); Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY); if (config == null) { config = Collections.emptyMap(); } notificationStore.init(config); } } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) { throw new RuntimeException(ex); } notificationService = new NotificationServiceImpl(notificationConf, notificationStore); String jarPath = ""; if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) { jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(), File.separator, notificationSink.getNotifierJarFileName()); } Properties props = new Properties(); props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties())); NotifierConfig notifierConfig = new NotifierConfigImpl(props, convertMapValuesToString(notificationSink.getNotifierFieldValues()), notificationSink.getNotifierClassName(), jarPath); notificationContext = new BoltNotificationContext(collector, notifierConfig); notificationService.register(notificationSink.getNotifierName(), notificationContext); }
NotificationBolt extends BaseTickTupleAwareRichBolt { @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { if (!stormConf.containsKey(CATALOG_ROOT_URL)) { throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL); } Map<String, Object> notificationConf = null; if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) { notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY); } else { notificationConf = Collections.emptyMap(); } NotificationStore notificationStore = null; try { if (!StringUtils.isEmpty(notificationStoreClazz)) { Class<?> clazz = Class.forName(notificationStoreClazz); notificationStore = (NotificationStore) clazz.newInstance(); Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY); if (config == null) { config = Collections.emptyMap(); } notificationStore.init(config); } } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) { throw new RuntimeException(ex); } notificationService = new NotificationServiceImpl(notificationConf, notificationStore); String jarPath = ""; if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) { jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(), File.separator, notificationSink.getNotifierJarFileName()); } Properties props = new Properties(); props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties())); NotifierConfig notifierConfig = new NotifierConfigImpl(props, convertMapValuesToString(notificationSink.getNotifierFieldValues()), notificationSink.getNotifierClassName(), jarPath); notificationContext = new BoltNotificationContext(collector, notifierConfig); notificationService.register(notificationSink.getNotifierName(), notificationContext); } }
NotificationBolt extends BaseTickTupleAwareRichBolt { @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { if (!stormConf.containsKey(CATALOG_ROOT_URL)) { throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL); } Map<String, Object> notificationConf = null; if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) { notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY); } else { notificationConf = Collections.emptyMap(); } NotificationStore notificationStore = null; try { if (!StringUtils.isEmpty(notificationStoreClazz)) { Class<?> clazz = Class.forName(notificationStoreClazz); notificationStore = (NotificationStore) clazz.newInstance(); Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY); if (config == null) { config = Collections.emptyMap(); } notificationStore.init(config); } } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) { throw new RuntimeException(ex); } notificationService = new NotificationServiceImpl(notificationConf, notificationStore); String jarPath = ""; if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) { jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(), File.separator, notificationSink.getNotifierJarFileName()); } Properties props = new Properties(); props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties())); NotifierConfig notifierConfig = new NotifierConfigImpl(props, convertMapValuesToString(notificationSink.getNotifierFieldValues()), notificationSink.getNotifierClassName(), jarPath); notificationContext = new BoltNotificationContext(collector, notifierConfig); notificationService.register(notificationSink.getNotifierName(), notificationContext); } NotificationBolt(String notificationSinkJson);  NotificationBolt(NotificationSink notificationSink); }
NotificationBolt extends BaseTickTupleAwareRichBolt { @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { if (!stormConf.containsKey(CATALOG_ROOT_URL)) { throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL); } Map<String, Object> notificationConf = null; if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) { notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY); } else { notificationConf = Collections.emptyMap(); } NotificationStore notificationStore = null; try { if (!StringUtils.isEmpty(notificationStoreClazz)) { Class<?> clazz = Class.forName(notificationStoreClazz); notificationStore = (NotificationStore) clazz.newInstance(); Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY); if (config == null) { config = Collections.emptyMap(); } notificationStore.init(config); } } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) { throw new RuntimeException(ex); } notificationService = new NotificationServiceImpl(notificationConf, notificationStore); String jarPath = ""; if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) { jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(), File.separator, notificationSink.getNotifierJarFileName()); } Properties props = new Properties(); props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties())); NotifierConfig notifierConfig = new NotifierConfigImpl(props, convertMapValuesToString(notificationSink.getNotifierFieldValues()), notificationSink.getNotifierClassName(), jarPath); notificationContext = new BoltNotificationContext(collector, notifierConfig); notificationService.register(notificationSink.getNotifierName(), notificationContext); } NotificationBolt(String notificationSinkJson);  NotificationBolt(NotificationSink notificationSink); NotificationBolt withNotificationStoreClass(String clazz); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void cleanup(); }
NotificationBolt extends BaseTickTupleAwareRichBolt { @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { if (!stormConf.containsKey(CATALOG_ROOT_URL)) { throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL); } Map<String, Object> notificationConf = null; if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) { notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY); } else { notificationConf = Collections.emptyMap(); } NotificationStore notificationStore = null; try { if (!StringUtils.isEmpty(notificationStoreClazz)) { Class<?> clazz = Class.forName(notificationStoreClazz); notificationStore = (NotificationStore) clazz.newInstance(); Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY); if (config == null) { config = Collections.emptyMap(); } notificationStore.init(config); } } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) { throw new RuntimeException(ex); } notificationService = new NotificationServiceImpl(notificationConf, notificationStore); String jarPath = ""; if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) { jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(), File.separator, notificationSink.getNotifierJarFileName()); } Properties props = new Properties(); props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties())); NotifierConfig notifierConfig = new NotifierConfigImpl(props, convertMapValuesToString(notificationSink.getNotifierFieldValues()), notificationSink.getNotifierClassName(), jarPath); notificationContext = new BoltNotificationContext(collector, notifierConfig); notificationService.register(notificationSink.getNotifierName(), notificationContext); } NotificationBolt(String notificationSinkJson);  NotificationBolt(NotificationSink notificationSink); NotificationBolt withNotificationStoreClass(String clazz); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void cleanup(); static final String LOCAL_NOTIFIER_JAR_PATH; }