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

ARTEMIS-2770 Update diverts using the management API

This commit is contained in:
brusdev 2020-05-22 18:46:54 +02:00 committed by Clebert Suconic
parent d3d7d2264d
commit 9a7620fd71
19 changed files with 386 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQDivertDoesNotExistException.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.api.core;
/**
* An operation failed because an address exists on the server.
*/
public final class ActiveMQDivertDoesNotExistException extends ActiveMQException {
public ActiveMQDivertDoesNotExistException() {
super(ActiveMQExceptionType.DIVERT_DOES_NOT_EXIST);
}
public ActiveMQDivertDoesNotExistException(String msg) {
super(ActiveMQExceptionType.DIVERT_DOES_NOT_EXIST, msg);
}
}

6
artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQExceptionType.java
View File

@ -261,6 +261,12 @@ public enum ActiveMQExceptionType {
public ActiveMQException createException(String msg) {
return new ActiveMQReplicationTimeooutException(msg);
}
},
DIVERT_DOES_NOT_EXIST(221) {
@Override
public ActiveMQException createException(String msg) {
return new ActiveMQDivertDoesNotExistException(msg);
}
};
private static final Map<Integer, ActiveMQExceptionType> TYPE_MAP;

8
artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java
View File

@ -2662,4 +2662,12 @@ public interface AuditLogger extends BasicLogger {
@LogMessage(level = Logger.Level.INFO)
@Message(id = 601726, value = "User {0} failed to browse messages from queue {1}", format = Message.Format.MESSAGE_FORMAT)
void browseMessagesFailure(String user, String queueName);
static void updateDivert(Object source, Object... args) {
LOGGER.updateDivert(getCaller(), source, arrayToString(args));
}
@LogMessage(level = Logger.Level.INFO)
@Message(id = 601727, value = "User {0} is updating a divert on target resource: {1} {2}", format = Message.Format.MESSAGE_FORMAT)
void updateDivert(String user, Object source, Object... args);
}

11
artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java
View File

@ -1573,6 +1573,17 @@ public interface ActiveMQServerControl {
@Parameter(name = "transformerPropertiesAsJSON", desc = "Configuration properties of the divert's transformer in JSON form") String transformerPropertiesAsJSON,
@Parameter(name = "routingType", desc = "How should the routing-type on the diverted messages be set?") String routingType) throws Exception;
/**
* update a divert
*/
@Operation(desc = "Update a divert", impact = MBeanOperationInfo.ACTION)
void updateDivert(@Parameter(name = "name", desc = "Name of the queue") String name,
@Parameter(name = "forwardingAddress", desc = "Address to divert to") String forwardingAddress,
@Parameter(name = "filterString", desc = "Filter of the divert") String filterString,
@Parameter(name = "transformerClassName", desc = "Class name of the divert's transformer") String transformerClassName,
@Parameter(name = "transformerProperties", desc = "Configuration properties of the divert's transformer") Map<String, String> transformerProperties,
@Parameter(name = "routingType", desc = "How should the routing-type on the diverted messages be set?") String routingType) throws Exception;
@Operation(desc = "Destroy a Divert", impact = MBeanOperationInfo.ACTION)
void destroyDivert(@Parameter(name = "name", desc = "Name of the divert") String name) throws Exception;

33
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
View File

