Adding worker event subscriber creation to messaging factory
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/e3997fd5 Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/e3997fd5 Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/e3997fd5 Branch: refs/heads/feature-workload-mgmt Commit: e3997fd53fec2d4765fe30af592b6ca1b6d490f9 Parents: 4bb2200 Author: Ajinkya Dhamnaskar <[email protected]> Authored: Tue May 9 12:39:27 2017 -0400 Committer: Ajinkya Dhamnaskar <[email protected]> Committed: Tue May 9 12:39:27 2017 -0400 ---------------------------------------------------------------------- .../airavata/messaging/core/MessagingFactory.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/airavata/blob/e3997fd5/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingFactory.java ---------------------------------------------------------------------- diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingFactory.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingFactory.java index af925d3..e19825a 100644 --- a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingFactory.java +++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingFactory.java @@ -23,6 +23,7 @@ package org.apache.airavata.messaging.core; import org.apache.airavata.common.exception.AiravataException; import org.apache.airavata.common.utils.DBEventManagerConstants; import org.apache.airavata.common.utils.ServerSettings; +import org.apache.airavata.common.utils.WorkerConstants; import org.apache.airavata.messaging.core.impl.*; import org.apache.airavata.model.messaging.event.ExperimentStatusChangeEvent; import org.apache.airavata.model.messaging.event.JobIdentifier; @@ -84,6 +85,21 @@ public class MessagingFactory { return subscriber; } + public static Subscriber getWorkerEventSubscriber(final MessageHandler messageHandler, String taskName) throws AiravataException { + RabbitMQProperties rProperties = getProperties(); + + //FIXME: Set autoAck to false and handle possible situations + rProperties.setExchangeName(WorkerConstants.WORKER_EVENT_EXCHANGE_NAME) + .setQueueName(WorkerConstants.getQueueName(taskName)) + .setAutoAck(false); + Subscriber subscriber = new RabbitMQSubscriber(rProperties); + subscriber.listen(((connection, channel) -> new MessageConsumer(messageHandler, connection, channel)), + rProperties.getQueueName(), + new ArrayList<String>(){{add(taskName);}}); + + return subscriber; + } + public static Publisher getPublisher(Type type) throws AiravataException { RabbitMQProperties rProperties = getProperties(); Publisher publiser = null;
