Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-5063 messageMoved addition in ActiveMQServerMessagePlugin

This commit is contained in:
Jean-Pascal Briquet 2024-09-25 15:51:24 +02:00 committed by clebertsuconic
parent 0c4c0542be
commit 8fc6f09195
4 changed files with 77 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -3665,6 +3665,10 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
tx.commit();
}
if (server.hasBrokerMessagePlugins()) {
server.callBrokerMessagePlugins(plugin -> plugin.messageMoved(tx, ref, reason, address, queueID, consumer, copyMessage, routingStatus));
}
return routingStatus;
}

24
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/ActiveMQServerMessagePlugin.java
View File

@ -290,4 +290,28 @@ public interface ActiveMQServerMessagePlugin extends ActiveMQServerBasePlugin {
//by default call the old method for backwards compatibility
this.messageAcknowledged(ref, reason, consumer);
}
/**
* A message has been moved
*
* @param tx The transaction associated with the move
* @param ref The ref of the message moved
* @param reason The move reason
* @param destAddress the destination address for the move operation
* @param destQueueID the destination queueID for the move operation - this field is optional and can be null
* @param consumer the consumer that moved the message - this field is optional and can be null
* @param newMessage the new message created by the move operation
* @param result routing status of the move operation
* @throws ActiveMQException
*/
default void messageMoved(final Transaction tx,
final MessageReference ref,
final AckReason reason,
final SimpleString destAddress,
final Long destQueueID,
final ServerConsumer consumer,
final Message newMessage,
final RoutingStatus result) throws ActiveMQException {
}
}

31
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/CorePluginTest.java
View File

@ -54,6 +54,7 @@ import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerPlugin;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.Wait;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -96,6 +97,7 @@ import static org.apache.activemq.artemis.tests.integration.plugin.MethodCalledV
import static org.apache.activemq.artemis.tests.integration.plugin.MethodCalledVerifier.BEFORE_UPDATE_ADDRESS;
import static org.apache.activemq.artemis.tests.integration.plugin.MethodCalledVerifier.MESSAGE_ACKED;
import static org.apache.activemq.artemis.tests.integration.plugin.MethodCalledVerifier.MESSAGE_EXPIRED;
import static org.apache.activemq.artemis.tests.integration.plugin.MethodCalledVerifier.MESSAGE_MOVED;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@ -412,6 +414,35 @@ public class CorePluginTest extends JMSTestBase {
verifier.validatePluginMethodsEquals(1, BEFORE_UPDATE_ADDRESS, AFTER_UPDATE_ADDRESS);
}
@Test
public void testMessageMoved() throws Exception {
final String queue1Name = "queue1";
final String queue2Name = "queue2";
createQueue(queue2Name);
org.apache.activemq.artemis.core.server.Queue artemisQueue = server.locateQueue(queue1Name);
org.apache.activemq.artemis.core.server.Queue artemisQueue2 = server.locateQueue(queue2Name);
conn = cf.createConnection();
conn.start();
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer prod = sess.createProducer(queue);
byte[] msgs = new byte[1024];
for (int i = 0; i < msgs.length; i++) {
msgs[i] = RandomUtil.randomByte();
}
TextMessage msg1 = sess.createTextMessage(new String(msgs));
prod.send(msg1);
conn.close();
artemisQueue.moveReferences(null, artemisQueue2.getAddress(), null);
Wait.assertEquals(1L, artemisQueue2::getMessageCount, 2000, 100);
verifier.validatePluginMethodsEquals(1, MESSAGE_MOVED);
}
private class AckPluginVerifier implements ActiveMQServerPlugin {
private BiConsumer<ServerConsumer, AckReason> assertion;

18
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/MethodCalledVerifier.java
View File

@ -87,6 +87,7 @@ public class MethodCalledVerifier implements ActiveMQServerPlugin {
public static final String AFTER_REMOVE_BINDING = "afterRemoveBinding";
public static final String MESSAGE_EXPIRED = "messageExpired";
public static final String MESSAGE_ACKED = "messageAcknowledged";
public static final String MESSAGE_MOVED = "messageMoved";
public static final String BEFORE_SEND = "beforeSend";
public static final String AFTER_SEND = "afterSend";
public static final String ON_SEND_EXCEPTION = "onSendException";
@ -299,6 +300,23 @@ public class MethodCalledVerifier implements ActiveMQServerPlugin {
methodCalled(MESSAGE_ACKED);
}
@Override
public void messageMoved(final Transaction tx,
final MessageReference ref,
final AckReason reason,
final SimpleString destAddress,
final Long destQueueID,
final ServerConsumer consumer,
final Message newMessage,
final RoutingStatus result) {
Objects.requireNonNull(ref);
Objects.requireNonNull(reason);
Objects.requireNonNull(destAddress);
Objects.requireNonNull(newMessage);
Objects.requireNonNull(result);
methodCalled(MESSAGE_MOVED);
}
@Override
public void beforeSend(ServerSession session, Transaction tx, Message message, boolean direct,
boolean noAutoCreateQueue) {