@ -97,6 +97,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.ConnectorServiceFactory;
import org.apache.activemq.artemis.core.server.Consumer;
import org.apache.activemq.artemis.core.server.ComponentConfigurationRoutingType;
import org.apache.activemq.artemis.core.server.Divert;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
@ -3415,6 +3416,38 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
}
}
@Override
public void updateDivert(final String name,
final String forwardingAddress,
final String filterString,
final String transformerClassName,
final Map<String, String> transformerProperties,
final String routingType) throws Exception {
if (AuditLogger.isEnabled()) {
AuditLogger.updateDivert(this.server, name, forwardingAddress, filterString,
transformerClassName, transformerProperties, routingType);
}
checkStarted();
clearIO();
try {
TransformerConfiguration transformerConfiguration = transformerClassName == null ? null :
new TransformerConfiguration(transformerClassName).setProperties(transformerProperties);
DivertConfiguration config = new DivertConfiguration().setName(name).setForwardingAddress(forwardingAddress).
setFilterString(filterString).setTransformerConfiguration(transformerConfiguration).
setRoutingType(ComponentConfigurationRoutingType.valueOf(routingType));
final Divert divert = server.updateDivert(config);
if (divert == null) {
throw ActiveMQMessageBundle.BUNDLE.divertDoesNotExist(config.getName());
}
} finally {
blockOnIO();
}
}
@Override
public void destroyDivert(final String name) throws Exception {
if (AuditLogger.isEnabled()) {

26
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/DivertControlImpl.java
View File

@ -19,14 +19,17 @@ package org.apache.activemq.artemis.core.management.impl;
import javax.management.MBeanAttributeInfo;
import javax.management.MBeanOperationInfo;
import java.util.Collections;
import java.util.Map;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.management.DivertControl;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.server.Divert;
import org.apache.activemq.artemis.core.server.transformer.RegisteredTransformer;
import org.apache.activemq.artemis.core.server.transformer.Transformer;
import org.apache.activemq.artemis.logs.AuditLogger;
public class DivertControlImpl extends AbstractControl implements DivertControl {
@ -37,8 +40,6 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
private final Divert divert;
private final DivertConfiguration configuration;
private final String internalNamingPrefix;
// Static --------------------------------------------------------
@ -49,11 +50,9 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
public DivertControlImpl(final Divert divert,
final StorageManager storageManager,
final DivertConfiguration configuration,
final String internalNamingPrefix) throws Exception {
super(DivertControl.class, storageManager);
this.divert = divert;
this.configuration = configuration;
this.internalNamingPrefix = internalNamingPrefix;
}
@ -64,7 +63,7 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getAddress();
return divert.getAddress().toString();
} finally {
blockOnIO();
}
@ -77,7 +76,8 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getFilterString();
Filter filter = divert.getFilter();
return filter != null ? filter.getFilterString().toString() : null;
} finally {
blockOnIO();
}
@ -90,7 +90,7 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getForwardingAddress();
return divert.getForwardAddress().toString();
} finally {
blockOnIO();
}
@ -116,7 +116,9 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getTransformerConfiguration() == null ? null : configuration.getTransformerConfiguration().getClassName();
Transformer transformer = divert.getTransformer();
return transformer != null ? (transformer instanceof RegisteredTransformer ?
((RegisteredTransformer)transformer).getTransformer() : transformer).getClass().getName() : null;
} finally {
blockOnIO();
}
@ -137,7 +139,9 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getTransformerConfiguration() == null ? null : configuration.getTransformerConfiguration().getProperties();
Transformer transformer = divert.getTransformer();
return transformer != null && transformer instanceof RegisteredTransformer ?
((RegisteredTransformer)transformer).getProperties() : Collections.emptyMap();
} finally {
blockOnIO();
}
@ -150,7 +154,7 @@ public class DivertControlImpl extends AbstractControl implements DivertControl
}
clearIO();
try {
return configuration.getRoutingType().toString();
return divert.getRoutingType().toString();
} finally {
blockOnIO();
}

8
artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/DivertBinding.java
View File

