dnet-core/dnet-information-service/src/main/java/eu/dnetlib/enabling/is/sn/resourcestate/ResourceStateSubscriptionRegistry.java

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;
	}

}