D-Net

package eu.dnetlib.enabling.is.sn;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * Asynchronous but in-order delivery of notifications.

 * 

 * @author marko

 * 

 */

public class AsynchronousNotificationSenderImpl extends AbstractNotificationSender implements Runnable { // NOPMD

	/**

	 * logger.

	 */

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

	/**

	 * Encapsulates an notification job.

	 * 

	 * @author marko

	 *

	 */

	class NotificationJob {

		/**

		 * notification destination.

		 */

		private final transient W3CEndpointReference destination;

		/**

		 * notification message. 

		 */

		private final transient NotificationMessage message;

		/**

		 * construct a new notification job.

		 * 

		 * @param destination destination

		 * @param message message

		 */

		public NotificationJob(final W3CEndpointReference destination, final NotificationMessage message) {

			super();

			this.destination = destination;

			this.message = message;

		}

		public W3CEndpointReference getDestination() {

			return destination;

		}

		public NotificationMessage getMessage() {

			return message;

		}

	}

	/**

	 * job queue.

	 */

	private BlockingQueue<NotificationJob> jobQueue = new LinkedBlockingQueue<NotificationJob>();

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.sn.NotificationSender#send(javax.xml.ws.wsaddressing.W3CEndpointReference,

	 *      eu.dnetlib.enabling.is.sn.NotificationMessage)

	 */

	@Override

	public void send(final W3CEndpointReference destination, final NotificationMessage message) {

		log.debug("queuing asynchronous notification");

		try {

			jobQueue.put(new NotificationJob(destination, message));

		} catch (InterruptedException e) {

			log.warn("possibly lost notification", e);

		}

	}

	/**

	 * start this notification sender (called by spring lifecycle).

	 */

	void start() {

		new Thread(this).start(); // NOPMD

	}

	/** 

	 * {@inheritDoc}

	 * @see java.lang.Runnable#run()

	 */

	@Override

	public void run() {

		while (true) {

			try {

				final NotificationJob job = jobQueue.take();

				try {

					getInvoker().send(job.getDestination(), job.getMessage(), 0);

				} catch (javax.xml.ws.soap.SOAPFaultException t) {

					log.fatal("error sending notification to " + job.getDestination().toString(), t);

				}

			} catch (InterruptedException e) {

				log.warn("possibly lost notification", e);

			}

		}

	}

	public BlockingQueue<NotificationJob> getJobQueue() {

		return jobQueue;

	}

	public void setJobQueue(final BlockingQueue<NotificationJob> jobQueue) {

		this.jobQueue = jobQueue;

	}

}

package eu.dnetlib.enabling.is.sn;

import javax.xml.xpath.XPathExpressionException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.w3c.dom.Document;

import eu.dnetlib.enabling.is.sn.resourcestate.ResourceStateNotificationDetector;

import eu.dnetlib.enabling.tools.DOMOpaqueResource;

import eu.dnetlib.enabling.tools.OpaqueResource;

import eu.dnetlib.xml.database.LoggingTrigger;

/**

 * This trigger generates notification events for the subscription notification layer. 

 * 

 * @author marko

 *

 */

public class NotificationTriggerImpl extends LoggingTrigger {

	/**

	 * logger.

	 */

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

	/**

	 * notification detector to notify on events.

	 */

	private ResourceStateNotificationDetector<OpaqueResource> detector;

	/** 

	 * {@inheritDoc}

	 * @see eu.dnetlib.xml.database.LoggingTrigger#created(java.lang.String, java.lang.String, org.w3c.dom.Document)

	 */

	@Override

	public void created(final String file, final String collection, final Document newDoc) {

		super.created(file, collection, newDoc);

		try {

			getDetector().resourceCreated(new DOMOpaqueResource(newDoc));

		} catch (XPathExpressionException e) {

			log.fatal("cannot detect notification because of xpath error; notification possibly missed", e);

		}

	}

	/** 

	 * {@inheritDoc}

	 * @see eu.dnetlib.xml.database.LoggingTrigger#deleted(java.lang.String, java.lang.String, org.w3c.dom.Document)

	 */

	@Override

	public void deleted(final String file, final String collection, final Document oldDoc) {

		super.deleted(file, collection, oldDoc);

		try {

			getDetector().resourceDeleted(new DOMOpaqueResource(oldDoc));

		} catch (XPathExpressionException e) {

			log.fatal("cannot detect notification because of xpath error; notification possibly missed", e);

		}

	}

