* The {@code String}-based values will be converted to the proper value types based on the underlying property. For
* example, if you pass the value "TRUE" for the key "auto-created" the {@code String} "TRUE" will be converted to
@@ -214,6 +218,8 @@ public final class BridgeConfiguration implements Serializable {
setCallTimeout(Long.parseLong(value));
} else if (key.equals(ROUTING_TYPE)) {
setRoutingType(ComponentConfigurationRoutingType.valueOf(value));
+ } else if (key.equals(CONCURRENCY)) {
+ setConcurrency(Integer.parseInt(value));
}
}
return this;
@@ -487,6 +493,21 @@ public final class BridgeConfiguration implements Serializable {
return this;
}
+ /**
+ * @return the bridge concurrency
+ */
+ public int getConcurrency() {
+ return concurrency;
+ }
+
+ /**
+ * @param concurrency the bridge concurrency to set
+ */
+ public BridgeConfiguration setConcurrency(int concurrency) {
+ this.concurrency = concurrency;
+ return this;
+ }
+
/**
* At this point this is only changed on testcases
* The bridge shouldn't be sending blocking anyways
@@ -546,6 +567,7 @@ public final class BridgeConfiguration implements Serializable {
builder.add(MAX_RETRY_INTERVAL, getMaxRetryInterval());
builder.add(MIN_LARGE_MESSAGE_SIZE, getMinLargeMessageSize());
builder.add(CALL_TIMEOUT, getCallTimeout());
+ builder.add(CONCURRENCY, getConcurrency());
// complex fields (only serialize if value is not null)
@@ -639,6 +661,7 @@ public final class BridgeConfiguration implements Serializable {
result = prime * result + ((transformerConfiguration == null) ? 0 : transformerConfiguration.hashCode());
result = prime * result + (useDuplicateDetection ? 1231 : 1237);
result = prime * result + ((user == null) ? 0 : user.hashCode());
+ result = prime * result + concurrency;
return result;
}
@@ -722,6 +745,8 @@ public final class BridgeConfiguration implements Serializable {
return false;
} else if (!user.equals(other.user))
return false;
+ if (concurrency != other.concurrency)
+ return false;
return true;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
index c5fd448935..6994860c47 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
@@ -2085,6 +2085,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
ComponentConfigurationRoutingType routingType = ComponentConfigurationRoutingType.valueOf(getString(brNode, "routing-type", ActiveMQDefaultConfiguration.getDefaultBridgeRoutingType(), Validators.COMPONENT_ROUTING_TYPE));
+ int concurrency = getInteger(brNode, "concurrency", ActiveMQDefaultConfiguration.getDefaultBridgeConcurrency(), Validators.GT_ZERO);
NodeList clusterPassNodes = brNode.getElementsByTagName("password");
String password = null;
@@ -2151,7 +2152,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
.setHA(ha)
.setUser(user)
.setPassword(password)
- .setRoutingType(routingType);
+ .setRoutingType(routingType)
+ .setConcurrency(concurrency);
if (!staticConnectorNames.isEmpty()) {
config.setStaticConnectors(staticConnectorNames);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
index ac483024e6..05008d0244 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
@@ -475,16 +475,22 @@ public final class ClusterManager implements ActiveMQComponent {
clusterLocators.add(serverLocator);
- Bridge bridge = new BridgeImpl(serverLocator, config.getInitialConnectAttempts(), config.getReconnectAttempts(), config.getReconnectAttemptsOnSameNode(), config.getRetryInterval(), config.getRetryIntervalMultiplier(), config.getMaxRetryInterval(), nodeManager.getUUID(), new SimpleString(config.getName()), queue, executorFactory.getExecutor(), FilterImpl.createFilter(config.getFilterString()), SimpleString.toSimpleString(config.getForwardingAddress()), scheduledExecutor, transformer, config.isUseDuplicateDetection(), config.getUser(), config.getPassword(), server, config.getRoutingType());
+ for (int i = 0; i < config.getConcurrency(); i++) {
+ String name = (config.getName() + "-" + i);
+ Bridge bridge = new BridgeImpl(serverLocator, config.getInitialConnectAttempts(), config.getReconnectAttempts(),
+ config.getReconnectAttemptsOnSameNode(), config.getRetryInterval(), config.getRetryIntervalMultiplier(),
+ config.getMaxRetryInterval(), nodeManager.getUUID(), new SimpleString(name), queue,
+ executorFactory.getExecutor(), FilterImpl.createFilter(config.getFilterString()),
+ SimpleString.toSimpleString(config.getForwardingAddress()), scheduledExecutor, transformer,
+ config.isUseDuplicateDetection(), config.getUser(), config.getPassword(), server,
+ config.getRoutingType());
+ bridges.put(name, bridge);
+ managementService.registerBridge(bridge, config);
+ bridge.start();
- bridges.put(config.getName(), bridge);
-
- managementService.registerBridge(bridge, config);
-
- bridge.start();
-
- if (server.hasBrokerBridgePlugins()) {
- server.callBrokerBridgePlugins(plugin -> plugin.afterDeployBridge(bridge));
+ if (server.hasBrokerBridgePlugins()) {
+ server.callBrokerBridgePlugins(plugin -> plugin.afterDeployBridge(bridge));
+ }
}
}
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index 6bf6097494..6788727bb4 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -1676,6 +1676,16 @@
+
+