D-Net

package eu.dnetlib.enabling.is.sn.resourcestate;

import java.util.Arrays;

import java.util.Collection;

import java.util.HashSet;

import java.util.Set;

import javax.xml.transform.dom.DOMResult;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import com.google.common.collect.Sets;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.oro.text.perl.Perl5Util;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

import eu.dnetlib.enabling.is.sn.AbstractSubscriptionRegistry;

import eu.dnetlib.enabling.is.sn.SubscriptionRegistry;

import eu.dnetlib.enabling.is.sn.SubscriptionRequest;

import eu.dnetlib.enabling.is.sn.TopicExpressionMatchResult;

import eu.dnetlib.enabling.is.sn.rmi.SubscriptionRequestRejectedException;

import org.springframework.cache.annotation.CacheEvict;

import org.springframework.cache.annotation.Cacheable;

/**

 * Manage subscription for UPDATE/CREATE/DELETE resource-related topic prefixes.

 *

 * @author marko

 *

 */

public class ResourceStateSubscriptionRegistry extends AbstractSubscriptionRegistry implements SubscriptionRegistry {

	private static final Log log = LogFactory.getLog(ResourceStateSubscriptionRegistry.class);

	/**

	 * subscription DAO.

	 */

	private ResourceStateSubscriptionDAO subscriptionDao;

	/**

	 * subscription request filter.

	 */

	@Autowired

	private SubscriptionRequestFilter subscriptionRequestFilter;

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.is.sn.SubscriptionRegistry#registerSubscription(eu.dnetlib.enabling.is.sn.SubscriptionRequest)

	 */

	@Override

	@CacheEvict(value = "subscriptions", allEntries = true)

	public String registerSubscription(final SubscriptionRequest subscription) throws SubscriptionRequestRejectedException {

		final TopicExpressionMatchResult prefixMatch = matchPrefix(subscription);

		if (prefixMatch == null)

			return null;

		final TopicExpressionMatchResult typeMatch = matchType(prefixMatch.getRest());

		if (typeMatch == null)

			return null; // TODO: decide whether to fail or not

		final TopicExpressionMatchResult idMatch = matchId(typeMatch.getRest());

		if (idMatch == null)

			return null; // TODO: decide whether to fail or not

		final ResourceStateSubscription rss = new ResourceStateSubscription(subscription, prefixMatch.getPrefix(), typeMatch.getPrefix(), idMatch.getPrefix(),

				idMatch.getRest());

		if (!getSubscriptionRequestFilter().accept(rss))

			throw new SubscriptionRequestRejectedException(String.format("rejected subscription request, resourceId: '%s', xpath: '%s', from: %s",

					rss.getResourceId(), rss.getXpath(), rss.getSubscriber()));

		return registerSubscription(rss);

	}

	/**

	 * this registers the real subscription.

	 *

	 * TODO: am I sure that the overload is a good thing here?

	 *

	 * @param subscription

	 *            subscription

	 * @return subscription id (potentially changed)

	 */