	/** 

	 * {@inheritDoc}

	 * @see eu.dnetlib.xml.database.LoggingTrigger#updated(java.lang.String, java.lang.String, org.w3c.dom.Document, org.w3c.dom.Document)

	 */

	@Override

	public void updated(final String file, final String collection, final Document oldDoc, final Document newDoc) {

		super.updated(file, collection, oldDoc, newDoc);

		try {

			getDetector().resourceUpdated(new DOMOpaqueResource(oldDoc), new DOMOpaqueResource(newDoc));

		} catch (XPathExpressionException e) {

			log.fatal("cannot detect notification because of xpath error; notification possibly missed", e);

		}

	}

	public ResourceStateNotificationDetector<OpaqueResource> getDetector() {

		return detector;

	}

	public void setDetector(final ResourceStateNotificationDetector<OpaqueResource> detector) {

		this.detector = detector;

	}

}

package eu.dnetlib.enabling.is.registry.schema;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.enabling.tools.OpaqueResource;

/**

 * Unlike the OpaqueResourceValidatorImpl, this implementation only logs an error but accepts every resource as valid.

 * 

 * Use only during testing/development!

 * 

 * @author marko

 * 

 */

public class PermissiveOpaqueResourceValidatorImpl extends OpaqueResourceValidatorImpl {

	/**

	 * logger.

	 */

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

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.schema.OpaqueResourceValidatorImpl#validate(eu.dnetlib.enabling.tools.OpaqueResource)

	 */

	@Override

	public void validate(final OpaqueResource resource) throws ValidationException {

		try {

			super.validate(resource);

		} catch (ValidationException e) {

			log.fatal("validation error, continuing (permissive mode)", e);

		}

	}

}

package eu.dnetlib.enabling.is.sn;

import java.util.Collection;

import java.util.Date;

import java.util.concurrent.ConcurrentLinkedQueue;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

/**

 * In memory implementation of a notification invocation logger.

 *

 * @author marko

 *

 */

public class MemoryNotificationInvocationLogger implements NotificationInvocationLogger {

	/**

	 * default log size.

	 */

	private static final int DEFAULT_SIZE = 100;

	/**

	 * Entry.

	 *

	 * @author marko

	 *

	 */

	public class MemoryEntry implements Entry {

		/**

		 * logged message.

		 */

		private NotificationMessage message;

		/**

		 * message status.

		 */

		private Status status = Status.Queued;

		/**

		 * saved exception in case of Failed state.

		 */

		private Throwable exception;

		/**

		 * destination.

		 */

		private W3CEndpointReference destinationEpr;

		/**

		 * start date time.

		 */

		private Date start;

		/**

		 * finish date time.

		 */

		private Date finish;

		/**

		 * construct a new message.

		 *

		 * @param endpointReference

		 *            destination

		 *

		 * @param message

		 *            message

		 */

		public MemoryEntry(final W3CEndpointReference endpointReference, final NotificationMessage message) {

			this.destinationEpr = endpointReference;

			this.message = message;

			this.start = new Date();

		}

		/**

		 * {@inheritDoc}

		 *

		 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger.Entry#ongoing()

		 */

		@Override

		public void ongoing() {

			status = Status.Ongoing;

		}

		/**

		 * {@inheritDoc}

		 *

		 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger.Entry#failure(java.lang.Throwable)

		 */

		@Override

		public void failure(final Throwable exc) {

			status = Status.Failed;

			this.exception = exc;

			commit();

		}

		/**

		 * {@inheritDoc}

		 *

		 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger.Entry#success()

		 */

		@Override

		public void success() {

			status = Status.Succeeded;

			commit();

		}

		/**

		 * finalize log entry.

		 */

		protected void commit() {

			this.finish = new Date();

		}

		@Override

		public NotificationMessage getMessage() {

			return message;

		}

		public void setMessage(final NotificationMessage message) {

			this.message = message;

		}

		@Override

		public Throwable getException() {

			return exception;

		}

		public void setException(final Throwable exception) {

			this.exception = exception;

		}

		@Override

		public Status getStatus() {

			return status;

		}

		public void setStatus(final Status status) {

			this.status = status;

		}

		public W3CEndpointReference getDestinationEpr() {

			return destinationEpr;

		}

		public void setDestinationEpr(final W3CEndpointReference destinationEpr) {

			this.destinationEpr = destinationEpr;

		}

		@Override

		public String getDestination() {

			return EPRUtil.getAddress(destinationEpr);

		}

