package eu.icolumbo.breeze;

import backtype.storm.task.TopologyContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.PropertySource;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import static java.lang.String.format;


/**
 * Spring registry.
 * @author Pascal S. de Kloe
 * @author Jethro Bakker
 */
public enum SingletonApplicationContext {

	INSTANCE;

	private static final Logger logger = LoggerFactory.getLogger(SingletonApplicationContext.class);

	private final Map<String,ApplicationContext> registry = new HashMap<>();


	/**
	 * Gets the Spring setup for the respective Storm topology.
	 */
	public static synchronized ApplicationContext get(Map stormConf, TopologyContext topologyContext) {
		String topologyName = (String) stormConf.get("topology.name");
		if (topologyName == null)
			throw new IllegalStateException("Topology name is mandatory");

		logger.debug("Application context lookup for topology '{}'", topologyName);

		Map<String,Object> properties = Collections.checkedMap(stormConf, String.class, Object.class);

		ApplicationContext entry = INSTANCE.registry.get(topologyName);
		if (entry == null) {
			logger.debug("Need new application context");
			entry = instantiate(properties);
			logger.info("Application context instantiated for topology '{}'", topologyName);
			INSTANCE.registry.put(topologyName, entry);
		}

		return entry;
	}

	private static ApplicationContext instantiate(Map<String, Object> stormConf) {
		String topologyName = (String) stormConf.get("topology.name");
		String[] configLocations = {format("classpath:/%s-context.xml", topologyName)};
		AbstractApplicationContext result = new ClassPathXmlApplicationContext(configLocations, false);
		result.setId(configLocations[0]);

		if (!stormConf.isEmpty()) {
			logger.debug("Applying Storm configuration: {}", stormConf);
			PropertySource propertySource = new MapPropertySource("storm-configuration", stormConf);
			result.getEnvironment().getPropertySources().addLast(propertySource);
		}

		result.refresh();
		return result;
	}

}

@@ -18,24 +18,54 @@
 bo.breeze;%0A%0A
+import backtype.storm.Config;%0A
 import backt
@@ -88,32 +88,32 @@
 opologyContext;%0A
-
 import org.slf4j
@@ -580,16 +580,76 @@
 format;%0A
+import static org.springframework.util.StringUtils.hasText;%0A
 %0A%0A/**%0A *
@@ -1177,31 +1177,46 @@
 get(
-%22topology.name%22);%0A%09%09if 
+Config.TOPOLOGY_NAME);%0A%09%09if (! hasText
 (top
@@ -1224,24 +1224,17 @@
 logyName
- == null
+)
 )%0A%09%09%09thr
@@ -1267,34 +1267,69 @@
 on(%22
-Topology name is mandatory
+Missing '%22 + Config.TOPOLOGY_NAME + %22' in Storm configuration
 %22);%0A
@@ -1967,23 +1967,28 @@
 get(
-%22topology.name%22
+Config.TOPOLOGY_NAME
 );%0A%09

Use and propagate topology name constant.