	private String registerSubscription(final ResourceStateSubscription subscription) {

	    if(log.isDebugEnabled()) {

		log.debug("evict subscriptions cache");

	    }

		// TODO: change the dao, and implement a method which finds a given subscription directly.

		final Collection<ResourceStateSubscription> similar = subscriptionDao.listSubscriptions(subscription.getPrefix(), subscription.getType(),

				subscription.getResourceId());

		for (final ResourceStateSubscription r : similar) {

			if (r != null && getAddress(subscription.getSubscriberAsEpr()).equals(getAddress(r.getSubscriberAsEpr()))

					&& (subscription.getXpath() == r.getXpath() || subscription.getXpath().equals(r.getXpath()))

					&& (subscription.getResourceId() == r.getResourceId() || subscription.getResourceId().equals(r.getResourceId())))

				return r.getSubscriptionId();

		}

		subscriptionDao.addSubscription(subscription);

		return subscription.getSubscriptionId();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.is.sn.SubscriptionRegistry#unsubscribe(java.lang.String)

	 */

	@Override

	@CacheEvict(value = "subscriptions", allEntries = true)

	public boolean unsubscribe(final String subId) {

		log.info("evict subscriptions cache");

		return subscriptionDao.removeSubscription(subId);

	}

	/**

	 * Obtains the address component of the EPR.

	 *

	 * TODO: refactor in a utility class.

	 *

	 * @param epr

	 *            endpoint reference

	 * @return address contained in the endpoint reference

	 */

	private Object getAddress(final W3CEndpointReference epr) {

		final DOMResult dom = new DOMResult();

		epr.writeTo(dom);

		try {

			return XPathFactory.newInstance().newXPath().evaluate("//*[local-name() = 'Address']", dom.getNode());

		} catch (final XPathExpressionException e) {

			throw new IllegalStateException("cannot construct xpath expression", e);

		}

	}

	/**

	 * extract resource type name.

	 *

	 * @param rest

	 *            topic expression without prefix

	 * @return tuple containing type name and rest of the prefix

	 */

	public TopicExpressionMatchResult matchType(final String rest) {

		final Perl5Util matcher = new Perl5Util(); // NOPMD

		if (!matcher.match("/^(\\*|[a-zA-Z_0-9]*)($|/)(.*)$/", rest))

			return null;

		return new TopicExpressionMatchResult(matcher.getMatch().group(1), matcher.getMatch().group(2 + 1));

	}

	/**

	 * extract resource id.

	 *

	 * @param rest

	 *            topic expression without prefix

	 * @return tuple containing type name and rest of the prefix

	 */

	public TopicExpressionMatchResult matchId(final String rest) {

		final Perl5Util matcher = new Perl5Util(); // NOPMD

		if (!matcher.match("/^([^/]*)(.*)/", rest))

			return null;

		return new TopicExpressionMatchResult(matcher.getMatch().group(1), matcher.getMatch().group(2));

	}

	/**

	 * return all subscriptions matching a given prefix and a given type. Wildcard subscriptions will match any resource type.

	 *

	 * @param prefix

	 *            prefix

	 * @param type

	 *            concrete type

	 * @param resId

	 *            resource identifier

	 * @return all matching subscriptions

	 */

	@Override

	@Cacheable(value="subscriptions", key="{ #prefix, #type, #resId }")

	public Collection<ResourceStateSubscription> listMatchingSubscriptions(final String prefix, final String type, final String resId) {

		if(log.isDebugEnabled()) {

			log.debug(String.format("uncached list subscriptions [prefix: '%s', type: '%s', resourceId: '%s']", prefix, type, resId));

		}

		final Set<ResourceStateSubscription> merged = new HashSet<ResourceStateSubscription>();

		merged.addAll(subscriptionDao.listSubscriptions(prefix, type, resId));

		merged.addAll(subscriptionDao.listSubscriptions(prefix, type, "*"));

		merged.addAll(subscriptionDao.listSubscriptions(prefix, "*", "*"));

		return merged;

	}

	@Override

	@Cacheable(value="subscriptions")

	public Collection<ResourceStateSubscription> listSubscriptions() {

		log.info("uncached list subscriptions");

		return Sets.newHashSet(subscriptionDao.listSubscriptions());

	}

	@Override

	@CacheEvict(value = "subscriptions", allEntries = true)

	public boolean removeSubscription(final String subscriptionId) {

		return subscriptionDao.removeSubscription(subscriptionId);

	}

	@Override

	protected Collection<String> getAcceptedPrefixes() {

		return Arrays.asList(new String[] { ResourceStateSubscription.PREFIX_CREATE, ResourceStateSubscription.PREFIX_DELETE,

				ResourceStateSubscription.PREFIX_UPDATE });

	}

	public ResourceStateSubscriptionDAO getSubscriptionDao() {

		return subscriptionDao;

	}

	@Required

	public void setSubscriptionDao(final ResourceStateSubscriptionDAO subscriptionDao) {

		this.subscriptionDao = subscriptionDao;

	}

	public SubscriptionRequestFilter getSubscriptionRequestFilter() {

		return subscriptionRequestFilter;

	}

	public void setSubscriptionRequestFilter(final SubscriptionRequestFilter subscriptionRequestFilter) {

		this.subscriptionRequestFilter = subscriptionRequestFilter;

	}

}