D-Net

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/cnr-blackboard-common/trunk/", "deploy_repository": "dnet4-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet4-snapshots", "name": "cnr-blackboard-common"}

package eu.dnetlib.enabling.tools.blackboard;

import javax.annotation.Resource;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(value = SpringJUnit4ClassRunner.class)

@ContextConfiguration

public class NotificationHandlerChainTest {

	public class RecJob implements Runnable {

		private int times;

		private int num;

		public RecJob(int num, int times) {

			this.num = num;

			this.times = times;

		}

		@Override

		public void run() {

			System.out.println("starting " + num);

			if(times >= 0)

				executor.execute(new  RecJob(num + 1, times - 1));

			try {

				Thread.sleep(4000);

			} catch (InterruptedException e) {

				// TODO Auto-generated catch block

				e.printStackTrace();

			}

			System.out.println("thread finished - " + num);

		}

	}

	private static final class Job implements Runnable {

		private final int value;

		public Job(final int i) {

			value = i;

		}

		@Override

		public void run() {

			System.out.println("thread started - " + value);

			try {

				Thread.sleep(4000);

			} catch (final InterruptedException e) {

				// 

			}

			System.out.println("thread finished - " + value);

		}

	}

	@Resource

	private transient ThreadPoolTaskExecutor executor;

	@Test

	public void testDelegateNotification() throws InterruptedException {

		System.out.println(executor);

		System.out.println("executing");

		for (int i = 0; i < 100; i++)

			executor.execute(new Job(i));

		System.out.println("executed - waiting");

		Thread.sleep(2000);

		System.out.println("active count: " + executor.getActiveCount());

		System.out.println("current pool size: " + executor.getCorePoolSize());

		System.out.println("pool size " + executor.getPoolSize());

		Thread.sleep(3000);

		System.out.println("ok");

	}

	@Test

	public void testRecursive() throws InterruptedException {

		for (int i = 0; i < 4; i++)

			executor.execute(new RecJob(i * 10, 4));

		System.out.println("executed - waiting");

		Thread.sleep(2000);

		System.out.println("active count: " + executor.getActiveCount());

		System.out.println("current pool size: " + executor.getCorePoolSize());

		System.out.println("pool size " + executor.getPoolSize());

		Thread.sleep(3000);

		System.out.println("ok");

	}

}

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

<beans xmlns="http://www.springframework.org/schema/beans"

	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

	xmlns:lang="http://www.springframework.org/schema/lang"

	xmlns:p="http://www.springframework.org/schema/p"

	xmlns:util="http://www.springframework.org/schema/util"

	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd

		http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.5.xsd

		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-2.5.xsd">

	<bean id="msroNotificationExecutor"

		class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"

		p:corePoolSize="32" p:keepAliveSeconds="3600" p:queueCapacity="0" />

</beans>

package eu.dnetlib.enabling.tools.blackboard;

import javax.annotation.PostConstruct;

/**

 * In most cases a service just dispatches every incoming blackboard execution request to the blackboard server

 * executor, so that the right action call back will be called.

 * 

 * <p>

 * This bean will wrap a BlackboardServerExecutor as a BlackboardNotificationHandler so that it can be installed as a

 * notification handler of a server.</p>

 * 

 * @author marko

 * 

 * @param <T>

 */

public class BlackboardServerExecutorNotificationHandler<T extends Enum<T>> extends AbstractBlackboardNotificationHandler<BlackboardServerHandler> {

	private BlackboardServerActionExecutor<T> blackboardExecutor;

	@PostConstruct

	public void init() {

		setBlackboardHandler(blackboardExecutor.getBlackboardHandler());

	}

	@Override

	protected void processJob(final BlackboardJob job) {

		blackboardExecutor.execute(job);

	}

	public BlackboardServerActionExecutor<T> getBlackboardExecutor() {

		return blackboardExecutor;

	}

