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 id = topologyContext.getStormId();
		logger.debug("Application context lookup for topology '{}'", id);

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

		ApplicationContext entry = INSTANCE.registry.get(id);
		if (entry == null) {
			logger.debug("Need new application context");
			entry = instantiate(properties);
			logger.info("Application context instantiated for topology '{}'", id);
			INSTANCE.registry.put(id, 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;
	}

}

@@ -1049,43 +1049,159 @@
 ing 
-id = topologyContext.getStormId(
+topologyName = (String) stormConf.get(%22topology.name%22);%0A%09%09if (topologyName == null)%0A%09%09%09throw new IllegalStateException(%22Topology name is mandatory%22
 );%0A
+%0A
 %09%09lo
@@ -1259,18 +1259,28 @@
  '%7B%7D'%22, 
-id
+topologyName
 );%0A%0A%09%09Ma
@@ -1424,18 +1424,28 @@
 try.get(
-id
+topologyName
 );%0A%09%09if 
@@ -1616,18 +1616,28 @@
  '%7B%7D'%22, 
-id
+topologyName
 );%0A%09%09%09IN
@@ -1660,10 +1660,20 @@
 put(
-id
+topologyName
 , en

use topology name for register in singleton app context