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; }
@Test(expected = IllegalArgumentException.class) public void testExecuteEmptyConf() throws Exception { bolt.prepare(new HashMap(), null, null); }
@Test public void testAck() throws Exception { Map<String, Object> fieldsAndValues = new HashMap<>(); fieldsAndValues.put("temperature", "100"); final StreamlineEvent event = StreamlineEventImpl.builder() .fieldsAndValues(fieldsAndValues) .dataSourceId("srcid") .build(); final Notification notification = new StreamlineEventAdapter(event); new MockUp<NotificationQueueHandler>() { @Mock public void enqueue(Notifier notifier, Notification notification1) { notifier.notify(notification); } }; new Expectations() {{ mockProxyUtil.loadClassFromJar(anyString, "TestClass"); result = notifier; tuple.getValueByField(anyString); result = event; }}; Map<String, String> stormConf = new HashMap<>(); stormConf.put("catalog.root.url", "http: stormConf.put("local.notifier.jar.path", "/tmp"); bolt.prepare(stormConf, null, collector); bolt.execute(tuple); new Verifications() { { hBaseNotificationStore.store(notification); times = 1; hBaseNotificationStore.updateNotificationStatus(notification.getId(), Notification.Status.DELIVERED); times = 1; collector.ack(tuple); times = 1; collector.fail(tuple); times = 0; } }; }
@Test public void testFail() throws Exception { Map<String, Object> fieldsAndValues = new HashMap<>(); fieldsAndValues.put("foobar", "100"); final StreamlineEvent event = StreamlineEventImpl.builder() .fieldsAndValues(fieldsAndValues) .dataSourceId("srcid") .build(); final Notification notification = new StreamlineEventAdapter(event); new MockUp<NotificationQueueHandler>() { @Mock public void enqueue(Notifier notifier, Notification notification1) { notifier.notify(notification); } @Mock public void resubmit(String notificationId) { notifier.notify(notification); } }; new Expectations() {{ mockProxyUtil.loadClassFromJar(anyString, "TestClass"); result = notifier; tuple.getValueByField(anyString); result = event; }}; Map<String, String> stormConf = new HashMap<>(); stormConf.put("catalog.root.url", "http: stormConf.put("local.notifier.jar.path", "/tmp"); bolt.prepare(stormConf, null, collector); bolt.execute(tuple); new Verifications() { { hBaseNotificationStore.store(notification); times = 1; hBaseNotificationStore.updateNotificationStatus(notification.getId(), Notification.Status.FAILED); times = 1; collector.ack(tuple); times = 0; collector.fail(tuple); times = 1; } }; }