| /* |
| * 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.protocol.amqp.connect.mirror; |
| |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| 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.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement; |
| import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl; |
| import org.apache.activemq.artemis.core.server.ActiveMQComponent; |
| import org.apache.activemq.artemis.core.server.ActiveMQServer; |
| 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.AckReason; |
| import org.apache.activemq.artemis.core.server.impl.AddressInfo; |
| import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl; |
| import org.apache.activemq.artemis.core.server.mirror.MirrorController; |
| import org.apache.activemq.artemis.core.transaction.Transaction; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessageBrokerAccessor; |
| import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManager; |
| import org.apache.activemq.artemis.protocol.amqp.connect.AMQPBrokerConnection; |
| import org.apache.qpid.proton.amqp.Symbol; |
| import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations; |
| import org.apache.qpid.proton.amqp.messaging.Properties; |
| import org.apache.qpid.proton.engine.Sender; |
| import org.jboss.logging.Logger; |
| |
| import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerTarget.getControllerInUse; |
| |
| public class AMQPMirrorControllerSource extends BasicMirrorController<Sender> implements MirrorController, ActiveMQComponent { |
| |
| private static final Logger logger = Logger.getLogger(AMQPMirrorControllerSource.class); |
| |
| public static final Symbol EVENT_TYPE = Symbol.getSymbol("x-opt-amq-mr-ev-type"); |
| public static final Symbol ADDRESS = Symbol.getSymbol("x-opt-amq-mr-adr"); |
| public static final Symbol QUEUE = Symbol.getSymbol("x-opt-amq-mr-qu"); |
| public static final Symbol BROKER_ID = Symbol.getSymbol("x-opt-amq-bkr-id"); |
| |
| // Events: |
| public static final Symbol ADD_ADDRESS = Symbol.getSymbol("addAddress"); |
| public static final Symbol DELETE_ADDRESS = Symbol.getSymbol("deleteAddress"); |
| public static final Symbol CREATE_QUEUE = Symbol.getSymbol("createQueue"); |
| public static final Symbol DELETE_QUEUE = Symbol.getSymbol("deleteQueue"); |
| public static final Symbol POST_ACK = Symbol.getSymbol("postAck"); |
| |
| // Delivery annotation property used on mirror control routing and Ack |
| public static final Symbol INTERNAL_ID = Symbol.getSymbol("x-opt-amq-mr-id"); |
| public static final Symbol INTERNAL_DESTINATION = Symbol.getSymbol("x-opt-amq-mr-dst"); |
| |
| // Capabilities |
| public static final Symbol MIRROR_CAPABILITY = Symbol.getSymbol("amq.mirror"); |
| public static final Symbol QPID_DISPATCH_WAYPOINT_CAPABILITY = Symbol.valueOf("qd.waypoint"); |
| |
| public static final SimpleString INTERNAL_ID_EXTRA_PROPERTY = SimpleString.toSimpleString(INTERNAL_ID.toString()); |
| public static final SimpleString INTERNAL_BROKER_ID_EXTRA_PROPERTY = SimpleString.toSimpleString(BROKER_ID.toString()); |
| |
| private static final ThreadLocal<MirrorControlRouting> mirrorControlRouting = ThreadLocal.withInitial(() -> new MirrorControlRouting(null)); |
| |
| final Queue snfQueue; |
| final ActiveMQServer server; |
| final ReferenceNodeStore idSupplier; |
| final boolean acks; |
| final boolean addQueues; |
| final boolean deleteQueues; |
| private final AMQPBrokerConnection brokerConnection; |
| |
| final AMQPMirrorBrokerConnectionElement replicaConfig; |
| |
| boolean started; |
| |
| @Override |
| public void start() throws Exception { |
| } |
| |
| @Override |
| public void stop() throws Exception { |
| } |
| |
| @Override |
| public boolean isStarted() { |
| return started; |
| } |
| |
| public AMQPMirrorControllerSource(ProtonProtocolManager protonProtocolManager, Queue snfQueue, ActiveMQServer server, AMQPMirrorBrokerConnectionElement replicaConfig, |
| AMQPBrokerConnection brokerConnection) { |
| super(server); |
| this.replicaConfig = replicaConfig; |
| this.snfQueue = snfQueue; |
| this.server = server; |
| this.idSupplier = protonProtocolManager.getReferenceIDSupplier(); |
| this.addQueues = replicaConfig.isQueueCreation(); |
| this.deleteQueues = replicaConfig.isQueueRemoval(); |
| this.acks = replicaConfig.isMessageAcknowledgements(); |
| this.brokerConnection = brokerConnection; |
| } |
| |
| public Queue getSnfQueue() { |
| return snfQueue; |
| } |
| |
| public AMQPBrokerConnection getBrokerConnection() { |
| return brokerConnection; |
| } |
| |
| @Override |
| public void addAddress(AddressInfo addressInfo) throws Exception { |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " addAddress " + addressInfo); |
| } |
| |
| if (getControllerInUse() != null && !addressInfo.isInternal()) { |
| return; |
| } |
| if (addQueues) { |
| Message message = createMessage(addressInfo.getName(), null, ADD_ADDRESS, null, addressInfo.toJSON()); |
| route(server, message); |
| } |
| } |
| |
| @Override |
| public void deleteAddress(AddressInfo addressInfo) throws Exception { |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " deleteAddress " + addressInfo); |
| } |
| if (invalidTarget(getControllerInUse()) || addressInfo.isInternal()) { |
| return; |
| } |
| if (deleteQueues) { |
| Message message = createMessage(addressInfo.getName(), null, DELETE_ADDRESS, null, addressInfo.toJSON()); |
| route(server, message); |
| } |
| } |
| |
| @Override |
| public void createQueue(QueueConfiguration queueConfiguration) throws Exception { |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " createQueue " + queueConfiguration); |
| } |
| if (invalidTarget(getControllerInUse()) || queueConfiguration.isInternal()) { |
| if (logger.isTraceEnabled()) { |
| logger.trace("Rejecting ping pong on create " + queueConfiguration + " as isInternal=" + queueConfiguration.isInternal() + " and mirror target = " + getControllerInUse()); |
| } |
| return; |
| } |
| if (addQueues) { |
| Message message = createMessage(queueConfiguration.getAddress(), queueConfiguration.getName(), CREATE_QUEUE, null, queueConfiguration.toJSON()); |
| route(server, message); |
| } |
| } |
| |
| @Override |
| public void deleteQueue(SimpleString address, SimpleString queue) throws Exception { |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " deleteQueue " + address + "/" + queue); |
| } |
| |
| if (invalidTarget(getControllerInUse())) { |
| return; |
| } |
| |
| if (deleteQueues) { |
| Message message = createMessage(address, queue, DELETE_QUEUE, null, queue.toString()); |
| route(server, message); |
| } |
| } |
| |
| private boolean invalidTarget(MirrorController controller) { |
| return controller != null && sameNode(getRemoteMirrorId(), controller.getRemoteMirrorId()); |
| } |
| |
| private boolean sameNode(String remoteID, String sourceID) { |
| return (remoteID != null && sourceID != null && remoteID.equals(sourceID)); |
| } |
| |
| @Override |
| public void sendMessage(Message message, RoutingContext context, List<MessageReference> refs) { |
| if (invalidTarget(context.getMirrorSource())) { |
| if (logger.isTraceEnabled()) { |
| logger.trace("server " + server + " is discarding send to avoid infinite loop (reflection with the mirror)"); |
| } |
| return; |
| } |
| |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " send message " + message); |
| } |
| |
| try { |
| context.setReusable(false); |
| |
| MessageReference ref = MessageReference.Factory.createReference(message, snfQueue); |
| String nodeID = setProtocolData(idSupplier, ref); |
| if (nodeID != null && nodeID.equals(getRemoteMirrorId())) { |
| if (logger.isTraceEnabled()) { |
| logger.trace("Message " + message + "already belonged to the node, " + getRemoteMirrorId() + ", it won't circle send"); |
| } |
| return; |
| } |
| snfQueue.refUp(ref); |
| refs.add(ref); |
| message.usageUp(); |
| |
| |
| if (message.isDurable() && snfQueue.isDurable()) { |
| PostOfficeImpl.storeDurableReference(server.getStorageManager(), message, context.getTransaction(), snfQueue, true); |
| } |
| |
| } catch (Throwable e) { |
| logger.warn(e.getMessage(), e); |
| } |
| } |
| |
| public static void validateProtocolData(ReferenceNodeStore referenceIDSupplier, MessageReference ref, SimpleString snfAddress) { |
| if (ref.getProtocolData() == null && !ref.getMessage().getAddressSimpleString().equals(snfAddress)) { |
| setProtocolData(referenceIDSupplier, ref); |
| } |
| } |
| |
| /** This method will return the brokerID used by the message */ |
| private static String setProtocolData(ReferenceNodeStore referenceIDSupplier, MessageReference ref) { |
| Map<Symbol, Object> daMap = new HashMap<>(); |
| DeliveryAnnotations deliveryAnnotations = new DeliveryAnnotations(daMap); |
| |
| String brokerID = referenceIDSupplier.getServerID(ref); |
| |
| // getListID will return null when the message was generated on this broker. |
| // on this case we do not send the brokerID, and the ControllerTarget will get the information from the link. |
| // this is just to safe a few bytes and some processing on the wire. |
| if (brokerID != null) { |
| // not sending the brokerID, will make the other side to get the brokerID from the remote link's property |
| daMap.put(BROKER_ID, brokerID); |
| } |
| |
| long id = referenceIDSupplier.getID(ref); |
| |
| daMap.put(INTERNAL_ID, id); |
| String address = ref.getMessage().getAddress(); |
| if (address != null) { // this is the message that was set through routing |
| Properties amqpProperties = getProperties(ref.getMessage()); |
| if (amqpProperties == null || !address.equals(amqpProperties.getTo())) { |
| // We set the internal destination property only if we need to |
| // otherwise we just use the one already set over Properties |
| daMap.put(INTERNAL_DESTINATION, ref.getMessage().getAddress()); |
| } |
| } |
| ref.setProtocolData(deliveryAnnotations); |
| |
| return brokerID; |
| } |
| |
| private static Properties getProperties(Message message) { |
| if (message instanceof AMQPMessage) { |
| return AMQPMessageBrokerAccessor.getCurrentProperties((AMQPMessage)message); |
| } else { |
| return null; |
| } |
| } |
| |
| @Override |
| public void postAcknowledge(MessageReference ref, AckReason reason) throws Exception { |
| |
| MirrorController controllerInUse = getControllerInUse(); |
| |
| if (!acks || ref.getQueue().isMirrorController()) { // we don't call postACK on snfqueues, otherwise we would get infinite loop because of this feedback/ |
| return; |
| } |
| |
| if (invalidTarget(controllerInUse)) { |
| return; |
| } |
| |
| if ((ref.getQueue() != null && (ref.getQueue().isInternalQueue() || ref.getQueue().isMirrorController()))) { |
| if (logger.isDebugEnabled()) { |
| logger.debug(server + " rejecting postAcknowledge queue=" + ref.getQueue().getName() + ", ref=" + ref + " to avoid infinite loop with the mirror (reflection)"); |
| } |
| return; |
| } |
| |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " postAcknowledge " + ref); |
| } |
| |
| String nodeID = idSupplier.getServerID(ref); // notice the brokerID will be null for any message generated on this broker. |
| long internalID = idSupplier.getID(ref); |
| if (logger.isTraceEnabled()) { |
| logger.trace(server + " sending ack message from server " + nodeID + " with messageID=" + internalID); |
| } |
| Message message = createMessage(ref.getQueue().getAddress(), ref.getQueue().getName(), POST_ACK, nodeID, internalID); |
| route(server, message); |
| ref.getMessage().usageDown(); |
| } |
| |
| private Message createMessage(SimpleString address, SimpleString queue, Object event, String brokerID, Object body) { |
| return AMQPMirrorMessageFactory.createMessage(snfQueue.getAddress().toString(), address, queue, event, brokerID, body); |
| } |
| |
| public static void route(ActiveMQServer server, Message message) throws Exception { |
| message.setMessageID(server.getStorageManager().generateID()); |
| MirrorControlRouting ctx = mirrorControlRouting.get(); |
| ctx.clear(); |
| server.getPostOffice().route(message, ctx, false); |
| } |
| |
| private static class MirrorControlRouting extends RoutingContextImpl { |
| |
| MirrorControlRouting(Transaction transaction) { |
| super(transaction); |
| } |
| |
| @Override |
| public boolean isMirrorController() { |
| return true; |
| } |
| } |
| } |
