package eu.dnetlib.functionality.index;

import java.io.StringReader;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.Node;
import org.dom4j.io.SAXReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Required;

import eu.dnetlib.enabling.actions.AbstractSubscriptionAction;
import eu.dnetlib.functionality.index.utils.MetadataReference;
import eu.dnetlib.functionality.index.utils.MetadataReferenceFactory;
import eu.dnetlib.functionality.index.utils.ServiceTools;
import eu.dnetlib.miscutils.dom4j.XPathHelper;

/**
 * MdFormatNotificationHandler updates index schema in response to metadata formats modifications.
 * 
 * @author claudio
 * 
 */
public class MdFormatNotificationHandler extends AbstractSubscriptionAction {

	/**
	 * Logger.
	 */
	private static final Log log = LogFactory.getLog(MdFormatNotificationHandler.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Resource
	private transient MetadataReferenceFactory mdFactory;

	@Autowired
	private IndexServerDAOMap indexServerDAOMap;

	@Autowired
	private ServiceTools serviceTools;

	private Executor executor = Executors.newSingleThreadExecutor();

	private final transient ThreadLocal<SAXReader> domFactory = new ThreadLocal<SAXReader>() {

		@Override
		protected SAXReader initialValue() {
			return new SAXReader();
		}
	};

	/**
	 * {@inheritDoc}
	 * 
	 * @see eu.dnetlib.enabling.tools.Enableable
	 */
	private boolean enabled;

	@Override
	public void notified(final String subscrId, final String topic, final String rsId, final String profile) {
		if (!topic.startsWith(getTopicPrefix()) || !isEnabled()) return;
		executor.execute(new Runnable() {

			@Override
			public void run() {
				try {
					final Document doc = parse(profile);
					final Node formatNode = doc.selectSingleNode("//CONFIGURATION/NAME/text()");

					if ((formatNode != null) && !formatNode.asXML().isEmpty()) {

						final String format = formatNode.asXML();
						final Iterable<Element> layouts = XPathHelper.selectElements(doc, "//STATUS/LAYOUTS/LAYOUT");
						final Node interpretationNode = doc.selectSingleNode("//CONFIGURATION/INTERPRETATION/text()");
						final String interpretation = interpretationNode.asXML();

						log.info("found a change in mdFormat: " + format);
						for (Element element : layouts) {
							final String layout = element.attributeValue("name");

							// updates index schema based on mdFormat and layout
							final MetadataReference mdRef = mdFactory.getMetadata(format, layout, interpretation);
							final Document fields = parse(element.selectSingleNode("./FIELDS").asXML());

							List<String> backends = serviceTools.getBackendIds(mdRef);

							if ((backends == null) || backends.isEmpty()) {
								log.warn("There is no backendId in profiles for mdref " + mdRef);
							}

							for (String backendId : backends) {
								IndexServerDAO idxDao = indexServerDAOMap.getIndexServerDAO(backendId);
								if (idxDao == null) throw new RuntimeException("No index found for the mdformat " + mdRef);
								log.info("Found index DAO which serve this mdFormat");
								idxDao.updateIndexCollection(mdRef, fields);
							}
						}
					}
					log.info("Upload schema completed");
				} catch (Exception e) {
					throw new RuntimeException(e); // NOPMD
				}
			}
		});
	}

	@Override
	@Required
	public void setEnabled(final boolean enabled) {
		this.enabled = enabled;
	}

	/**
	 * Helper method, parses an xml string.
	 * 
	 * @param source
	 *            the xml.
	 * @return the parsed xml.
	 * @throws DocumentException
	 *             could happen
	 */
	private Document parse(final String source) throws DocumentException {
		return domFactory.get().read(new StringReader(source));
	}

	@Override
	public boolean isEnabled() {
		return enabled;
	}

}