D-Net

package eu.dnetlib.enabling.blackboard;

import java.util.Comparator;

import java.util.Date;

import java.util.Map;

import java.util.UUID;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import eu.dnetlib.common.ifaces.BlackboardCallback;

import eu.dnetlib.common.services.locators.DnetServiceLocator;

import eu.dnetlib.common.services.locators.ServiceRunningInstance;

import eu.dnetlib.enabling.annotations.Blackboard;

import eu.dnetlib.enabling.nodeManager.NodeManagerServiceImpl;

import eu.dnetlib.rmi.objects.is.BlackboardActionStatus;

import eu.dnetlib.rmi.objects.is.BlackboardMessageContainer;

import eu.dnetlib.rmi.objects.is.Operation;

import eu.dnetlib.rmi.objects.is.Subscription;

import eu.dnetlib.rmi.soap.InformationService;

import eu.dnetlib.rmi.soap.exceptions.InformationServiceException;

public class BlackboardDispatcher {

	@Resource

	private DnetServiceLocator serviceLocator;

	@Resource

	private NodeManagerServiceImpl nodeManagerService;

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

	private Map<String, BlackboardCallback<?>> responseActions = Maps.newHashMap();

	private Map<String, Class<?>> responseClasses = Maps.newHashMap();

	public <T> String createDispatcher(final T message, final BlackboardCallback<T> callback) {

		final Blackboard annotation = getMessageAnnotation(message);

		final String serviceId = serviceLocator.getServiceId(annotation.serviceClass());

		createDispatcher(annotation, serviceId, message, callback);

		return serviceId;

	}

	public <T> String createDispatcher(final T message, final String dsId, final BlackboardCallback<T> callback) {

		final Blackboard annotation = getMessageAnnotation(message);

		final String serviceId = serviceLocator.getServiceId(annotation.serviceClass(), dsId);

		createDispatcher(annotation, serviceId, message, callback);

		return serviceId;

	}

	public <T> String createDispatcher(final T message, final Comparator<ServiceRunningInstance> comparator, final BlackboardCallback<T> callback) {

		final Blackboard annotation = getMessageAnnotation(message);

		final String serviceId = serviceLocator.getServiceId(annotation.serviceClass(), comparator);

		createDispatcher(annotation, serviceId, message, callback);

		return serviceId;

	}

	public <T> void createDispatcher(final String serviceId, final T message, final BlackboardCallback<T> callback) {

		final Blackboard annotation = getMessageAnnotation(message);

		createDispatcher(annotation, serviceId, message, callback);

	}

	private <T> void createDispatcher(final Blackboard annotation, final String serviceId, final T message, final BlackboardCallback<T> callback) {

		try {

			final InformationService is = serviceLocator.getService(InformationService.class);

			final BlackboardMessageContainer bbMessage = asSoapMessage(annotation.action(), message);

			final Map<String, Object> cond = Maps.newHashMap();

			cond.put("id", bbMessage.getId());

			final String subscrId = is.addSubscription(new Subscription(null, nodeManagerService.getServiceId(), Operation.UPDATE, "blackboard", cond));

			responseActions.put(subscrId, callback);

			responseClasses.put(subscrId, message.getClass());

			is.addBlackBoardMessage(serviceId, bbMessage);

		} catch (InformationServiceException e) {

			log.error("Error managing blackboard message", e);

			throw new IllegalStateException("Error managing blackboard message", e);

		}

	}

	private <T> Blackboard getMessageAnnotation(final T message) {

		if (message != null && message.getClass().isAnnotationPresent(Blackboard.class)) {

			return message.getClass().getAnnotation(Blackboard.class);

		} else if (message != null) {

			log.error("BlackboardMessage annotation is missing, class: " + message.getClass());

			throw new IllegalStateException("BlackboardMessage annotation is missing, class: " + message.getClass());

		} else {

			log.error("Message is null");

			throw new IllegalStateException("Message is null");

		}

	}

	private BlackboardMessageContainer asSoapMessage(final String action, final Object message) {

		final BlackboardMessageContainer bb = new BlackboardMessageContainer();

		bb.setAction(action);

		bb.setDate(new Date());

		bb.setId("bb-" + UUID.randomUUID());

		bb.setStatus(BlackboardActionStatus.ASSIGNED);

		bb.setJsonMessage(new Gson().toJson(message));

		return bb;

	}

	public boolean isRegisteredSubscription(final String subscriptionId) {

		return responseActions.containsKey(subscriptionId);

	}

	@SuppressWarnings("unchecked")

	public <T> void activateResponseAction(final String subscrId, final BlackboardMessageContainer bbMessage) {

		final BlackboardCallback<T> callback = (BlackboardCallback<T>) responseActions.get(subscrId);

		if (callback == null) {

			throw new RuntimeException("No action has been registered for subscription " + subscrId);

		} else {

			final BlackboardActionStatus status = bbMessage.getStatus();

			final T response = new Gson().fromJson(bbMessage.getJsonMessage(), (Class<T>) responseClasses.get(subscrId));

			invokeCallback(status, response, callback);

			if (status == BlackboardActionStatus.DONE || status == BlackboardActionStatus.FAILED) {

				complete(subscrId, bbMessage);

			}

		}

	}

	private void complete(final String subscrId, final BlackboardMessageContainer bbMessage) {

		responseActions.remove(subscrId);

		responseClasses.remove(subscrId);

		try {

			final InformationService is = serviceLocator.getService(InformationService.class);

			is.deleteSubscription(subscrId);

			is.deleteBlackBoardMessage(bbMessage);

		} catch (InformationServiceException e) {

			log.error("Error removing blackboard message", e);

			throw new RuntimeException("Error removing blackboard message", e);

		}

	}

	public <T> void invokeCallback(final BlackboardActionStatus status, final T response, final BlackboardCallback<T> callback) {

		switch (status) {

		case DONE:

			callback.onSuccess(response);

			break;

		case FAILED:

			callback.onFail(response);

			break;

		case ONGOING:

			callback.onOngoing(response);

			break;

		default:

			break;

		}

	}

}