ARTEMIS-3711 support AMQ_SCHEDULED_DELAY for OpenWire clients

apache · Mar 9, 2022 · 2e1a373 · 2e1a373
1 parent f6372d8
commit 2e1a373
Show file tree

Hide file tree

Showing 2 changed files with 76 additions and 0 deletions.
diff --git a/...ain/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java b/...ain/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
@@ -38,6 +38,7 @@
 import java.util.zip.InflaterOutputStream;
 
 import com.google.common.io.BaseEncoding;
+import org.apache.activemq.ScheduledMessage;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
 import org.apache.activemq.artemis.api.core.ICoreMessage;
@@ -224,6 +225,13 @@ public static org.apache.activemq.artemis.api.core.Message inbound(final Message
          coreMessage.putStringProperty(AMQ_MSG_ORIG_DESTINATION, origDest.getQualifiedName());
       }
 
+      final Object scheduledDelay = messageSend.getProperties().get(ScheduledMessage.AMQ_SCHEDULED_DELAY);
+      if (scheduledDelay != null && scheduledDelay instanceof Long) {
+         coreMessage.putLongProperty(org.apache.activemq.artemis.api.core.Message.HDR_SCHEDULED_DELIVERY_TIME, System.currentTimeMillis() + ((Long) scheduledDelay));
+         // this property may have already been copied, but we don't need it anymore
+         coreMessage.removeProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY);
+      }
+
       return coreMessage;
    }
 

diff --git a/...va/org/apache/activemq/artemis/tests/integration/openwire/OpenWireScheduledDelayTest.java b/...va/org/apache/activemq/artemis/tests/integration/openwire/OpenWireScheduledDelayTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.openwire;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.util.Map;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ScheduledMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.RandomUtil;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OpenWireScheduledDelayTest extends OpenWireTestBase {
+
+   @Override
+   protected void configureAddressSettings(Map<String, AddressSettings> addressSettingsMap) {
+      addressSettingsMap.put("#", new AddressSettings().setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ")));
+   }
+
+   @Test
+   public void testScheduledDelay() throws Exception {
+      final String QUEUE_NAME = RandomUtil.randomString();
+      final long DELAY = 10000;
+
+      ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
+      Connection connection = connectionFactory.createConnection();
+      connection.start();
+      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+      Destination destination = session.createQueue(QUEUE_NAME);
+      MessageProducer producer = session.createProducer(destination);
+      Message message = session.createMessage();
+      message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, DELAY);
+      final long ETA = System.currentTimeMillis() + DELAY;
+      producer.send(message);
+
+      MessageConsumer consumer = session.createConsumer(destination);
+      Message received = consumer.receive(DELAY + 250);
+      assertNotNull(received);
+      Assert.assertTrue(System.currentTimeMillis() >= ETA);
+
+      connection.close();
+   }
+}
+
+