Skip to content

ARTEMIS-6231 Allow auto create for federated address and queue targets #6669

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tabish121 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ARTEMIS-6231 Allow auto create for federated address and queue targets #6669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions ...ache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,10 @@

package org.apache.activemq.artemis.protocol.amqp.connect.federation;

import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_DELAY;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_MSG_COUNT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_CREATE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE_DELAY;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE_MSG_COUNT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_ADDRESS_RECEIVER;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_POLICY_NAME;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationPolicySupport.FEDERATED_ADDRESS_SOURCE_PROPERTIES;
Expand Down Expand Up @@ -154,11 +155,10 @@ protected final void doCreateReceiver() {
target.setAddress(consumerInfo.getTargetAddress());

final Map<String, Object> addressSourceProperties = new HashMap<>();
// If the remote needs to create the address then it should apply these
// settings during the create.
addressSourceProperties.put(ADDRESS_AUTO_DELETE, policy.isAutoDelete());
addressSourceProperties.put(ADDRESS_AUTO_DELETE_DELAY, policy.getAutoDeleteDelay());
addressSourceProperties.put(ADDRESS_AUTO_DELETE_MSG_COUNT, policy.getAutoDeleteMessageCount());
addressSourceProperties.put(AUTO_CREATE, policy.isAutoCreate());
addressSourceProperties.put(AUTO_DELETE, policy.isAutoDelete());
addressSourceProperties.put(AUTO_DELETE_DELAY, policy.getAutoDeleteDelay());
addressSourceProperties.put(AUTO_DELETE_MSG_COUNT, policy.getAutoDeleteMessageCount());

final Map<Symbol, Object> receiverProperties = new HashMap<>();
receiverProperties.put(FEDERATED_ADDRESS_SOURCE_PROPERTIES, addressSourceProperties);
Expand Down
27 changes: 18 additions & 9 deletions ...ivemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressSenderController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,21 @@

package org.apache.activemq.artemis.protocol.amqp.connect.federation;

import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_DELAY;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_MSG_COUNT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_CREATE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE_DELAY;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.AUTO_DELETE_MSG_COUNT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_ADDRESS_RECEIVER;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.DEFAULT_ADDRESS_AUTO_CREATE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.DEFAULT_AUTO_DELETE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.DEFAULT_AUTO_DELETE_DELAY;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.DEFAULT_AUTO_DELETE_MSG_COUNT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationPolicySupport.FEDERATED_ADDRESS_SOURCE_PROPERTIES;

import java.lang.invoke.MethodHandles;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;

Expand Down Expand Up @@ -99,15 +106,17 @@ public ServerConsumer createServerConsumer(ProtonServerSenderContext senderConte

final Map<String, Object> addressSourceProperties;

if (sender.getRemoteProperties() == null || !sender.getRemoteProperties().containsKey(FEDERATED_ADDRESS_SOURCE_PROPERTIES)) {
if (sender.getRemoteProperties() == null) {
addressSourceProperties = Collections.emptyMap();
} else {
addressSourceProperties = (Map<String, Object>) sender.getRemoteProperties().get(FEDERATED_ADDRESS_SOURCE_PROPERTIES);
addressSourceProperties = (Map<String, Object>)
Objects.requireNonNullElse(sender.getRemoteProperties().get(FEDERATED_ADDRESS_SOURCE_PROPERTIES), Collections.emptyMap());
}

final boolean autoDelete = (boolean) addressSourceProperties.getOrDefault(ADDRESS_AUTO_DELETE, false);
final long autoDeleteDelay = ((Number) addressSourceProperties.getOrDefault(ADDRESS_AUTO_DELETE_DELAY, 0)).longValue();
final long autoDeleteMsgCount = ((Number) addressSourceProperties.getOrDefault(ADDRESS_AUTO_DELETE_MSG_COUNT, 0)).longValue();
final boolean autoCreate = (boolean) addressSourceProperties.getOrDefault(AUTO_CREATE, DEFAULT_ADDRESS_AUTO_CREATE);
final boolean autoDelete = (boolean) addressSourceProperties.getOrDefault(AUTO_DELETE, DEFAULT_AUTO_DELETE);
final long autoDeleteDelay = ((Number) addressSourceProperties.getOrDefault(AUTO_DELETE_DELAY, DEFAULT_AUTO_DELETE_DELAY)).longValue();
final long autoDeleteMsgCount = ((Number) addressSourceProperties.getOrDefault(AUTO_DELETE_MSG_COUNT, DEFAULT_AUTO_DELETE_MSG_COUNT)).longValue();

// An address receiver may opt to filter on things like max message hops or no local message
// reflection so we must check for a filter here and apply it if it exists.
Expand Down Expand Up @@ -142,7 +151,7 @@ public ServerConsumer createServerConsumer(ProtonServerSenderContext senderConte
final AddressQueryResult addressQueryResult;

try {
addressQueryResult = sessionSPI.addressQuery(address, RoutingType.MULTICAST, true);
addressQueryResult = sessionSPI.addressQuery(address, RoutingType.MULTICAST, autoCreate);
} catch (ActiveMQSecurityException e) {
throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage());
} catch (ActiveMQAMQPException e) {
Expand Down
1 change: 0 additions & 1 deletion ...apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_LINK_QUIESCE_TIMEOUT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_QUIESCE_TIMEOUT;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
Expand Down
55 changes: 48 additions & 7 deletions ...org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,38 +273,43 @@ public final class AMQPFederationConstants {
*/
public static final String ADDRESS_EXCLUDES = "address-excludes";

/**
* Encodes a boolean value that indicates if federated resource auto creation should be enabled.
*/
public static final String AUTO_CREATE = "auto-create";

/**
* Encodes a boolean value that indicates if queue auto delete option should be enabled.
*/
public static final String ADDRESS_AUTO_DELETE = "auto-delete";
public static final String AUTO_DELETE = "auto-delete";

/**
* Encodes a signed long value that controls the delay before auto deletion if auto delete is enabled.
*/
public static final String ADDRESS_AUTO_DELETE_DELAY = "auto-delete-delay";
public static final String AUTO_DELETE_DELAY = "auto-delete-delay";

/**
* Encodes a signed long value that controls the message count value that allows for address auto delete.
* Encodes a signed long value that controls the message count value that allows for auto delete when enabled.
*/
public static final String ADDRESS_AUTO_DELETE_MSG_COUNT = "auto-delete-msg-count";
public static final String AUTO_DELETE_MSG_COUNT = "auto-delete-msg-count";

/**
* Encodes a signed integer value that controls the maximum number of hops allowed for federated messages.
*/
public static final String ADDRESS_MAX_HOPS = "max-hops";
public static final String MAX_HOPS = "max-hops";

/**
* Encodes boolean value that controls if the address federation should include divert bindings.
*/
public static final String ADDRESS_ENABLE_DIVERT_BINDINGS = "enable-divert-bindings";
public static final String ENABLE_DIVERT_BINDINGS = "enable-divert-bindings";

/**
* Encodes a boolean value that controls if the address federation should check the matcher value in the
* policy and if its a wildcard, create a consumer that indicates its source is the wildcard address of
* the matcher and place the normal address binding under that wildcard but add a filter to prevent any
* messages other than those of its target address from being routed to it.
*/
public static final String ADDRESS_ALLOW_WILDCARD_GROUPINGS = "allow-wildcard-groupings";
public static final String ALLOW_WILDCARD_GROUPINGS = "allow-wildcard-groupings";

/**
* Encodes a {@link Map} of String keys and values that are carried along in the federation policy (address or
Expand Down Expand Up @@ -388,4 +393,40 @@ public final class AMQPFederationConstants {
*/
public static final Symbol FEDERATION_POLICY_NAME = Symbol.valueOf("federationPolicyName");

/**
* Default for configuring a federation address receiver to request that the address resource be auto created on the remote
* if not already in existence. This can allow for multi-hop federation topologies to work in cases where the brokers
* are not configured with static addresses. The default will be to allow auto creation these as has always been done
* prior to adding this configuration option. Users can disable this now which will cause the federation to wait for
* an event indicating the target address has been created and attempt to attach again if local demand is still present.
*/
public static final boolean DEFAULT_ADDRESS_AUTO_CREATE = true;

/**
* Default for configuring a federation queue receiver to request that the queue resource (and possibly its associated address)
* be auto created on the remote if not already in existence. This can allow for multi-hop federation topologies to work in
* cases where the brokers are not configured with static addresses and queues. The default is to never auto create queues on
* the remote target and instead simply await notification that the remote queue has been created and attempt to attach if
* local demand is still present.
*/
public static final boolean DEFAULT_QUEUE_AUTO_CREATE = false;

/**
* Default auto deletion setting applied when a federation resource auto creates the queue from which messages are
* federated from on the remote peer.
*/
public static final boolean DEFAULT_AUTO_DELETE = false;

/**
* Default auto deletion delay setting applied when a federation resource auto creates the queue from which messages are
* federated from on the remote peer.
*/
public static final long DEFAULT_AUTO_DELETE_DELAY = 0L;

/**
* Default auto deletion message count setting applied when a federation resource auto creates the queue from which messages
* are federated from on the remote peer.
*/
public static final long DEFAULT_AUTO_DELETE_MSG_COUNT = 0L;

}
Loading
Loading