ARTEMIS-6164 MQTT retained message support for mirror and federation

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java

@@ -68,6 +68,7 @@ public String getModuleName() {
    @Override
    public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) {
       services.add(new MQTTPeriodicTasks(server, server.getScheduledPool()));
+      server.registerBrokerPlugin(new MQTTRetainMessagePlugin());
    }
 
    public class MQTTPeriodicTasks extends ActiveMQScheduledComponent {

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java

@@ -24,7 +24,6 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufUtil;
-import io.netty.buffer.EmptyByteBuf;
 import io.netty.handler.codec.mqtt.MqttFixedHeader;
 import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttProperties;
@@ -224,11 +223,6 @@ void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception
                   state.getPubRec().add(packetId);
                }
 
-               if (message.fixedHeader().isRetain()) {
-                  ByteBuf payload = message.payload();
-                  boolean reset = payload instanceof EmptyByteBuf || payload.capacity() == 0;
-                  session.getRetainMessageManager().handleRetainedMessage(serverMessage, topic, reset, tx);
-               }
                tx.commit();
             } catch (ActiveMQSecurityException e) {
                tx.rollback();

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessageManager.java

@@ -17,15 +17,10 @@
 package org.apache.activemq.artemis.core.protocol.mqtt;
 
 import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.api.core.QueueConfiguration;
 import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.persistence.StorageManager;
-import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl;
 import org.apache.activemq.artemis.core.server.BindingQueryResult;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.core.server.RoutingContext;
-import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
 
@@ -39,32 +34,6 @@ public MQTTRetainMessageManager(MQTTSession session) {
       this.session = session;
    }
 
-   /**
-    * FIXME
-    * Retained messages should be handled in the core API.  There is currently no support for retained messages
-    * at the time of writing.  Instead we handle retained messages here.  This method will create a new queue for
-    * every address that is used to store retained messages.  THere should only ever be one message in the retained
-    * message queue.  When a new subscription is created the queue should be browsed and the message copied onto
-    * the subscription queue for the consumer.  When a new retained message is received the message will be sent to
-    * the retained queue and the previous retain message consumed to remove it from the queue.
-    */
-   void handleRetainedMessage(Message messageParameter, String address, boolean reset, Transaction tx) throws Exception {
-      String retainAddress = MQTTUtil.getCoreRetainAddressFromMqttTopic(address, session.getWildcardConfiguration());
-
-      Queue queue = session.getServer().locateQueue(retainAddress);
-      if (queue == null) {
-         queue = session.getServer().createQueue(QueueConfiguration.of(retainAddress).setAutoCreated(true));
-      }
-
-      queue.deleteAllReferences();
-
-      if (!reset) {
-         StorageManager storageManager = session.getServer().getStorageManager();
-         Message message = LargeServerMessageImpl.checkLargeMessage(messageParameter, storageManager);
-         MQTTUtil.sendMessageDirectlyToQueue(storageManager, session.getServer().getPostOffice(), message.copy(storageManager.generateID()), queue, tx);
-      }
-   }
-
    void addRetainedMessagesToQueue(Queue queue, String address) throws Exception {
       // The address filter that matches all retained message queues.
       String retainAddress = MQTTUtil.getCoreRetainAddressFromMqttTopic(address, session.getWildcardConfiguration());
@@ -96,10 +65,4 @@ void addRetainedMessagesToQueue(Queue queue, String address) throws Exception {
       }
       tx.commit();
    }
-
-   private void sendToQueue(Message message, Queue queue, Transaction tx) throws Exception {
-      RoutingContext context = new RoutingContextImpl(tx);
-      queue.route(message, context);
-      session.getServer().getPostOffice().processRoute(message, context, false);
-   }
 }

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTRetainMessagePlugin.java

@@ -0,0 +1,81 @@
+/*
+ * 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.core.protocol.mqtt;
+
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.core.postoffice.RoutingStatus;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
+import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerMessagePlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.lang.invoke.MethodHandles;
+
+public class MQTTRetainMessagePlugin implements ActiveMQServerMessagePlugin {
+
+   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+   private ActiveMQServer server;
+
+   @Override
+   public void registered(ActiveMQServer server) {
+      this.server = server;
+   }
+
+   @Override
+   public void afterMessageRoute(Message message, RoutingContext context, boolean direct, boolean rejectDuplicates,
+                                 RoutingStatus result) throws ActiveMQException {
+      try {
+         Boolean isRetain = message.getBooleanProperty(MQTTUtil.MQTT_MESSAGE_RETAIN_KEY);
+         if (isRetain == null || !isRetain) {
+            return;
+         }
+
+         String address = message.getAddress();
+         if (address == null) {
+            return;
+         }
+
+         String retainQueueName = MQTTUtil.MQTT_RETAIN_ADDRESS_PREFIX + address;
+         Queue retainQueue = server.createQueue(QueueConfiguration.of(retainQueueName).setAutoCreated(true), true);
+         retainQueue.deleteAllReferences();
+
+         if (message.toCore().getBodyBufferSize() > 0) {
+            Message copy = message.copy(server.getStorageManager().generateID());
+            RoutingContext retainContext = new RoutingContextImpl(context.getTransaction()).setMirrorOption(RoutingContext.MirrorOption.disabled);
+            retainQueue.route(copy, retainContext);
+            server.getPostOffice().processRoute(copy, retainContext, false);
+         }
+      } catch (Exception e) {
+         logger.warn("Failed to handle MQTT retained message for address {}: {}", message.getAddress(), e.getMessage(), e);
+      }
+   }
+
+   @Override
+   public int hashCode() {
+      return MQTTRetainMessagePlugin.class.hashCode();
+   }
+
+   @Override
+   public boolean equals(Object obj) {
+      return obj instanceof MQTTRetainMessagePlugin;
+   }
+}

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java

@@ -592,7 +592,7 @@ public static void sendMessageDirectlyToQueue(StorageManager storageManager, Pos
          tx = new TransactionImpl(storageManager);
          tx.setAsync(true);
       }
-      RoutingContext context = new RoutingContextImpl(tx);
+      RoutingContext context = new RoutingContextImpl(tx).setMirrorOption(RoutingContext.MirrorOption.disabled);
       queue.route(message, context);
       postOffice.processRoute(message, context, false);
       if (incomingTx == null) {

artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

@@ -1251,6 +1251,10 @@ private RoutingStatus route(final Message message,
       final RoutingStatus finalStatus;
       try {
          if (status == RoutingStatus.NO_BINDINGS) {
+            if (mirrorControllerSource != null && !context.isMirrorDisabled() &&
+                addressInfo != null && addressInfo.getRoutingTypes().contains(RoutingType.MULTICAST)) {
+               mirrorControllerSource.sendMessage(context.getTransaction(), message, context);
+            }
             finalStatus = maybeSendToDLA(message, context, address, sendToDLA);
          } else {
             finalStatus = status;