[
https://issues.apache.org/jira/browse/ARTEMIS-2919?focusedWorklogId=518377&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-518377
]
ASF GitHub Bot logged work on ARTEMIS-2919:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 01/Dec/20 10:47
Start Date: 01/Dec/20 10:47
Worklog Time Spent: 10m
Work Description: gemmellr commented on a change in pull request #3279:
URL: https://github.com/apache/activemq-artemis/pull/3279#discussion_r533223398
##########
File path:
artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
##########
@@ -636,6 +641,19 @@ default Object getBrokerProperty(SimpleString key) {
return getObjectProperty(key);
}
+ default Message setIngressTimestamp() {
+ setBrokerProperty(HDR_INGRESS_TIMESTAMP, System.currentTimeMillis());
+ return this;
+ }
+
+ default Long getIngressTimestamp() {
+ Object result = getBrokerProperty(HDR_INGRESS_TIMESTAMP);
+ if (result == null) {
+ return null;
+ }
Review comment:
Unnecessary null check since the variable is Object typed?
##########
File path: docs/user-manual/en/address-model.md
##########
@@ -964,3 +965,13 @@ queues created on the matching address. Defaults to 0.
Read more about
`enable-metrics` determines whether or not metrics will be published to any
configured metrics plugin for the matching address. Default is `true`. Read
more
about [metrics](metrics.md).
+
+`enable-ingress-timestamp` determines whether or not the broker will add its
+timestamp to messages sent to the matching address. When `true` the exact
behavior
+will depend on the specific protocol in use. For AMQP messages the broker will
+add a `long` *message annotation* named `x-opt-ingress-timestamp-millis`. For
core
+messages (used by the core and OpenWire protocols) the broker will add a
property
Review comment:
Perhaps "`long` property" instead of just "property" ?
##########
File path:
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
##########
@@ -218,7 +230,7 @@ protected TransportConfiguration
addAcceptorConfiguration(ActiveMQServer server,
}
protected String getConfiguredProtocols() {
- return "AMQP,OPENWIRE";
+ return "AMQP,OPENWIRE,CORE";
Review comment:
This seems like the wrong place for this change. If a specific test
wants to enable CORE it can override the method, thats what it exists for.
(Same applies for the OPENWIRE one already there, but another time..)
##########
File path:
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
##########
@@ -1339,6 +1339,21 @@ public Object getBrokerProperty(SimpleString key) {
return extra.getProperty(key);
}
+ @Override
+ public final org.apache.activemq.artemis.api.core.Message
setIngressTimestamp() {
+ setMessageAnnotation(AMQPMessageSupport.INGRESS_TIMESTAMP,
System.currentTimeMillis());
+ return this;
+ }
+
+ @Override
+ public Long getIngressTimestamp() {
+ Object result =
getMessageAnnotation(AMQPMessageSupport.INGRESS_TIMESTAMP);
+ if (result == null) {
+ return null;
+ }
Review comment:
Unnecessary null check since the variable is Object typed?
##########
File path:
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AMQPMessageSupport.java
##########
@@ -99,6 +99,11 @@
*/
public static final Symbol ROUTING_TYPE =
Symbol.getSymbol("x-opt-routing-type");
+ /**
+ * Attribute used to mark the Broker-defined ingress time assigned to the
message
+ */
+ public static final Symbol INGRESS_TIMESTAMP =
Symbol.getSymbol("x-opt-ingress-timestamp-millis");
Review comment:
The units in the name seem unnecessary, even a bit ugly, given most
(all?) related values are in millis. I know you are adding that to avoid a
collision with behaviour elsewhere though, but I'd go another way personally to
achieve that.
In thinking on it, I'd suggest just "x-opt-ingress-time" instead.
(The spec-defined time related fields are named <foo>-time, albeit the field
name doesn't go on the wire at all for those)
##########
File path: docs/user-manual/en/address-model.md
##########
@@ -964,3 +965,13 @@ queues created on the matching address. Defaults to 0.
Read more about
`enable-metrics` determines whether or not metrics will be published to any
configured metrics plugin for the matching address. Default is `true`. Read
more
about [metrics](metrics.md).
+
+`enable-ingress-timestamp` determines whether or not the broker will add its
+timestamp to messages sent to the matching address. When `true` the exact
behavior
+will depend on the specific protocol in use. For AMQP messages the broker will
+add a `long` *message annotation* named `x-opt-ingress-timestamp-millis`. For
core
+messages (used by the core and OpenWire protocols) the broker will add a
property
+named `_AMQ_INGRESS_TIMESTAMP`. For STOMP messages the broker will add a frame
+header named `ingress-timestamp`. The value will be a 64-bit
+[epoch](https://en.wikipedia.org/wiki/Unix_time) timestamp with millisecond
+precision. Default is `false`.
Review comment:
I'd suggest "The value will be the number of milliseconds since the
epoch" in order to be clearer on the units (vs precision, which is different).
(Unix time is in units of seconds, and though currentTimeMillis is in units
of milliseconds, it is not guaranteed as being millisecond precision)
##########
File path:
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveTest.java
##########
@@ -1260,5 +1262,60 @@ public void testReceiveRejecting() throws Exception {
connection.close();
}
+ @Test(timeout = 60000)
+ public void testIngressTimestampSendCore() throws Exception {
+ internalTestIngressTimestamp(Protocol.CORE);
+ }
+
+ @Test(timeout = 60000)
+ public void testIngressTimestampSendAMQP() throws Exception {
+ internalTestIngressTimestamp(Protocol.AMQP);
+ }
+
+ @Test(timeout = 60000)
+ public void testIngressTimestampSendOpenWire() throws Exception {
+ internalTestIngressTimestamp(Protocol.OPENWIRE);
+ }
+
+ private void internalTestIngressTimestamp(Protocol protocol) throws
Exception {
+ server.getAddressSettingsRepository().addMatch(getQueueName(), new
AddressSettings().setEnableIngressTimestamp(true));
+ long beforeSend = System.currentTimeMillis();
+ if (protocol == Protocol.CORE) {
+ sendMessagesCore(getQueueName(), 1, true);
+ } else if (protocol == Protocol.OPENWIRE) {
+ sendMessagesOpenWire(getQueueName(), 1, true);
+ } else {
+ sendMessages(getQueueName(), 1, true);
+ }
+ long afterSend = System.currentTimeMillis();
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+ AmqpSession session = connection.createSession();
+
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
+
+ Queue queueView = getProxyToQueue(getQueueName());
+ assertEquals(1, queueView.getMessageCount());
+
+ receiver.flow(1);
+ AmqpMessage receive = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(receive);
+ instanceLog.info(receive);
+ Object ingressTimestamp =
receive.getMessageAnnotation(AMQPMessageSupport.INGRESS_TIMESTAMP.toString());
+ assertNotNull(ingressTimestamp);
+ assertTrue(ingressTimestamp instanceof Long);
+ long ingressTs = (Long) ingressTimestamp;
+ assertTrue(beforeSend <= ingressTs && afterSend >= ingressTs);
+ receiver.close();
+
+ assertEquals(1, queueView.getMessageCount());
+
+ connection.close();
+ }
+
+ private enum Protocol {
+ CORE, AMQP, OPENWIRE
+ }
+
Review comment:
I'd probably give these their own test class, its testing a specific
broker feature rather than 'general AMQP sending and receiving' that this class
is suited to, and its got its own class in the other module. That would also
help with its need for enabling the CORE protocol being specific to these tests
without having to do so for all the others here.
##########
File path:
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
##########
@@ -283,6 +284,8 @@ protected static ServerJMSMessage
processMessageAnnotations(ServerJMSMessage jms
if (delay > 0) {
jms.setLongProperty(HDR_SCHEDULED_DELIVERY_TIME.toString(),
System.currentTimeMillis() + delay);
}
+ } else if
(AMQPMessageSupport.INGRESS_TIMESTAMP.toString().equals(key) &&
entry.getValue() != null) {
+ jms.setLongProperty(HDR_INGRESS_TIMESTAMP.toString(), ((Number)
entry.getValue()).longValue());
Review comment:
I cant help think that defining a string constant (then using it in
definition of a Symbol constant if needed elsewhere) would be nicer than
calling toString() on the Symbol constant all over the place.
Its an existing issue for the surrounding bits as the diff context shows,
not suggesting you fix them here, but might be worth changing for this one if
you agree rather than building in more examples of it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 518377)
Time Spent: 2.5h (was: 2h 20m)
> Support timestamping incoming messages
> --------------------------------------
>
> Key: ARTEMIS-2919
> URL: https://issues.apache.org/jira/browse/ARTEMIS-2919
> Project: ActiveMQ Artemis
> Issue Type: New Feature
> Reporter: Justin Bertram
> Assignee: Justin Bertram
> Priority: Major
> Time Spent: 2.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