		/**

		 * {@inheritDoc}

		 *

		 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger.Entry#getErrorMessage()

		 */

		@Override

		public String getErrorMessage() {

			if (getException() == null)

				return "";

			return getException().getMessage();

		}

		@Override

		public Date getStart() {

			return start;

		}

		public void setStart(final Date start) {

			this.start = start;

		}

		@Override

		public Date getFinish() {

			return finish;

		}

		public void setFinish(final Date finish) {

			this.finish = finish;

		}

		/**

		 * {@inheritDoc}

		 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger.Entry#getDuration()

		 */

		@Override

		public String getDuration() {

			if (finish == null || start == null)

				return "-";

			return (finish.getTime() - start.getTime()) + " ms";

		}

	}

	/**

	 * the actual log storage.

	 */

	private ConcurrentLinkedQueue<Entry> queue = new ConcurrentLinkedQueue<Entry>();

	/**

	 * max log size.

	 */

	private int size = DEFAULT_SIZE;

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.is.sn.NotificationInvocationLogger#startLogging(eu.dnetlib.enabling.is.sn.NotificationMessage)

	 */

	@Override

	public Entry startLogging(final W3CEndpointReference dest, final NotificationMessage message) {

		final Entry entry = createLoggingEntry(dest, message);

		queue.add(entry);

		ensureQueueLength();

		return entry;

	}

	/**

	 * ensure that the queue doesn't grow beyond size.

	 */

	private void ensureQueueLength() {

		if (queue.size() > size)

			queue.poll();

	}

	/**

	 * create a new memory log object.

	 *

	 * @param dest

	 *            destination

	 *

	 * @param message

	 *            message to be logged

	 * @return new entry instance

	 */

	protected Entry createLoggingEntry(final W3CEndpointReference dest, final NotificationMessage message) {

		return new MemoryEntry(dest, message);

	}

	public ConcurrentLinkedQueue<Entry> getQueue() {

		return queue;

	}

	public void setQueue(final ConcurrentLinkedQueue<Entry> queue) {

		this.queue = queue;

	}

	public int getSize() {

		return size;

	}

	public void setSize(final int size) {

		this.size = size;

	}

	@Override

