D-Net

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;

	}

}