	public void setBlackboardExecutor(BlackboardServerActionExecutor<T> blackboardExecutor) {

		this.blackboardExecutor = blackboardExecutor;

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.enabling.tools.blackboard.AbstractBlackboardNotificationHandler#setBlackboardHandler(eu.dnetlib.enabling.tools.blackboard.BlackboardHandler)

	 * 

	 *      Redefined here to avoid inheriting the @Required property. It simplifies the spring config for an

	 *      unnecessarry property, since we can get it from the blackboardExecutor.

	 */

	@Override

	public void setBlackboardHandler(BlackboardServerHandler handler) {

		super.setBlackboardHandler(handler);

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import java.util.HashMap;

import java.util.Map;

import java.util.Map.Entry;

/**

 * High level representation of a blackboard job.

 *

 * @author marko

 *

 */

public class BlackboardJobImpl implements BlackboardJob {

	/**

	 * underlying low level blackboard message.

	 */

	private BlackboardMessage message;

	/**

	 * service identifier.

	 */

	private String serviceId;

	/**

	 * parameters.

	 */

	private final transient Map<String, String> parameters = new HashMap<String, String>();

	/**

	 * Construct a new blackboard job from a blackboard message.

	 *

	 * @param serviceId

	 *            service identifier

	 * @param message

	 *            underlying low-level blackboard message

	 */

	public BlackboardJobImpl(final String serviceId, final BlackboardMessage message) {

		super();

		this.message = message;

		this.serviceId = serviceId;

		for (final BlackboardParameter param : message.getParameters())

			parameters.put(param.getName(), param.getValue());

	}

	@Override

	public String getId() {

		return message.getId();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJob#setId(java.lang.String)

	 */

	@Override

	public void setId(final String identifier) {

		message.setId(identifier);

	}

	@Override

	public String getAction() {

		return message.getAction();

	}

	@Override

	public ActionStatus getActionStatus() {

		return message.getActionStatus();

	}

	@Override

	public String getDate() {

		return message.getDate();

	}

	@Override

	public String getError() {

		return getParameters().get("error");

	}

	@Override

	public Map<String, String> getParameters() {

		return parameters;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJob#setAction(java.lang.String)

	 */

	@Override

	public void setAction(final String action) {

		message.setAction(action);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJob#setActionStatus(eu.dnetlib.enabling.tools.blackboard.ActionStatus)

	 */

	@Override

	public void setActionStatus(final ActionStatus actionStatus) {

		message.setActionStatus(actionStatus);

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJob#setDate(java.lang.String)

	 */

	@Override

	public void setDate(final String date) {

		message.setDate(date);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJob#setError(java.lang.String)

	 */

	@Override

	public void setError(final String error) {

		getParameters().put("error", error);

	}

	/**

	 * returns the blackboard message, potentially modified.

	 *

	 * @return underlying blackboard message

	 */

	public BlackboardMessage getMessage() {

		message.getParameters().clear();

		for (final Entry<String, String> entry : getParameters().entrySet()) {

			final BlackboardParameterImpl param = new BlackboardParameterImpl(); // NOPMD

			param.setName(entry.getKey());

			param.setValue(entry.getValue());

			message.getParameters().add(param);

		}

		return message;

	}

	public void setMessage(final BlackboardMessage message) {

		this.message = message;

	}

	public String getServiceId() {

		return serviceId;

	}

	public void setServiceId(final String serviceId) {

		this.serviceId = serviceId;

	}

	@Override

	public boolean isCompleted() {

		return getActionStatus() == ActionStatus.DONE || getActionStatus() == ActionStatus.FAILED;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import java.util.Map;

import javax.xml.bind.JAXBException;

import javax.xml.transform.dom.DOMSource;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.w3c.dom.Element;

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

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

import eu.dnetlib.enabling.tools.OpaqueResource;

import eu.dnetlib.enabling.tools.ServiceLocator;

import eu.dnetlib.enabling.tools.UniqueIdentifierGenerator;

import eu.dnetlib.enabling.tools.UniqueIdentifierGeneratorImpl;

import eu.dnetlib.miscutils.jaxb.JaxbFactory;

/**

 * Blackboard client.

 *

 * @author marko

 *

 */

public class BlackboardClientHandlerImpl implements BlackboardClientHandler {

	/**

	 * blackboard message factory.

	 */

	private JaxbFactory<BlackboardMessage> messageFactory;

	/**

	 * registry locator.

	 */

	private ServiceLocator<ISRegistryService> registryLocator;

	/**

	 * generates blackboard message identifiers for new jobs.

	 */

	private UniqueIdentifierGenerator uuidGenerator = new UniqueIdentifierGeneratorImpl("bb-");

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardClientHandler#assign(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	public void assign(final BlackboardJob job) {

		checkJob(job);

		try {

			final BlackboardJobImpl impl = (BlackboardJobImpl) job;

			registryLocator.getService().addBlackBoardMessage(impl.getServiceId(), job.getId(), messageFactory.serialize(impl.getMessage()));

		} catch (final ISRegistryException e) {

			throw new IllegalStateException("cannot register blackboard message", e);

		} catch (final JAXBException e) {

			throw new IllegalArgumentException("cannot serialize blackboard message", e);

		}

	}

	/**

	 * Check that the job has sane values.

	 * 

	 * @param job

	 */

	protected void checkJob(BlackboardJob job) {

		for(Map.Entry<String, String> param : job.getParameters().entrySet()) {

			if(param.getValue() == null)

				throw new IllegalStateException("job parameter value cannot be null: " + param.getKey());

		}

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardClientHandler#delete(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	public void delete(final BlackboardJob job) {

		try {

			final BlackboardJobImpl impl = (BlackboardJobImpl) job;

			registryLocator.getService().deleteBlackBoardMessage(impl.getServiceId(), job.getId());

		} catch (final ISRegistryException e) {

			throw new IllegalStateException("cannot delete blackboard message", e);

		}

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardClientHandler#newJob()

	 */

	@Override

	public BlackboardJob newJob(final String serviceId) {

		final BlackboardJobImpl job = new BlackboardJobImpl(serviceId, messageFactory.newInstance());

		job.setActionStatus(ActionStatus.ASSIGNED);

		job.getParameters().put("id", "");

		job.getParameters().put("error", "");

		job.setId(uuidGenerator.generateIdentifier());

		return job;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardHandler#getJob(eu.dnetlib.enabling.tools.OpaqueResource)

	 */

	@Override

	public BlackboardJob getJob(final OpaqueResource profile) {

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

		try {

			final Element source = (Element) xpa.evaluate("/RESOURCE_PROFILE/BODY/BLACKBOARD/MESSAGE[@id = /RESOURCE_PROFILE/BODY/BLACKBOARD/LAST_RESPONSE]",

					profile.asDom(), XPathConstants.NODE);

			if (source == null)

				throw new IllegalStateException("cannot find last blackboard message in the service profile");

			return new BlackboardJobImpl(profile.getResourceId(), messageFactory.parse(new DOMSource(source)));

		} catch (final JAXBException e) {

			throw new IllegalStateException("cannot parse blackboard message", e);

		} catch (final XPathExpressionException e) {

			throw new IllegalStateException("cannot find last blackboard message in the service profile", e);

		}

	}

	public JaxbFactory<BlackboardMessage> getMessageFactory() {

		return messageFactory;

	}

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

		this.messageFactory = messageFactory;

	}

	public ServiceLocator<ISRegistryService> getRegistryLocator() {

		return registryLocator;

	}

	public void setRegistryLocator(final ServiceLocator<ISRegistryService> registryLocator) {

		this.registryLocator = registryLocator;

	}

	public UniqueIdentifierGenerator getUuidGenerator() {

		return uuidGenerator;

	}

	public void setUuidGenerator(final UniqueIdentifierGenerator uuidGenerator) {

		this.uuidGenerator = uuidGenerator;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import java.io.PrintWriter;

import java.io.StringWriter;

import javax.xml.bind.JAXBException;

import javax.xml.transform.dom.DOMSource;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.w3c.dom.Element;

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

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

import eu.dnetlib.enabling.tools.OpaqueResource;

import eu.dnetlib.enabling.tools.ServiceLocator;

import eu.dnetlib.miscutils.jaxb.JaxbFactory;

/**

 * Blackboard handler implementation.

 *

 * @author marko

 *

 */

public class BlackboardHandlerImpl implements BlackboardServerHandler {

	/**

	 * blackboard message factory.

	 */

	private JaxbFactory<BlackboardMessage> messageFactory;

	/**

	 * registry locator.

	 */

	private ServiceLocator<ISRegistryService> registryLocator;

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardHandler#getJob(eu.dnetlib.enabling.tools.OpaqueResource)

	 */

	@Override

	public BlackboardJob getJob(final OpaqueResource profile) {

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

		try {

			final Element source = (Element) xpa.evaluate("/RESOURCE_PROFILE/BODY/BLACKBOARD/MESSAGE[@id = /RESOURCE_PROFILE/BODY/BLACKBOARD/LAST_REQUEST]",

					profile.asDom(), XPathConstants.NODE);

			if (source == null)

				throw new IllegalStateException("cannot find last blackboard message in the service profile");

			return new BlackboardJobImpl(profile.getResourceId(), messageFactory.parse(new DOMSource(source)));

		} catch (final JAXBException e) {

			throw new IllegalStateException("cannot parse blackboard message", e);

		} catch (final XPathExpressionException e) {

			throw new IllegalStateException("cannot find last blackboard message in the service profile", e);

		}

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardHandler#done(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	public void done(final BlackboardJob job) {

		job.setActionStatus(ActionStatus.DONE);

		replyJob(job);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardHandler#failed(eu.dnetlib.enabling.tools.blackboard.BlackboardJob,

	 *      java.lang.Exception)

	 */

	@Override

	public void failed(final BlackboardJob job, final Throwable exception) {

		job.setActionStatus(ActionStatus.FAILED);

		final StringWriter stackTrace = new StringWriter();

		exception.printStackTrace(new PrintWriter(stackTrace));

		job.getParameters().put("error", exception.toString());

		job.getParameters().put("errorDetails", stackTrace.toString());

		replyJob(job);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardHandler#ongoing(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	public void ongoing(final BlackboardJob job) {

		job.setActionStatus(ActionStatus.ONGOING);

		replyJob(job);

	}

	/**

	 * Internal helper method which replies a blackboard job.

	 *

	 * @param job

	 *            blackboard job

	 */

	protected void replyJob(final BlackboardJob job) {

		try {

			final BlackboardJobImpl impl = (BlackboardJobImpl) job;

			registryLocator.getService().replyBlackBoardMessage(impl.getServiceId(), messageFactory.serialize(impl.getMessage()));

		} catch (final ISRegistryException e) {

			throw new IllegalStateException("cannot reply the blackboard message", e);

		} catch (final JAXBException e) {

			throw new IllegalArgumentException("cannot serialize blackboard message", e);

		}

	}

	public JaxbFactory<BlackboardMessage> getMessageFactory() {

		return messageFactory;

	}

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

		this.messageFactory = messageFactory;

	}

	public ServiceLocator<ISRegistryService> getRegistryLocator() {

		return registryLocator;

	}

	public void setRegistryLocator(final ServiceLocator<ISRegistryService> registryLocator) {

		this.registryLocator = registryLocator;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import javax.annotation.Resource;

import javax.xml.bind.JAXBException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.miscutils.jaxb.JaxbFactory;

/**

 * This BB handler deletes completed (successful or unsuccessful) blackboard messages after dispatching.

 * Job should be registered to only one "deleting BB handler".

 *

 * @author marko

 *

 */

public class DeletingBlackboardNotificationHandler extends BlackboardNotificationHandler<BlackboardClientHandler> {

	/**

	 * Logger.

	 */

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

	/**

	 * blackboard message factory.

	 */

	@Resource(name = "blackboardMessageFactory")

	private JaxbFactory<BlackboardMessage> messageFactory;

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler#processJob(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	protected void processJob(final BlackboardJob job) {

		if (getListeners().containsKey(job.getId()) && job.isCompleted()) {

			if (log.isDebugEnabled())

				log.debug(serializeBlackBoardMessage(job));

			getBlackboardHandler().delete(job);

		}

		super.processJob(job);

	}

	/**

	 * Helper method, serializes a Blackboard message using a blackboardMessageFactory. 

	 * 

	 * @param job

	 * @return

	 */

	private String serializeBlackBoardMessage(BlackboardJob job) {

		try {

			final BlackboardJobImpl impl = (BlackboardJobImpl) job;

			return getMessageFactory().serialize(impl.getMessage());

		} catch (JAXBException e) {

			return "cannot serialize blackboard message: " + e.getMessage();

		}

	}

	public void setMessageFactory(JaxbFactory<BlackboardMessage> messageFactory) {

		this.messageFactory = messageFactory;

	}

	public JaxbFactory<BlackboardMessage> getMessageFactory() {

		return messageFactory;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import javax.xml.bind.annotation.XmlAttribute;

import javax.xml.bind.annotation.XmlRootElement;

import org.apache.commons.lang.builder.EqualsBuilder;

import org.apache.commons.lang.builder.HashCodeBuilder;

/**

 * Blackboard parameter.

 * 

 * @author marko

 * 

 */

@XmlRootElement(name = "PARAMETER")

public class BlackboardParameterImpl implements BlackboardParameter {

	/**

	 * hash seed.

	 */

	private static final int HASH_SEED_2 = 59;

	/**

	 * hash seed.

	 */

	private static final int HASH_SEED = 35;

	/**

	 * parameter name.

	 */

	private String name;

	/**

	 * parameter value.

	 */

	private String value;

	/**

	 * {@inheritDoc}

	 * 

	 * @see java.lang.Object#equals(java.lang.Object)

	 */

	@Override

	public boolean equals(final Object obj) {

		if (!(obj instanceof BlackboardParameter))

			return false;

		if (this == obj)

			return true;

		final BlackboardParameter rhs = (BlackboardParameter) obj;

		return new EqualsBuilder().append(name, rhs.getName()).append(value, rhs.getValue()).isEquals();

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see java.lang.Object#hashCode()

	 */

	@Override

	public int hashCode() {

		return new HashCodeBuilder(HASH_SEED, HASH_SEED_2).append(name).append(value).toHashCode();

	}

	@Override

	@XmlAttribute

	public String getName() {

		return name;

	}

	@Override

	public void setName(final String name) {

		this.name = name;

	}

	@Override

	@XmlAttribute

	public String getValue() {

		return value;

	}

	@Override

	public void setValue(final String value) {

		this.value = value;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Registers BlackboardJobListeners to be called when related incoming blackboard job notifications arrive.

 * @author marko

 *

 */

public interface BlackboardJobRegistry {

	/**

	 * Registers a new job listener. It will be notified when the job changes state. The listener will be unregistered

	 * when the job arrives to a final state (DONE, or FAILED).

	 *

	 * @param job

	 *            job

	 * @param listener

	 *            job listener

	 */

	void registerJobListener(BlackboardJob job, BlackboardJobListener listener);

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Utility abstract class which dispatches to commodity onSomething() methods, one for each interesting

 * blackboard job state.

 *

 * @author marko

 *

 */

public abstract class AbstractBlackboardJobListener implements BlackboardJobListener {

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.enabling.tools.blackboard.BlackboardJobListener#processJob(eu.dnetlib.enabling.tools.blackboard.BlackboardJob)

	 */

	@Override

	public void processJob(final BlackboardJob job) {

		if (job.getActionStatus() == ActionStatus.DONE)

			onDone(job);

		else if (job.getActionStatus() == ActionStatus.FAILED)

			onFailed(job);

		else if (job.getActionStatus() == ActionStatus.ASSIGNED)

			onAssigned(job);

		else if (job.getActionStatus() == ActionStatus.ONGOING)

			onOngoing(job);

	}

	/**

	 * Called when the job enters the ASSIGNED state.

	 *

	 * @param job job

	 */

	protected void onAssigned(final BlackboardJob job) { // NOPMD

		// default no operation

		// TODO: increase job expiry time

	}

	/**

	 * Called when the job enters the ONGOING state.

	 *

	 * @param job job

	 */

	protected void onOngoing(final BlackboardJob job) { // NOPMD

		// default no operation

		// TODO: increase job expiry time

	}

	/**

	 * Called when the job finishes in the FAILED state.

	 *

	 * @param job job

	 */

	protected abstract void onFailed(BlackboardJob job);

	/**

	 * Called when the job finishes in the DONE state.

	 *

	 * @param job job

	 */

	protected abstract void onDone(BlackboardJob job);

}

package eu.dnetlib.enabling.tools.blackboard;

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

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

/**

 * Callback interface for serverside blackboard actions.

 * 

 * <p>

 * Actions should use the 'handler' methods to set the 'ongoing' or 'done' status.

 * </p>

 * 

 * <p>

 * failed status is automatically set upon catching an exception thrown by this interface, so feel free to simply throw

 * whatever you want

 * </p>

 * 

 * @author marko

 * 

 * @param <X>

 */

public interface BlackboardServerAction<X extends Enum<?>> {

	void execute(BlackboardServerHandler handler, BlackboardJob job) throws Exception;

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Helper component used to bridge the high level blackboard job handling from the low level blackboard protocol based

 * on notifications and blackboard messages.

 *

 * @author marko

 *

 */

public interface BlackboardServerHandler extends BlackboardHandler {

	/**

	 * Sets the ongoing action status to the given job, publishing this new state.

	 *

	 * @param job

	 *            blackboard job

	 */

	void ongoing(BlackboardJob job);

	/**

	 * Sets the "failed" action status to the given job, publishing this new state along with the error message obtained

	 * from the exception.

	 *

	 * @param job

	 *            blackboard job

	 * @param exception

	 *            exception which caused the failure

	 */

	void failed(BlackboardJob job, Throwable exception);

	/**

	 * Set the "done" action status to the given job, publishing the new state.

	 *

	 * @param job

	 *            blackboard job

	 */

	void done(BlackboardJob job);

}

package eu.dnetlib.enabling.tools.blackboard;

import java.util.ArrayList;

import java.util.List;

import javax.xml.bind.annotation.XmlAccessType;

import javax.xml.bind.annotation.XmlAccessorType;

import javax.xml.bind.annotation.XmlAttribute;

import javax.xml.bind.annotation.XmlElement;

import javax.xml.bind.annotation.XmlRootElement;

import org.apache.commons.lang.builder.EqualsBuilder;

import org.apache.commons.lang.builder.HashCodeBuilder;

/**

 * standard serialization of the blackboard message.

 *

 * @author marko

 *

 */

@XmlRootElement(namespace = "", name = "MESSAGE")

@XmlAccessorType(XmlAccessType.NONE)

public class BlackboardMessageImpl implements BlackboardMessage {

	/**

	 * hash seed.

	 */

	private static final int HASH_SEED_2 = 63;

	/**

	 * hash seed.

	 */

	private static final int HASH_SEED = 13;

	/**

	 * blackboard message timestamp.

	 */

	@XmlAttribute

	private String date;

	/**

	 * blackboard message identifier.

	 */

	@XmlAttribute

	private String id; // NOPMD

	/**

	 * blackboard message action name.

	 */

	@XmlElement(name = "ACTION", required = true)

	private String action;

	/**

	 * blackboard message parameters (key/value).

	 */

	@XmlElement(name = "PARAMETER", type = BlackboardParameterImpl.class)

	private List<BlackboardParameter> parameters = new ArrayList<BlackboardParameter>();

	/**

	 * the status of the action described by this message.

	 */

	@XmlElement(name = "ACTION_STATUS", required = true)

	private ActionStatus actionStatus;

	/**

	 * {@inheritDoc}

	 *

	 * @see java.lang.Object#equals(java.lang.Object)

	 */

	@Override

	public boolean equals(final Object obj) {

		if (!(obj instanceof BlackboardMessage))

			return false;

		if (this == obj)

			return true;

		final BlackboardMessage rhs = (BlackboardMessage) obj;

		return new EqualsBuilder().append(id, rhs.getId()).append(date, rhs.getDate()).append(action, rhs.getAction()).append(actionStatus,

				rhs.getActionStatus()).append(parameters, rhs.getParameters()).isEquals();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see java.lang.Object#hashCode()

	 */

	@Override

	public int hashCode() {

		return new HashCodeBuilder(HASH_SEED, HASH_SEED_2).append(id).append(date).append(action).append(actionStatus).append(parameters).toHashCode();

	}

	@Override

	public String getDate() {

		return date;

	}

	@Override

	public void setDate(final String date) {

		this.date = date;

	}

	@Override

	public String getId() {

		return id;

	}

	@Override

	public void setId(final String id) { // NOPMD

		this.id = id;

	}

	@Override

	public String getAction() {

		return action;

	}

	@Override

	public void setAction(final String action) {

		this.action = action;

	}

	@Override

	public List<BlackboardParameter> getParameters() {

		return parameters;

	}

	public void setParameters(final List<BlackboardParameter> parameters) {

		this.parameters = parameters;

	}

	@Override

	public ActionStatus getActionStatus() {

		return actionStatus;

	}

	@Override

	public void setActionStatus(final ActionStatus actionStatus) {

		this.actionStatus = actionStatus;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import java.io.IOException;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

import javax.xml.parsers.ParserConfigurationException;

import javax.xml.xpath.XPathExpressionException;

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

import org.xml.sax.SAXException;

import eu.dnetlib.enabling.tools.Enableable;

import eu.dnetlib.enabling.tools.StringOpaqueResource;

/**

 * Common blackboard notification handler. This notification handler processes only message with ...BODY.BLACKBOARD.LAST* as topic.

 * 

 * @param <T>

 *            type of blackboard handler used to extract the blackboard message (client or server)

 * @author marko

 * 

 */

public abstract class AbstractBlackboardNotificationHandler<T extends BlackboardHandler> implements NotificationHandler, Enableable {

	/**

	 * blackboard handler.

	 */

	private T blackboardHandler;

	/**

	 * true if enabled.

	 */

	private boolean enabled = true;

	/**

	 * Executor handles the notified request in a dedicated thread and allows to return immediately.

	 */

	private Executor executor = Executors.newCachedThreadPool();

	/**

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.data.mdstore.NotificationHandler#notified(java.lang.String, java.lang.String, java.lang.String, java.lang.String)

	 */

	@Override

	public void notified(final String subscrId, final String topic, final String rsId, final String profile) {

		if (!topic.contains("BODY.BLACKBOARD.LAST")) return;

		executor.execute(new Runnable() {

			@Override

			public void run() {

				try {

					processJob(blackboardHandler.getJob(new StringOpaqueResource(profile)));

				} catch (final XPathExpressionException e) {

					throw new IllegalStateException(e);

				} catch (final SAXException e) {

					throw new IllegalStateException(e);

				} catch (final IOException e) {

					throw new IllegalStateException(e);

				} catch (final ParserConfigurationException e) {

					throw new IllegalStateException(e);

				}

			}

		});

	}

	/**

	 * Subclassess override this to process incoming blackboard jobs.

	 * 

	 * @param job

	 *            blackboard job

	 */

	protected abstract void processJob(BlackboardJob job);

	public T getBlackboardHandler() {

		return blackboardHandler;

	}

	@Required

	public void setBlackboardHandler(final T blackboardHandler) {

		this.blackboardHandler = blackboardHandler;

	}

	@Override

	public boolean isEnabled() {

		return enabled;

	}

	@Override

	public void setEnabled(final boolean enabled) {

		this.enabled = enabled;

	}

}

package eu.dnetlib.enabling.tools.blackboard;

import java.util.Map;

/**

 * High level representation of a blackboard job.

 *

 * @author marko

 *

 */

public interface BlackboardJob {

	/**

	 * Get the message id.

	 *

	 * @return id

	 */

	String getId();

	/**

	 * Set message id.

	 *

	 * @param identifier

	 *            id

	 */

	void setId(String identifier);

	/**

	 * Get the action name.

	 *

	 * @return action name

	 */

	String getAction();

	/**

	 * Set the action name.

	 *

	 * @param action

	 *            action name

	 */

	void setAction(String action);

	/**

	 * Get the action status.

	 *

	 * @return action status

	 */

	ActionStatus getActionStatus();

	/**

	 * Set the action status.

	 *

	 * @param actionStatus

	 *            action status

	 */

	void setActionStatus(ActionStatus actionStatus);

	/**

	 * Get the message date.

	 *

	 * @return date

	 */

	String getDate();

	/**

	 * Set the message date.

	 *

	 * @param date date

	 */

	void setDate(String date);

	/**

	 * obtains a mutable parameter map (key/value).

	 *

	 * @return mutable parameter map

	 */

	Map<String, String> getParameters();

	/**

	 * Get the error message, if the actionStatus is FAILED. Usually the error message is also present in the parameter

	 * map but it's preferred to use this method.

	 *

	 * @return error message

	 */

	String getError();

	/**

	 * Set the error message, if the actionStatus is FAILED. Usually the error message is also present in the parameter

	 * map but it's preferred to use this method.

	 *

	 * @param error

	 *            error message

	 */

	void setError(String error);

	/**

	 * True if the action status is one of the termination statuses (e.g. done and failed).

	 *

	 * @return true if completed

	 */

	boolean isCompleted();

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Helpers for the blackboard protocol client.

 *

 * @author marko

 *

 */

public interface BlackboardClientHandler extends BlackboardHandler {

	/**

	 * Create a new job.

	 *

	 * @param serviceId service identifier

	 * @return newly created blackboard job

	 */

	BlackboardJob newJob(String serviceId);

	/**

	 * Assign a blackboard job to a service.

	 *

	 * @param job blackboard job to send

	 */

	void assign(BlackboardJob job);

	/**

	 * The client can delete the job after it has reached a final state

	 * or the job timeout has expired.

	 *

	 * @param job blackboard job to delete.

	 */

	void delete(BlackboardJob job);

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Blackboard action status.

 * 

 * @author marko

 *

 */

public enum ActionStatus {

	/**

	 * The job/action is assigned but not jet taken into started execution.

	 */

	ASSIGNED,

	/**

	 * The job/action is ongoing.

	 */

	ONGOING,

	/**

	 * The job/action is completed successfully.

	 */

	DONE,

	/**

	 * The job/action is completed with failure.

	 */

	FAILED

}

package eu.dnetlib.enabling.tools.blackboard;

import eu.dnetlib.enabling.tools.OpaqueResource;

/**

 * Basic blackboard handler.

 *

 * @author marko

 *

 */

public interface BlackboardHandler {

	/**

	 * Get the current job from, as notified in the service profile.

	 *

	 * @param profile

	 *            service profile

	 * @return notified blackboard job

	 */

	BlackboardJob getJob(OpaqueResource profile);

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Blackboard parameter used in a BlackboardMessage.

 * 

 * @author marko

 *

 */

public interface BlackboardParameter {

	/**

	 * parameter name (key).

	 * @return name

	 */

	String getName();

	/**

	 * setter.

	 * 

	 * @param name name

	 */

	void setName(String name);

	/**

	 * parameter value.

	 * @return value

	 */

	String getValue();

	/**

	 * setter.

	 * 

	 * @param value value

	 */

	void setValue(String value);

}

package eu.dnetlib.enabling.tools.blackboard;

/**

 * Implement this interface in order to receive notifications for specific blackboard messages.

 *

 * @author marko

 *

 */

public interface BlackboardJobListener {

	/**

	 * process the given job.

	 *