Merge branch 'DEVELOP_BE' of ssh://172.18.22.160:29418/oK/PlannedGridMeasures/mics-backend into DEVELOP_BE
# Conflicts:
# devlocal.yml
# qserver.yml
diff --git a/devlocal.yml b/devlocal.yml
index 52a25d9..b00be35 100644
--- a/devlocal.yml
+++ b/devlocal.yml
@@ -11,6 +11,8 @@
exchangeName: openk-pgm-exchange
autoSetup: true
queueNames : pgm-applied-queue, pgm-forapproval-queue, pgm-canceled-queue, pgm-approved-queue, pgm-rejected-queue, pgm-requested-queue, pgm-released-queue, pgm-finished-queue, pgm-closed-queue
+ unroutedMessagesExchangeName: openk-pgm-unrouted-exchange
+ unroutedMessagesQueueName: pgm-unrouted-queue
dbConn:
driver: org.postgresql.Driver
diff --git a/devserver.yml b/devserver.yml
index e8e5d7b..2f8c4a7 100644
--- a/devserver.yml
+++ b/devserver.yml
@@ -11,6 +11,8 @@
exchangeName: openk-pgm-exchange
autoSetup: true
queueNames : pgm-applied-queue, pgm-forapproval-queue, pgm-canceled-queue, pgm-approved-queue, pgm-rejected-queue, pgm-requested-queue, pgm-released-queue, pgm-finished-queue, pgm-closed-queue
+ unroutedMessagesExchangeName: openk-pgm-unrouted-exchange
+ unroutedMessagesQueueName: pgm-unrouted-queue
dbConn:
driver: org.postgresql.Driver
diff --git a/prodserver.yml b/prodserver.yml
index aade14a..85eebe3 100644
--- a/prodserver.yml
+++ b/prodserver.yml
@@ -10,6 +10,9 @@
password: admin
exchangeName: openk-pgm-exchange
autoSetup: true
+ queueNames : pgm-applied-queue, pgm-forapproval-queue, pgm-canceled-queue, pgm-approved-queue, pgm-rejected-queue, pgm-requested-queue, pgm-released-queue, pgm-finished-queue, pgm-closed-queue
+ unroutedMessagesExchangeName: openk-pgm-unrouted-exchange
+ unroutedMessagesQueueName: pgm-unrouted-queue
dbConn:
driver: org.postgresql.Driver
diff --git a/qserver.yml b/qserver.yml
index b157874..c703af8 100644
--- a/qserver.yml
+++ b/qserver.yml
@@ -16,7 +16,8 @@
exchangeName: openk-pgm-exchange
autoSetup: true
queueNames : pgm-applied-queue, pgm-forapproval-queue, pgm-canceled-queue, pgm-approved-queue, pgm-rejected-queue, pgm-requested-queue, pgm-released-queue, pgm-finished-queue, pgm-closed-queue
-
+ unroutedMessagesExchangeName: openk-pgm-unrouted-exchange
+ unroutedMessagesQueueName: pgm-unrouted-queue
dbConn:
driver: org.postgresql.Driver
diff --git a/src/main/java/org/eclipse/openk/PlannedGridMeasuresConfiguration.java b/src/main/java/org/eclipse/openk/PlannedGridMeasuresConfiguration.java
index 4c891c4..a25d505 100644
--- a/src/main/java/org/eclipse/openk/PlannedGridMeasuresConfiguration.java
+++ b/src/main/java/org/eclipse/openk/PlannedGridMeasuresConfiguration.java
@@ -42,6 +42,11 @@
@JsonProperty
private String queueNames;
+ @JsonProperty
+ private String unroutedMessagesExchangeName;
+
+ @JsonProperty
+ private String unroutedMessagesQueueName;
public String getHost() {
@@ -96,6 +101,22 @@
public void setQueueNames(String queueNames) { this.queueNames = queueNames; }
+ public String getUnroutedMessagesExchangeName() {
+ return unroutedMessagesExchangeName;
+ }
+
+ public void setUnroutedMessagesExchangeName(String unroutedMessagesExchangeName) {
+ this.unroutedMessagesExchangeName = unroutedMessagesExchangeName;
+ }
+
+ public String getUnroutedMessagesQueueName() {
+ return unroutedMessagesQueueName;
+ }
+
+ public void setUnroutedMessagesQueueName(String unroutedMessagesQueueName) {
+ this.unroutedMessagesQueueName = unroutedMessagesQueueName;
+ }
+
public List<String> getQueueNamesAsList() {
List<String> singleQueueNameList = new ArrayList<>();
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureAppliedMessageQ.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureAppliedMessageQ.java
index f823a6b..24a54cd 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureAppliedMessageQ.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureAppliedMessageQ.java
@@ -44,8 +44,6 @@
LOGGER.debug(">>execute: Send Mail on 'beantragt'");
- createMessageQueue("pgm-applied-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "applied");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureApproved.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureApproved.java
index c3c0d77..1aa590d 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureApproved.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureApproved.java
@@ -44,8 +44,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'genehmigt'");
- createMessageQueue("pgm-approved-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "approved");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureCanceled.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureCanceled.java
index c1fd4e7..7657592 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureCanceled.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureCanceled.java
@@ -44,8 +44,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'storniert'");
- createMessageQueue("pgm-canceled-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "canceled");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureClosed.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureClosed.java
index 0aef8f3..0d9be5a 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureClosed.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureClosed.java
@@ -41,8 +41,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'closed'");
- createMessageQueue("pgm-closed-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "closed");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureFinished.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureFinished.java
index 46b6d89..26910f3 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureFinished.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureFinished.java
@@ -38,8 +38,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'Beendet'");
- createMessageQueue("pgm-finished-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "finished");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureForApproval.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureForApproval.java
index 17600ac..f7f9be1 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureForApproval.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureForApproval.java
@@ -27,8 +27,6 @@
@Override
protected void onLeaveStep(PlgmProcessSubject model) throws ProcessException {
- createMessageQueue("pgm-forapproval-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "forapproval");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRejected.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRejected.java
index a0c0b05..2941385 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRejected.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRejected.java
@@ -43,8 +43,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'zurückgewiesen'");
- createMessageQueue("pgm-rejected-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "rejected");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureReleased.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureReleased.java
index 6dffcd0..dee089f 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureReleased.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureReleased.java
@@ -41,8 +41,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'freigegeben'");
- createMessageQueue("pgm-released-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "released");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRequested.java b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRequested.java
index b2a15cd..48aaaf7 100644
--- a/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRequested.java
+++ b/src/main/java/org/eclipse/openk/core/bpmn/gridmeasure/tasks/ServiceMeasureRequested.java
@@ -40,8 +40,6 @@
}
LOGGER.debug(">>execute: Send Mail on 'Angefordert'");
- createMessageQueue("pgm-requested-queue");
-
try (Producer prod = createMessageQueueProducer()) {
prod.sendMessageAsJson(model.getGridMeasure(), "requested");
} catch (Exception e) {
diff --git a/src/main/java/org/eclipse/openk/core/controller/InitRabbitMqSetup.java b/src/main/java/org/eclipse/openk/core/controller/InitRabbitMqSetup.java
index fbc2a5d..f528449 100644
--- a/src/main/java/org/eclipse/openk/core/controller/InitRabbitMqSetup.java
+++ b/src/main/java/org/eclipse/openk/core/controller/InitRabbitMqSetup.java
@@ -29,7 +29,7 @@
private static final Logger LOGGER = Logger.getLogger(GridMeasureBackendController.class);
public void init() {
- LOGGER.debug("InitRabbitMqSetup called");
+ LOGGER.debug("InitRabbitMqSetup started");
RabbitmqConfiguration rabbitmqConfiguration = BackendConfig.getInstance().getRabbitmqConfiguration();
if (rabbitmqConfiguration == null) {
LOGGER.warn("InitRabbitMqSetup: No RabbitmqConfiguration found!");
@@ -43,8 +43,19 @@
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
+ String unroutedMessagesExchangeName = BackendConfig.getInstance().getRabbitmqConfiguration().getUnroutedMessagesExchangeName();
+ Map<String, Object> exchangeArgs = new HashMap<String, Object>();
+ exchangeArgs.put("alternate-exchange", unroutedMessagesExchangeName);
+ channel
+ .exchangeDeclare(BackendConfig.getInstance().getRabbitmqConfiguration().getExchangeName(),
+ BuiltinExchangeType.DIRECT, true, false, exchangeArgs);
+
+ String unroutedMessagesQueueName = BackendConfig.getInstance().getRabbitmqConfiguration().getUnroutedMessagesQueueName();
+
// durable = true
- channel.exchangeDeclare(rabbitmqConfiguration.getExchangeName(), BuiltinExchangeType.DIRECT, true);
+ channel.exchangeDeclare(unroutedMessagesExchangeName, BuiltinExchangeType.FANOUT, true);
+ channel.queueDeclare(unroutedMessagesQueueName, true, false, false, null);
+ channel.queueBind(unroutedMessagesQueueName, unroutedMessagesExchangeName, "");
Map<String, Object> args = new HashMap<>();
args.put("x-queue-mode", "lazy");
@@ -57,6 +68,7 @@
channel.queueBind(queueName, rabbitmqConfiguration.getExchangeName(), routingKey);
}
+ LOGGER.debug("InitRabbitMqSetup finished");
} catch (TimeoutException | IOException e) {
LOGGER.error("Error in InitRabbitMqSetup", e);
}
diff --git a/src/main/java/org/eclipse/openk/core/messagebroker/Producer.java b/src/main/java/org/eclipse/openk/core/messagebroker/Producer.java
index e2817eb..8f91e03 100644
--- a/src/main/java/org/eclipse/openk/core/messagebroker/Producer.java
+++ b/src/main/java/org/eclipse/openk/core/messagebroker/Producer.java
@@ -14,7 +14,12 @@
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
+import com.rabbitmq.client.BuiltinExchangeType;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import org.apache.log4j.Logger;
import org.eclipse.openk.api.GridMeasure;
import org.eclipse.openk.common.JsonGeneratorBase;
@@ -28,20 +33,50 @@
private final String exchangeName;
public Producer() throws ProcessException {
- this.exchangeName = BackendConfig.getInstance().getRabbitmqConfiguration().getExchangeName();
+ this.exchangeName = BackendConfig.getInstance().getRabbitmqConfiguration().getExchangeName();
}
public void sendMessageAsJson(GridMeasure gridMeasure, String routingKey) throws ProcessException {
BasicProperties basicProperties = new Builder().contentType("application/json").deliveryMode(2).priority(0).build();
String jsonString = JsonGeneratorBase.getGson().toJson(createPayload(gridMeasure));
try {
- channel.basicPublish(exchangeName,routingKey, basicProperties, jsonString.getBytes("UTF-8"));
+
+ createExchangeIfNotExists();
+ createQueueIfNotExists(routingKey);
+ channel.basicPublish(exchangeName, routingKey, basicProperties, jsonString.getBytes("UTF-8"));
} catch (IOException e) {
LOGGER.error("Error in sendMessageAsJson (RabbitMQ)", e);
throw new ProcessException("Error in sendMessageAsJson (RabbitMQ)");
}
}
+ private void createExchangeIfNotExists() throws IOException {
+ String unroutedMessagesExchangeName = BackendConfig.getInstance().getRabbitmqConfiguration().getUnroutedMessagesExchangeName();
+ Map<String, Object> exchangeArgs = new HashMap<String, Object>();
+ exchangeArgs.put("alternate-exchange", unroutedMessagesExchangeName);
+ channel
+ .exchangeDeclare(BackendConfig.getInstance().getRabbitmqConfiguration().getExchangeName(),
+ BuiltinExchangeType.DIRECT, true, false, exchangeArgs);
+
+ String unroutedMessagesQueueName = BackendConfig.getInstance().getRabbitmqConfiguration().getUnroutedMessagesQueueName();
+ // durable = true
+ channel.exchangeDeclare(unroutedMessagesExchangeName, BuiltinExchangeType.FANOUT, true);
+ channel.queueDeclare(unroutedMessagesQueueName, true, false, false, null);
+ channel.queueBind(unroutedMessagesQueueName, unroutedMessagesExchangeName, "");
+ }
+
+ private void createQueueIfNotExists(String routingKey) throws IOException {
+ List<String> queueNamesAsList = BackendConfig.getInstance().getRabbitmqConfiguration().getQueueNamesAsList();
+ Optional<String> optionalQueuename = queueNamesAsList.stream().filter(it -> it.contains("-" + routingKey)).findFirst();
+ if (optionalQueuename.isPresent()) {
+ String queueName = optionalQueuename.get();
+ Map<String, Object> args = new HashMap<>();
+ args.put("x-queue-mode", "lazy");
+ channel.queueDeclare(queueName, true, false, false, args);
+ channel.queueBind(queueName, exchangeName, routingKey);
+ }
+ }
+
public void sendMessageAsString(String message, String routingKey) throws ProcessException {
BasicProperties basicProperties = new Builder().contentType("text/plain").deliveryMode(2).priority(0).build();
try {
