Fix TODO in AmqpConsumerActor; add "to" AMQP property.

Signed-off-by: Yufei Cai <yufei.cai@bosch-si.com>
eclipse-ditto · Oct 31, 2019 · ea5fac6 · ea5fac6
1 parent 5a51063
commit ea5fac6
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 1 deletion.
diff --git a/...c/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/AmqpConsumerActor.java b/...c/main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/AmqpConsumerActor.java
@@ -286,13 +286,15 @@ private void throttleMessageConsumer() {
             return new ThrottleState(interval, nextMessages);
         });
         if (state.currentMessagePerInterval >= throttlingLimit) {
-            // TODO: add monitoring logs after merge
             log.info("Stopping message consumer, message limit of {}/{} exceeded.", throttlingLimit,
                     throttlingInterval);
             stopMessageConsumer();
             // calculate timestamp of next interval when the consumer should be restarted
             final long restartConsumerAt = (interval + 1) * throttlingInterval.toMillis();
             getSelf().tell(new RestartMessageConsumer(restartConsumerAt), ActorRef.noSender());
+            inboundMonitor.getCounter().recordFailure();
+            inboundMonitor.getLogger()
+                    .failure("Source <{0}> is rate-limited due to excessive messaging.", sourceAddress);
         }
     }
 

diff --git a/.../main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/AmqpPublisherActor.java b/.../main/java/org/eclipse/ditto/services/connectivity/messaging/amqp/AmqpPublisherActor.java
@@ -79,6 +79,7 @@ public final class AmqpPublisherActor extends BasePublisherActor<AmqpTarget> {
         JMS_HEADER_MAPPING.put("message-id", wrap(Message::setJMSMessageID));
         JMS_HEADER_MAPPING.put("reply-to", wrap((message, value) -> message.setJMSReplyTo(new JmsQueue(value))));
         JMS_HEADER_MAPPING.put("subject", wrap(Message::setJMSType));
+        JMS_HEADER_MAPPING.put("to", wrap(Message::setJMSType));
         JMS_HEADER_MAPPING.put(DittoHeaderDefinition.CONTENT_TYPE.getKey(), wrap((message, value) -> {
             if (message instanceof JmsMessage) {
                 final JmsMessageFacade facade = ((JmsMessage) message).getFacade();