@ -31,8 +31,6 @@ public class DivertBinding implements Binding {
private final Divert divert;
private final Filter filter;
private final SimpleString uniqueName;
private final SimpleString routingName;
@ -48,8 +46,6 @@ public class DivertBinding implements Binding {
this.divert = divert;
filter = divert.getFilter();
uniqueName = divert.getUniqueName();
routingName = divert.getRoutingName();
@ -64,7 +60,7 @@ public class DivertBinding implements Binding {
@Override
public Filter getFilter() {
return filter;
return divert.getFilter();
}
@Override
@ -129,7 +125,7 @@ public class DivertBinding implements Binding {
", divert=" +
divert +
", filter=" +
filter +
divert.getFilter() +
", uniqueName=" +
uniqueName +
", routingName=" +

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

@ -26,6 +26,7 @@ import org.apache.activemq.artemis.api.core.ActiveMQClusterSecurityException;
import org.apache.activemq.artemis.api.core.ActiveMQConnectionTimedOutException;
import org.apache.activemq.artemis.api.core.ActiveMQDeleteAddressException;
import org.apache.activemq.artemis.api.core.ActiveMQDisconnectedException;
import org.apache.activemq.artemis.api.core.ActiveMQDivertDoesNotExistException;
import org.apache.activemq.artemis.api.core.ActiveMQDuplicateMetaDataException;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
@ -488,4 +489,7 @@ public interface ActiveMQMessageBundle {
@Message(id = 229230, value = "Failed to bind acceptor {0} to {1}", format = Message.Format.MESSAGE_FORMAT)
IllegalStateException failedToBind(String acceptor, String hostPort, @Cause Exception e);
@Message(id = 229231, value = "Divert Does Not Exist: {0}", format = Message.Format.MESSAGE_FORMAT)
ActiveMQDivertDoesNotExistException divertDoesNotExist(String divert);
}

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

@ -664,7 +664,9 @@ public interface ActiveMQServer extends ServiceComponent {
FederationManager getFederationManager();
void deployDivert(DivertConfiguration config) throws Exception;
Divert deployDivert(DivertConfiguration config) throws Exception;
Divert updateDivert(DivertConfiguration config) throws Exception;
void destroyDivert(SimpleString name) throws Exception;

12
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Divert.java
View File

@ -22,6 +22,8 @@ import org.apache.activemq.artemis.core.server.transformer.Transformer;
public interface Divert extends Bindable {
SimpleString getAddress();
Filter getFilter();
boolean isExclusive();
@ -33,4 +35,14 @@ public interface Divert extends Bindable {
Transformer getTransformer();
SimpleString getForwardAddress();
ComponentConfigurationRoutingType getRoutingType();
void setFilter(Filter filter);
void setTransformer(Transformer transformer);
void setForwardAddress(SimpleString forwardAddress);
void setRoutingType(ComponentConfigurationRoutingType routingType);
}

52
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
View File

@ -2544,7 +2544,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
@Override
public void deployDivert(DivertConfiguration config) throws Exception {
public Divert deployDivert(DivertConfiguration config) throws Exception {
if (config.getName() == null) {
throw ActiveMQMessageBundle.BUNDLE.divertWithNoName();
}
@ -2552,13 +2552,13 @@ public class ActiveMQServerImpl implements ActiveMQServer {
if (config.getAddress() == null) {
ActiveMQServerLogger.LOGGER.divertWithNoAddress();
return;
return null;
}
if (config.getForwardingAddress() == null) {
ActiveMQServerLogger.LOGGER.divertWithNoForwardingAddress();
return;
return null;
}
SimpleString sName = new SimpleString(config.getName());
@ -2566,7 +2566,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
if (postOffice.getBinding(sName) != null) {
ActiveMQServerLogger.LOGGER.divertBindingAlreadyExists(sName);
return;
return null;
}
SimpleString sAddress = new SimpleString(config.getAddress());
@ -2575,13 +2575,53 @@ public class ActiveMQServerImpl implements ActiveMQServer {
Filter filter = FilterImpl.createFilter(config.getFilterString());
Divert divert = new DivertImpl(new SimpleString(config.getForwardingAddress()), sName, new SimpleString(config.getRoutingName()), config.isExclusive(), filter, transformer, postOffice, storageManager, config.getRoutingType());
Divert divert = new DivertImpl(sName, sAddress, new SimpleString(config.getForwardingAddress()),
new SimpleString(config.getRoutingName()), config.isExclusive(),
filter, transformer, postOffice, storageManager, config.getRoutingType());
Binding binding = new DivertBinding(storageManager.generateID(), sAddress, divert);
postOffice.addBinding(binding);
managementService.registerDivert(divert, config);
managementService.registerDivert(divert);
return divert;
}
@Override
public Divert updateDivert(DivertConfiguration config) throws Exception {
final DivertBinding divertBinding = (DivertBinding) postOffice.getBinding(SimpleString.toSimpleString(config.getName()));
if (divertBinding == null) {
return null;
}
final Divert divert = divertBinding.getDivert();
Filter filter = FilterImpl.createFilter(config.getFilterString());
if (filter != null && !filter.equals(divert.getFilter())) {
divert.setFilter(filter);
}
if (config.getTransformerConfiguration() != null) {
getServiceRegistry().removeDivertTransformer(divert.getUniqueName().toString());
Transformer transformer = getServiceRegistry().getDivertTransformer(
config.getName(), config.getTransformerConfiguration());
divert.setTransformer(transformer);
}
if (config.getForwardingAddress() != null) {
SimpleString forwardAddress = SimpleString.toSimpleString(config.getForwardingAddress());
if (!forwardAddress.equals(config)) {
divert.setForwardAddress(forwardAddress);
}
}
if (config.getRoutingType() != null && divert.getRoutingType() != config.getRoutingType()) {
divert.setRoutingType(config.getRoutingType());
}
return divert;
}
@Override

47
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java
View File

@ -37,7 +37,9 @@ public class DivertImpl implements Divert {
private final PostOffice postOffice;
private final SimpleString forwardAddress;
private final SimpleString address;
private volatile SimpleString forwardAddress;
private final SimpleString uniqueName;
@ -45,16 +47,17 @@ public class DivertImpl implements Divert {
private final boolean exclusive;
private final Filter filter;
private volatile Filter filter;
private final Transformer transformer;
private volatile Transformer transformer;
private final StorageManager storageManager;
private final ComponentConfigurationRoutingType routingType;
private volatile ComponentConfigurationRoutingType routingType;
public DivertImpl(final SimpleString forwardAddress,
final SimpleString uniqueName,
public DivertImpl(final SimpleString uniqueName,
final SimpleString address,
final SimpleString forwardAddress,
final SimpleString routingName,
final boolean exclusive,
final Filter filter,
@ -62,6 +65,8 @@ public class DivertImpl implements Divert {
final PostOffice postOffice,
final StorageManager storageManager,
final ComponentConfigurationRoutingType routingType) {
this.address = address;
this.forwardAddress = forwardAddress;
this.uniqueName = uniqueName;
@ -153,6 +158,11 @@ public class DivertImpl implements Divert {
return exclusive;
}
@Override
public SimpleString getAddress() {
return address;
}
@Override
public Filter getFilter() {
return filter;
@ -168,6 +178,31 @@ public class DivertImpl implements Divert {
return forwardAddress;
}
@Override
public ComponentConfigurationRoutingType getRoutingType() {
return routingType;
}
@Override
public void setFilter(Filter filter) {
this.filter = filter;
}
@Override
public void setTransformer(Transformer transformer) {
this.transformer = transformer;
}
@Override
public void setForwardAddress(SimpleString forwardAddress) {
this.forwardAddress = forwardAddress;
}
@Override
public void setRoutingType(ComponentConfigurationRoutingType routingType) {
this.routingType = routingType;
}
/* (non-Javadoc)
* @see java.lang.Object#toString()
*/

3
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java
View File

@ -34,6 +34,7 @@ import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
import org.apache.activemq.artemis.core.server.ConnectorServiceFactory;
import org.apache.activemq.artemis.core.server.ServiceRegistry;
import org.apache.activemq.artemis.core.server.transformer.RegisteredTransformer;
import org.apache.activemq.artemis.core.server.transformer.Transformer;
import org.apache.activemq.artemis.spi.core.remoting.AcceptorFactory;
import org.apache.activemq.artemis.utils.ClassloadingUtil;
@ -249,7 +250,7 @@ public class ServiceRegistryImpl implements ServiceRegistry {
if (transformerConfiguration != null && transformerConfiguration.getClassName() != null) {
try {
transformer = loadClass(transformerConfiguration.getClassName());
transformer = new RegisteredTransformer(loadClass(transformerConfiguration.getClassName()));
transformer.init(Collections.unmodifiableMap(transformerConfiguration.getProperties()));
} catch (Exception e) {
throw ActiveMQMessageBundle.BUNDLE.errorCreatingTransformerClass(e, transformerConfiguration.getClassName());

3
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java
View File

@ -30,7 +30,6 @@ import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.management.impl.ActiveMQServerControlImpl;
import org.apache.activemq.artemis.core.messagecounter.MessageCounterManager;
import org.apache.activemq.artemis.core.paging.PagingManager;
@ -104,7 +103,7 @@ public interface ManagementService extends NotificationService, ActiveMQComponen
void unregisterAcceptors();
void registerDivert(Divert divert, DivertConfiguration config) throws Exception;
void registerDivert(Divert divert) throws Exception;
void unregisterDivert(SimpleString name, SimpleString address) throws Exception;

9
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
View File

@ -57,7 +57,6 @@ import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.management.impl.AcceptorControlImpl;
import org.apache.activemq.artemis.core.management.impl.ActiveMQServerControlImpl;
import org.apache.activemq.artemis.core.management.impl.AddressControlImpl;
@ -289,11 +288,11 @@ public class ManagementServiceImpl implements ManagementService {
}
@Override
public synchronized void registerDivert(final Divert divert, final DivertConfiguration config) throws Exception {
ObjectName objectName = objectNameBuilder.getDivertObjectName(divert.getUniqueName().toString(), config.getAddress());
DivertControl divertControl = new DivertControlImpl(divert, storageManager, config, messagingServer.getInternalNamingPrefix());
public synchronized void registerDivert(final Divert divert) throws Exception {
ObjectName objectName = objectNameBuilder.getDivertObjectName(divert.getUniqueName().toString(), divert.getAddress().toString());
DivertControl divertControl = new DivertControlImpl(divert, storageManager, messagingServer.getInternalNamingPrefix());
registerInJMX(objectName, divertControl);
registerInRegistry(ResourceNames.DIVERT + config.getName(), divertControl);
registerInRegistry(ResourceNames.DIVERT + divert.getUniqueName(), divertControl);
if (logger.isDebugEnabled()) {
logger.debug("registered divert " + objectName);

51
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/transformer/RegisteredTransformer.java Normal file
View File

@ -0,0 +1,51 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.server.transformer;
import java.util.Map;
import org.apache.activemq.artemis.api.core.Message;
public class RegisteredTransformer implements Transformer {
private final Transformer transformer;
private Map<String, String> properties;
public Transformer getTransformer() {
return transformer;
}
public Map<String, String> getProperties() {
return properties;
}
public RegisteredTransformer(Transformer transformer) {
this.transformer = transformer;
}
@Override
public void init(Map<String, String> properties) {
this.properties = properties;
this.transformer.init(properties);
}
@Override
public Message transform(Message message) {
return transformer.transform(message);
}
}

3
artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java
View File

@ -33,7 +33,6 @@ import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.management.impl.ActiveMQServerControlImpl;
import org.apache.activemq.artemis.core.messagecounter.MessageCounterManager;
import org.apache.activemq.artemis.core.paging.PagingManager;
@ -278,7 +277,7 @@ public class ClusteredResetMockTest extends ActiveMQTestBase {
}
@Override
public void registerDivert(Divert divert, DivertConfiguration config) throws Exception {
public void registerDivert(Divert divert) throws Exception {
}

105
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
View File

@ -34,6 +34,7 @@ import java.util.List;
import java.util.Set;
import java.util.UUID;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
@ -1320,6 +1321,110 @@ public class ActiveMQServerControlTest extends ManagementTestBase {
}
@Test
public void testCreateAndUpdateDivert() throws Exception {
String address = RandomUtil.randomString();
String name = RandomUtil.randomString();
String routingName = RandomUtil.randomString();
String forwardingAddress = RandomUtil.randomString();
String updatedForwardingAddress = RandomUtil.randomString();
ActiveMQServerControl serverControl = createManagementControl();
checkNoResource(ObjectNameBuilder.DEFAULT.getDivertObjectName(name, address));
assertEquals(0, serverControl.getDivertNames().length);
serverControl.createDivert(name.toString(), routingName, address, forwardingAddress, true, null, null);
checkResource(ObjectNameBuilder.DEFAULT.getDivertObjectName(name, address));
DivertControl divertControl = ManagementControlHelper.createDivertControl(name.toString(), address, mbeanServer);
assertEquals(name.toString(), divertControl.getUniqueName());
assertEquals(address, divertControl.getAddress());
assertEquals(forwardingAddress, divertControl.getForwardingAddress());
assertEquals(routingName, divertControl.getRoutingName());
assertTrue(divertControl.isExclusive());
assertNull(divertControl.getFilter());
assertNull(divertControl.getTransformerClassName());
String[] divertNames = serverControl.getDivertNames();
assertEquals(1, divertNames.length);
assertEquals(name, divertNames[0]);
// check that a message sent to the address is diverted exclusively
ServerLocator locator = createInVMNonHALocator();
ClientSessionFactory csf = createSessionFactory(locator);
ClientSession session = csf.createSession();
String updatedDivertQueue = RandomUtil.randomString();
String divertQueue = RandomUtil.randomString();
String queue = RandomUtil.randomString();
if (legacyCreateQueue) {
session.createQueue(updatedForwardingAddress, RoutingType.ANYCAST, updatedDivertQueue);
session.createQueue(forwardingAddress, RoutingType.ANYCAST, divertQueue);
session.createQueue(address, RoutingType.ANYCAST, queue);
} else {
session.createQueue(new QueueConfiguration(updatedDivertQueue).setAddress(updatedForwardingAddress).setRoutingType(RoutingType.ANYCAST).setDurable(false));
session.createQueue(new QueueConfiguration(divertQueue).setAddress(forwardingAddress).setRoutingType(RoutingType.ANYCAST).setDurable(false));
session.createQueue(new QueueConfiguration(queue).setAddress(address).setRoutingType(RoutingType.ANYCAST).setDurable(false));
}
ClientProducer producer = session.createProducer(address);
ClientMessage message = session.createMessage(false);
String text = RandomUtil.randomString();
message.putStringProperty("prop", text);
producer.send(message);
ClientConsumer consumer = session.createConsumer(queue);
ClientConsumer divertedConsumer = session.createConsumer(divertQueue);
ClientConsumer updatedDivertedConsumer = session.createConsumer(updatedDivertQueue);
session.start();
assertNull(consumer.receiveImmediate());
message = divertedConsumer.receive(5000);
assertNotNull(message);
assertEquals(text, message.getStringProperty("prop"));
assertNull(updatedDivertedConsumer.receiveImmediate());
serverControl.updateDivert(name.toString(), updatedForwardingAddress, null, null, null, ActiveMQDefaultConfiguration.getDefaultDivertRoutingType());
checkResource(ObjectNameBuilder.DEFAULT.getDivertObjectName(name, address));
divertControl = ManagementControlHelper.createDivertControl(name.toString(), address, mbeanServer);
assertEquals(name.toString(), divertControl.getUniqueName());
assertEquals(address, divertControl.getAddress());
assertEquals(updatedForwardingAddress, divertControl.getForwardingAddress());
assertEquals(routingName, divertControl.getRoutingName());
assertTrue(divertControl.isExclusive());
assertNull(divertControl.getFilter());
assertNull(divertControl.getTransformerClassName());
divertNames = serverControl.getDivertNames();
assertEquals(1, divertNames.length);
assertEquals(name, divertNames[0]);
// check that a message is no longer exclusively diverted
message = session.createMessage(false);
String text2 = RandomUtil.randomString();
message.putStringProperty("prop", text2);
producer.send(message);
assertNull(consumer.receiveImmediate());
assertNull(divertedConsumer.receiveImmediate());
message = updatedDivertedConsumer.receive(5000);
assertNotNull(message);
assertEquals(text2, message.getStringProperty("prop"));
consumer.close();
divertedConsumer.close();
updatedDivertedConsumer.close();
session.deleteQueue(queue);
session.deleteQueue(divertQueue);
session.deleteQueue(updatedDivertQueue);
session.close();
locator.close();
}
@Test
public void testCreateAndDestroyBridge() throws Exception {
String name = RandomUtil.randomString();

10
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
View File

@ -1238,6 +1238,16 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
proxy.invokeOperation("createDivert", name, routingName, address, forwardingAddress, exclusive, filterString, transformerClassName, transformerPropertiesAsJSON, routingType);
}
@Override
public void updateDivert(String name,
String forwardingAddress,
String filterString,
String transformerClassName,
Map<String, String> transformerProperties,
String routingType) throws Exception {
proxy.invokeOperation("updateDivert", name, forwardingAddress, filterString, transformerClassName, transformerProperties, routingType);
}
@Override
public void destroyDivert(String name) throws Exception {
proxy.invokeOperation("destroyDivert", name);