Merge pull request #2721 from eclipse/fix-addReceivedTimestamp

Add message received timestamp header property
eclipse-kapua · Sep 26, 2019 · e7527a9 · e7527a9
2 parents c325e25 + c8ebebb
commit e7527a9
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/broker-core/src/main/java/org/eclipse/kapua/broker/core/message/MessageConstants.java b/broker-core/src/main/java/org/eclipse/kapua/broker/core/message/MessageConstants.java
@@ -37,6 +37,7 @@ private MessageConstants() {
     public static final String METRIC_NODE_ID = "nodeId";
 
     public static final String HEADER_KAPUA_CONNECTION_ID = "KAPUA_CONNECTION_ID";
+    public static final String HEADER_KAPUA_RECEIVED_TIMESTAMP = "KAPUA_RECEIVED_TIMESTAMP";
     public static final String HEADER_KAPUA_CLIENT_ID = "KAPUA_CLIENT_ID";
     public static final String HEADER_KAPUA_CONNECTOR_DEVICE_PROTOCOL = "KAPUA_DEVICE_PROTOCOL";
     public static final String HEADER_KAPUA_SESSION = "KAPUA_SESSION";

diff --git a/...er-core/src/main/java/org/eclipse/kapua/broker/core/plugin/KapuaSecurityBrokerFilter.java b/...er-core/src/main/java/org/eclipse/kapua/broker/core/plugin/KapuaSecurityBrokerFilter.java
@@ -53,6 +53,7 @@
 import org.eclipse.kapua.commons.security.KapuaSession;
 import org.eclipse.kapua.commons.setting.system.SystemSetting;
 import org.eclipse.kapua.commons.util.ClassUtil;
+import org.eclipse.kapua.commons.util.KapuaDateUtils;
 import org.eclipse.kapua.locator.KapuaLocator;
 import org.eclipse.kapua.service.account.Account;
 import org.eclipse.kapua.service.account.AccountService;
@@ -551,6 +552,7 @@ private void internalSend(ProducerBrokerExchange producerExchange, Message messa
             String message = MessageFormat.format("The caracters '+' and '#' cannot be included in a topic! Destination: {0}", messageSend.getDestination());
             throw new SecurityException(message);
         }
+        messageSend.setProperty(MessageConstants.HEADER_KAPUA_RECEIVED_TIMESTAMP, KapuaDateUtils.getKapuaSysDate().toEpochMilli());
         if (!isBrokerContext(producerExchange.getConnectionContext())) {
             KapuaSecurityContext kapuaSecurityContext = getKapuaSecurityContext(producerExchange.getConnectionContext());
             if (!messageSend.getDestination().isTemporary()) {