	public Collection<Entry> getEntries() {

		return getQueue();

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet45-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>cnr-enabling-services</artifactId>

	<packaging>jar</packaging>

	<version>2.2.2</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/cnr-enabling-services/tags/cnr-enabling-services-2.2.2</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>commons-beanutils</groupId>

			<artifactId>commons-beanutils</artifactId>

			<version>1.8.0</version>

		</dependency>

		<dependency>

			<groupId>apache</groupId>

			<artifactId>oro</artifactId>

			<version>2.0.8</version>

		</dependency>

		<dependency>

			<groupId>com.google.guava</groupId>

			<artifactId>guava</artifactId>

			<version>${google.guava.version}</version>

		</dependency>

		<dependency>

			<groupId>org.quartz-scheduler</groupId>

			<artifactId>quartz</artifactId>

			<version>${quartz.version}</version>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-beans</artifactId>

			<version>${spring.version}</version>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-jdbc</artifactId>

			<version>${spring.version}</version>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-tx</artifactId>

			<version>${spring.version}</version>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-orm</artifactId>

			<version>${spring.version}</version>

		</dependency>

		<dependency>

			<groupId>org.postgresql</groupId>

			<artifactId>jdbc4driver</artifactId>

			<version>8.3</version>

		</dependency>

		<dependency>

			<groupId>org.xerial</groupId>

			<artifactId>sqlite-jdbc</artifactId>

			<version>3.7.2</version>

		</dependency>

		<dependency>

			<groupId>org.hibernate</groupId>

			<artifactId>hibernate-core</artifactId>

			<version>4.3.2.Final</version>

			<exclusions>

				<exclusion>

					<artifactId>antlr</artifactId>

					<groupId>antlr</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>commons-collections</groupId>

			<artifactId>commons-collections</artifactId>

			<version>${commons.collections.version}</version>

		</dependency>

		<dependency>

			<groupId>commons-dbcp</groupId>

			<artifactId>commons-dbcp</artifactId>

			<version>1.4</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-rmi-api</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-misc-utils</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-service-utils</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-blackboard-common</artifactId>

			<version>[2.1.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-xmldb</artifactId>

			<version>[2.0.0,3.0.0)</version>

			<exclusions>

				<exclusion>

					<artifactId>commons-collections</artifactId>

					<groupId>apache</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>${google.gson.version}</version>

		</dependency>

		<dependency>

			<groupId>com.sun.xml.messaging.saaj</groupId>

			<artifactId>saaj-impl</artifactId>

			<version>1.3</version>

		</dependency>

		<dependency>

			<groupId>jaxen</groupId>

			<artifactId>jaxen</artifactId>

			<version>1.1.6</version>

		</dependency>

		<!-- TEST DEPS -->

		<dependency>

			<groupId>hsqldb</groupId>

			<artifactId>hsqldb</artifactId>

			<version>1.8.0.7</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.aspectj</groupId>

			<artifactId>aspectjrt</artifactId>

			<version>1.6.3</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.aspectj</groupId>

			<artifactId>aspectjweaver</artifactId>

			<version>1.6.3</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.hibernate</groupId>

			<artifactId>hibernate-annotations</artifactId>

			<version>3.3.1</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>xerces</groupId>

			<artifactId>xercesImpl</artifactId>

			<version>2.11.0</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-test-utils</artifactId>

			<version>[1.0.0,2.0.0)</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-webmvc</artifactId>

			<version>${spring.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.springframework</groupId>

			<artifactId>spring-test</artifactId>

			<version>${spring.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-frontend-jaxws</artifactId>

			<version>${cxf.version}</version>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-ws-policy</artifactId>

			<version>${cxf.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-frontend-jaxrs</artifactId>

			<scope>test</scope>

			<version>${cxf.version}</version>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-transports-local</artifactId>

			<version>${cxf.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-ws-rm</artifactId>

			<version>${cxf.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>org.mockito</groupId>

			<artifactId>mockito-core</artifactId>

			<version>${mockito.version}</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.enabling.is.sn;

/**

 * NotificationSender interface defines an injection point for the NotificationInvoker instance.

 * 

 * <p>

 * It will be fairly common for implementors of NotificationSender to back the invoker reference into an instance

 * variable. This abstract class offers this for free. </p>

 * 

 * @author marko

 * 

 */

public abstract class AbstractNotificationSender implements NotificationSender {

	/**

	 * notification invoker used to send messages to destinations.

	 */

	private NotificationInvoker invoker;

	public NotificationInvoker getInvoker() {

		return invoker;

	}

	@Override

	public void setInvoker(final NotificationInvoker invoker) {

		this.invoker = invoker;

	}

}

package eu.dnetlib.enabling.is.registry;

import javax.xml.transform.dom.DOMResult;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.tools.OpaqueResource;

import eu.dnetlib.enabling.tools.StringOpaqueResource;

import eu.dnetlib.enabling.tools.blackboard.BlackboardMessage;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.miscutils.jaxb.JaxbFactory;

/**

 * Simple registry blackboard manager implementation.

 * 

 * @author marko

 * 

 */

public class RegistryBlackboardManagerImpl implements RegistryBlackboardManager { // NOPMD

	/**

	 * logger.

	 */

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

	/**

	 * timestamp padding.

	 */

	private static final String PADDING = "000";

	/**

	 * milliseconds in a second.

	 */

	private static final int MILLIS = 1000;

	/**

	 * IS Lookup

	 */

	private ISLookUpService isLookup;

	/**

	 * the registry which is bound with this registry blackboard manager implementation.

	 */

	private ISRegistryService registryService;

	/**

	 * blackboard message factory.

	 */

	private JaxbFactory<BlackboardMessage> messageFactory;

	/**

	 * provides the current date. Testers can override.

	 */

	private MessageDater messageDater = new MessageDater();

	/**

	 * Testers can override.

	 * 

	 * @author marko

	 * 

	 */

	public static class MessageDater {

		public String getCurrentDate() {

			return DateUtils.now_ISO8601();

		}

		public String getNumericStamp() {

			return Long.toString(System.currentTimeMillis() / MILLIS) + "." + System.currentTimeMillis() % MILLIS + PADDING;

		}

	}

	/**

	 * LAST_REQUEST or LAST_RESPONSE blackboard time stamp holders.

	 * 

	 * @author marko

	 * 

	 */

	public enum LastStamp {

		/**

		 * LAST_REQUEST, used when the blackboard message flows from the client to the server.

		 */

		REQUEST,

		/**

		 * LAST_RESPONSE, used when the blackboard message flows from the server to the client, signaling the completion/progress of the

		 * action..

		 */

		RESPONSE

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.RegistryBlackboardManager#addMessage(java.lang.String, java.lang.String, java.lang.String)

	 */

	@Override

	public void addMessage(final String profId, final String messageId, final String message) { // NOPMD

		try {

			final BlackboardMessage bbm = messageFactory.parse(message);

			bbm.setDate(messageDater.getCurrentDate()); // preserve compatibility.

			if (bbm.getId() == null || !bbm.getId().equals(messageId)) { throw new IllegalArgumentException("invalid blackboard message id"); }

			synchronized (this) {

				final OpaqueResource serviceProfile = new StringOpaqueResource(isLookup.getResourceProfile(profId));

				final Document doc = serviceProfile.asDom();

				final Node bboard = (Node) XPathFactory.newInstance().newXPath().evaluate("//BLACKBOARD", doc, XPathConstants.NODE);

				// delete current element if exists

				final Node messageElement = (Node) XPathFactory.newInstance().newXPath().evaluate("//BLACKBOARD/MESSAGE[@id='" + messageId + "']", doc,

						XPathConstants.NODE);

				if (messageElement != null) {

					bboard.removeChild(messageElement);

				}

				// append the serialized node to the blackboard node

				messageFactory.serialize(bbm, new DOMResult(bboard));

				updateLastStamps(doc, LastStamp.REQUEST, messageId);

				registryService.updateProfile(profId, serviceProfile.asString(), serviceProfile.getResourceType());

			}

		} catch (Exception e) {

			throw new IllegalStateException(e);

		}

	}

	/**

	 * Helper method which updates the LAST_* stamps in the blackboard header.

	 * 

	 * @param doc

	 *            profile DOM document

	 * @param stamp

	 *            which stamp to modify

	 * @param messageId

	 *            message id to point the stamp to

	 * @throws XPathExpressionException

	 *             could happen

	 */

	protected void updateLastStamps(final Document doc, final LastStamp stamp, final String messageId) throws XPathExpressionException {

		final Element lastRequest = (Element) XPathFactory.newInstance().newXPath().evaluate("//BLACKBOARD/LAST_" + stamp, doc, XPathConstants.NODE);

		lastRequest.setTextContent(messageId);

		lastRequest.setAttribute("date", messageDater.getNumericStamp());

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.RegistryBlackboardManager#deleteMessage(java.lang.String, java.lang.String)

	 */

	@Override

	public void deleteMessage(final String profId, final String messageId) {

		try {

			synchronized (this) {

				final OpaqueResource serviceProfile = new StringOpaqueResource(isLookup.getResourceProfile(profId));

				final Document doc = serviceProfile.asDom();

				final Node message = (Node) XPathFactory.newInstance().newXPath().evaluate("//BLACKBOARD/MESSAGE[@id='" + messageId + "']", doc,

						XPathConstants.NODE);

				message.getParentNode().removeChild(message);

				registryService.updateProfile(profId, serviceProfile.asString(), serviceProfile.getResourceType());

			}

			log.debug("Deleted bb message " + messageId + " from profile " + profId);

		} catch (Exception e) {

			log.error("Error deleting bb message " + messageId + " from profile " + profId, e);

			throw new IllegalStateException(e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.RegistryBlackboardManager#replyMessage(java.lang.String, java.lang.String)

	 */

	@Override

	public void replyMessage(final String profId, final String message) {

		try {

			final BlackboardMessage bbm = messageFactory.parse(message);

			bbm.setDate(messageDater.getCurrentDate()); // preserve compatibility.

			final String messageId = bbm.getId();

			synchronized (this) {

				final OpaqueResource serviceProfile = new StringOpaqueResource(isLookup.getResourceProfile(profId));

				final Document doc = serviceProfile.asDom();

				final Node messageElement = (Node) XPathFactory.newInstance().newXPath().evaluate("//BLACKBOARD/MESSAGE[@id='" + messageId + "']", doc,

						XPathConstants.NODE);

				if (messageElement == null) { throw new IllegalArgumentException("no such blackboard message " + messageId + ". Unably to reply"); }

				final Node bboard = messageElement.getParentNode();

				bboard.removeChild(messageElement);

				// append the serialized node to the blackboard node

				messageFactory.serialize(bbm, new DOMResult(bboard));

				updateLastStamps(doc, LastStamp.RESPONSE, messageId);

				registryService.updateProfile(profId, serviceProfile.asString(), serviceProfile.getResourceType());

			}

		} catch (Exception e) {

			throw new IllegalStateException(e);

		}

	}

	public ISRegistryService getRegistryService() {

		return registryService;

	}

	@Required

	public void setRegistryService(final ISRegistryService registryService) {

		this.registryService = registryService;

	}

	public JaxbFactory<BlackboardMessage> getMessageFactory() {

		return messageFactory;

	}

	@Required

	public void setMessageFactory(final JaxbFactory<BlackboardMessage> messageFactory) {

		this.messageFactory = messageFactory;

	}

	public MessageDater getMessageDater() {

		return messageDater;

	}

	public void setMessageDater(final MessageDater messageDater) {

		this.messageDater = messageDater;

	}

	public ISLookUpService getIsLookup() {

		return isLookup;

	}

	@Required

	public void setIsLookup(final ISLookUpService isLookup) {

		this.isLookup = isLookup;

	}

}

package eu.dnetlib.enabling.is.sn;

/**

 * Implementors of this interface are able to detect events which will trigger notifications.

 * @author marko

 *

 */

public interface NotificationDetector {

	/**

	 * injection point for NotificationSender.

	 * 

	 * @param sender sender

	 */

	void setSender(NotificationSender sender);

}

package eu.dnetlib.enabling.is.sn;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * This implementation of NotificationSender simply sends notification synchronously.

 * 

 * NOTE: this is only for testing. Real-wold cases should use AsynchrnonousNotificationSender, because we have to 

 * get out the thread serving eXist queries.

 * 

 * @author marko

 * 

 */

public class SynchronousNotificationSenderImpl extends AbstractNotificationSender {

	/**

	 * logger.

	 */

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

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.sn.NotificationSender#send(javax.xml.ws.wsaddressing.W3CEndpointReference,

	 *      eu.dnetlib.enabling.is.sn.NotificationMessage)

	 */

	@Override

	public void send(final W3CEndpointReference destination, final NotificationMessage message) {

		log.info("synchronously sending message " + message.getSubscriptionId() + ", " + message.getTopic() + ", " + message.getResourceId() + ", "

				+ message.getBody());

		try {

			getInvoker().send(destination, message, 0);

		} catch (javax.xml.ws.soap.SOAPFaultException t) {

			log.fatal("error sending notification to " + destination.toString(), t);

		}

	}

}

package eu.dnetlib.enabling.is.registry; // NOPMD

import java.io.IOException;

import java.util.Date;

import java.util.List;

import javax.annotation.Resource;

import javax.jws.WebService;

import javax.xml.parsers.ParserConfigurationException;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.w3c.dom.Element;

import org.xml.sax.SAXException;

import com.sun.xml.messaging.saaj.util.Base64;

import eu.dnetlib.common.rmi.APIDeprecatedException;

import eu.dnetlib.common.rmi.UnimplementedException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpDocumentNotFoundException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.is.registry.schema.OpaqueResourceValidator;

import eu.dnetlib.enabling.is.registry.schema.ValidationException;

import eu.dnetlib.enabling.is.registry.validation.ProfileValidationStrategy;

import eu.dnetlib.enabling.is.registry.validation.RegistrationPhase;

import eu.dnetlib.enabling.is.store.rmi.ISStoreException;

import eu.dnetlib.enabling.is.store.rmi.ISStoreService;

import eu.dnetlib.enabling.tools.AbstractBaseService;

import eu.dnetlib.enabling.tools.CompatResourceIdentifierResolverImpl;

import eu.dnetlib.enabling.tools.OpaqueResource;

import eu.dnetlib.enabling.tools.ResourceIdentifierResolver;

import eu.dnetlib.enabling.tools.ResourceType;

import eu.dnetlib.enabling.tools.StringOpaqueResource;

import eu.dnetlib.enabling.tools.UniqueIdentifierGenerator;

import eu.dnetlib.enabling.tools.UniqueIdentifierGeneratorImpl;

import eu.dnetlib.enabling.tools.XQueryUtils;

/**

 * Registry service implementation.

 * 

 * @author marko

 * 

 */

@WebService(targetNamespace = "http://services.dnetlib.eu/")

public class ISRegistryServiceImpl extends AbstractBaseService implements ISRegistryService { // NOPMD by marko on

	/**

	 * logger.

	 */

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

	/**

	 * error message: secure profile registration.

	 */

	private static final String ERROR_SEC_PROFILE = "cannot register secure profile";

	/**

	 * error message: trying to retrieve the current stored version of the resource profile.

	 */

	private static final String CANT_FETCH = "cannot fetch original profile";

	/**

	 * error message: cannot create a resource.

	 */

	private static final String CANT_CREATE = "cannot create resource";

	/**

	 * IS Store

	 */

	private ISStoreService isStore;

	/**

	 * IS Lookup

	 */

	private ISLookUpService isLookup;

	/**

	 * uuid generator.

	 */

	@Resource

	private UniqueIdentifierGenerator uuidGenerator = new UniqueIdentifierGeneratorImpl();

	/**

	 * manages the pending state of resource. Different implementations can be plugged in.

	 */

	private PendingResourceManager pendingManager = new CompatPendingResourceManagerImpl();

	/**

	 * manages resource identifier mappings with the abstract xmldb namespace (files/collections).

	 */

	@Resource

	private ResourceIdentifierResolver resIdResolver = new CompatResourceIdentifierResolverImpl();

	/**

	 * used to validate resources.

	 */

	private OpaqueResourceValidator resourceValidator;

	/**

	 * used to validate resources w.r.t. a set of defined properties.

	 */

	@Resource

	private ProfileValidationStrategy profileValidationStrategy;

	/**

	 * query utils. Used to obtain xmldb collection names and other things useful for interacting with the IS_Store.

	 */

	private XQueryUtils xqueryUtils;

	/**

	 * the blackboard management stuff is factored out here.

	 */

	@Resource

	private RegistryBlackboardManager blackboardManager;

	@Override

	public void start() {

		super.start();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#addBlackBoardMessage(java.lang.String, java.lang.String, java.lang.String)

	 */

	@Override

	public void addBlackBoardMessage(final String profId, final String messageId, final String message) throws ISRegistryException {

		blackboardManager.addMessage(profId, messageId, message);

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#addOrUpdateResourceType(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean addOrUpdateResourceType(final String resourceType, final String resourceSchema) throws ISRegistryException {

		return addResourceType(resourceType, resourceSchema);

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#addProfileNode(java.lang.String, java.lang.String, java.lang.String)

	 */

	@Override

	public boolean addProfileNode(final String profId, final String xpath, final String node) throws ISRegistryException {

		// TODO Auto-generated method stub

		throw new UnimplementedException();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#addResourceType(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean addResourceType(final String resourceType, final String resourceSchema) throws ISRegistryException {

		try {

			return isStore.insertXML(resourceType, xqueryUtils.getRootCollection() + ResourceType.RESOURCE_TYPES,

					resourceSchema);

		} catch (ISStoreException e) {

			throw new ISRegistryException("cannot add resource type", e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#deleteBlackBoardMessage(java.lang.String, java.lang.String)

	 */

	@Override

	public void deleteBlackBoardMessage(final String profId, final String messageId) throws ISRegistryException {

		blackboardManager.deleteMessage(profId, messageId);

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#deleteProfile(java.lang.String)

	 */

	@Override

	public boolean deleteProfile(final String resId) throws ISRegistryException {

		try {

			final boolean res = isStore.deleteXML(resIdResolver.getFileName(resId),

					"/db/DRIVER/" + resIdResolver.getCollectionName(resId));

			if (!res) { throw new ISRegistryDocumentNotFoundException("document " + resId + " not found"); }

			return true;

		} catch (ISStoreException e) {

			throw new ISRegistryException("cannot delete profile " + resId, e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#deleteProfiles(java.util.List)

	 */

	@Override

	public boolean deleteProfiles(final List<String> arrayprofId) throws ISRegistryException {

		throw new APIDeprecatedException();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#deleteResourceType(java.lang.String, java.lang.Boolean)

	 */

	@Override

	public boolean deleteResourceType(final String resourceType, final Boolean hierarchical) throws ISRegistryException {

		try {

			return isStore.deleteXML(resourceType, xqueryUtils.getRootCollection() + ResourceType.RESOURCE_TYPES);

		} catch (ISStoreException e) {

			throw new ISRegistryException("error deleting resource type " + resourceType, e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#executeXUpdate(java.lang.String)

	 */

	@Override

	public boolean executeXUpdate(final String xquery) throws ISRegistryException {

		try {

			return isStore.executeXUpdate(xquery);

		} catch (ISStoreException e) {

			throw new ISRegistryException(e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#insertProfileForValidation(java.lang.String, java.lang.String)

	 */

	@Override

	public String insertProfileForValidation(final String resourceType, final String resourceProfile) throws ISRegistryException {

		try {

			final OpaqueResource resource = new StringOpaqueResource(resourceProfile);

			if (!resourceType.equals(resource.getResourceType())) { throw new ISRegistryException("expected resource type doesn't match to resource"); }

			pendingManager.setPending(resource, true);

			return registerProfile(resource.asString());

		} catch (XPathExpressionException e) {

			throw new ISRegistryException(e);

		} catch (SAXException e) {

			throw new ISRegistryException(e);

		} catch (IOException e) {

			throw new ISRegistryException(e);

		} catch (ParserConfigurationException e) {

			throw new ISRegistryException(e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#refreshProfile(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean refreshProfile(final String profId, final String resourceType) throws ISRegistryException {

		// TODO Auto-generated method stub

		throw new UnimplementedException();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#registerProfile(java.lang.String)

	 */

	@Override

	public String registerProfile(final String resourceProfile) throws ISRegistryException {

		log.debug("registering profile");

		try {

			final OpaqueResource resource = new StringOpaqueResource(resourceProfile);

			// TODO: factor this out (possibly not within the OpaqueResource class)

			final String coll = xqueryUtils.getCollectionPath(resource);

			final String fileName = uuidGenerator.generateIdentifier();

			final String newId = fileName + "_" + new String(Base64.encode(coll.getBytes()));

			resource.setResourceId(newId);

			resource.setModificationDate(new Date());

			// log.info("validating to xml schema: " + resource.asString());

			resourceValidator.validate(resource);

			profileValidationStrategy.accept(resource, RegistrationPhase.Register);

			// TODO: factor out ResourceType class

			isStore.insertXML(fileName, xqueryUtils.getRootCollection() + coll, resource.asString());

			return resource.getResourceId();

		} catch (XPathExpressionException e) {

			throw new ISRegistryException(e);

		} catch (SAXException e) {

			throw new ISRegistryException(e);

		} catch (IOException e) {

			throw new ISRegistryException(e);

		} catch (ParserConfigurationException e) {

			throw new ISRegistryException(e);

		} catch (ISStoreException e) {

			throw new ISRegistryException(e);

		} catch (ValidationException e) {

			throw new ISRegistryException("profile is not conforming to the schema: " + e.getMessage(), e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#registerSecureProfile(java.lang.String, java.lang.String)

	 */

	@Override

	public String registerSecureProfile(final String resourceProfId, final String secureProfId) throws ISRegistryException {

		try {

			synchronized (this) {

				final String secureProfSrc;

				try {

					secureProfSrc = isLookup.getResourceProfile(secureProfId); // NOPMD

				} catch (ISLookUpDocumentNotFoundException e) {

					throw new ISRegistryException("cannot register secure profile, the given secure profile doesn't exist", e);

				}

				final OpaqueResource secureProf = new StringOpaqueResource(secureProfSrc);

				final String profId = validateProfile(resourceProfId);

				final XPath xpath = XPathFactory.newInstance().newXPath();

				final Element idEl = (Element) xpath.evaluate("/RESOURCE_PROFILE/BODY/CONFIGURATION/resourceId", secureProf.asDom(), XPathConstants.NODE);

				idEl.setTextContent(profId);

				if (!updateProfile(secureProfId, secureProf.asString(), secureProf.getResourceType())) { throw new ISRegistryException(

						"cannot update security profile (updateProfile returned false)"); }

				return profId;

			}

		} catch (XPathExpressionException e) {

			throw new ISRegistryException(ERROR_SEC_PROFILE, e);

		} catch (SAXException e) {

			throw new ISRegistryException(ERROR_SEC_PROFILE, e);

		} catch (IOException e) {

			throw new ISRegistryException(ERROR_SEC_PROFILE, e);

		} catch (ParserConfigurationException e) {

			throw new ISRegistryException(ERROR_SEC_PROFILE, e);

		} catch (ISLookUpException e) {

			throw new ISRegistryException("cannot register secure profile, problem fetching the given secure profile", e);

		}

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#removeProfileNode(java.lang.String, java.lang.String)

	 */

	@Override

	public boolean removeProfileNode(final String profId, final String nodeId) throws ISRegistryException {

		// TODO Auto-generated method stub

		throw new UnimplementedException();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.is.registry.rmi.ISRegistryService#replyBlackBoardMessage(java.lang.String, java.lang.String)

	 */

	@Override

	public void replyBlackBoardMessage(final String profId, final String message) throws ISRegistryException {