| /* 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.qpid.test.utils; |
| |
| import java.io.File; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.PrintStream; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.jms.BytesMessage; |
| import javax.jms.Connection; |
| import javax.jms.Destination; |
| import javax.jms.JMSException; |
| import javax.jms.MapMessage; |
| import javax.jms.Message; |
| import javax.jms.MessageConsumer; |
| import javax.jms.MessageProducer; |
| import javax.jms.ObjectMessage; |
| import javax.jms.Queue; |
| import javax.jms.Session; |
| import javax.jms.StreamMessage; |
| import javax.jms.TextMessage; |
| import javax.jms.Topic; |
| import javax.naming.Context; |
| import javax.naming.InitialContext; |
| import javax.naming.NamingException; |
| |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.log4j.Logger; |
| |
| import org.apache.qpid.AMQException; |
| import org.apache.qpid.client.AMQConnectionFactory; |
| import org.apache.qpid.client.AMQConnectionURL; |
| import org.apache.qpid.client.AMQQueue; |
| import org.apache.qpid.client.AMQTopic; |
| import org.apache.qpid.exchange.ExchangeDefaults; |
| import org.apache.qpid.framing.AMQShortString; |
| import org.apache.qpid.jms.BrokerDetails; |
| import org.apache.qpid.jms.ConnectionURL; |
| import org.apache.qpid.server.Broker; |
| import org.apache.qpid.server.BrokerOptions; |
| import org.apache.qpid.server.configuration.BrokerProperties; |
| import org.apache.qpid.server.configuration.updater.TaskExecutor; |
| import org.apache.qpid.server.configuration.updater.TaskExecutorImpl; |
| import org.apache.qpid.server.model.ConfiguredObject; |
| import org.apache.qpid.server.model.Port; |
| import org.apache.qpid.server.model.Protocol; |
| import org.apache.qpid.server.model.VirtualHostNode; |
| import org.apache.qpid.server.store.MemoryConfigurationStore; |
| import org.apache.qpid.server.virtualhostnode.AbstractVirtualHostNode; |
| import org.apache.qpid.server.virtualhostnode.JsonVirtualHostNode; |
| import org.apache.qpid.url.URLSyntaxException; |
| import org.apache.qpid.util.FileUtils; |
| import org.apache.qpid.util.SystemUtils; |
| |
| /** |
| * Qpid base class for system testing test cases. |
| */ |
| public class QpidBrokerTestCase extends QpidTestCase |
| { |
| private TaskExecutor _taskExecutor; |
| |
| public enum BrokerType |
| { |
| EXTERNAL /** Test case relies on a Broker started independently of the test-suite */, |
| INTERNAL /** Test case starts an embedded broker within this JVM */, |
| SPAWNED /** Test case spawns a new broker as a separate process */ |
| } |
| |
| public static final String GUEST_USERNAME = "guest"; |
| public static final String GUEST_PASSWORD = "guest"; |
| |
| protected final static String QpidHome = System.getProperty("QPID_HOME"); |
| private final File _configFile = new File(System.getProperty("broker.config")); |
| private File _logConfigFile; |
| protected final String _brokerStoreType = System.getProperty("broker.config-store-type", "JSON"); |
| protected static final Logger _logger = Logger.getLogger(QpidBrokerTestCase.class); |
| protected static final int LOGMONITOR_TIMEOUT = 5000; |
| |
| protected long RECEIVE_TIMEOUT = 1000l; |
| |
| private Map<String, String> _propertiesSetForBroker = new HashMap<String, String>(); |
| |
| private Map<Integer, TestBrokerConfiguration> _brokerConfigurations; |
| |
| protected static final String INDEX = "index"; |
| protected static final String CONTENT = "content"; |
| |
| private static final String DEFAULT_INITIAL_CONTEXT = "org.apache.qpid.jndi.PropertiesFileInitialContextFactory"; |
| |
| static |
| { |
| String initialContext = System.getProperty(Context.INITIAL_CONTEXT_FACTORY); |
| |
| if (initialContext == null || initialContext.length() == 0) |
| { |
| System.setProperty(Context.INITIAL_CONTEXT_FACTORY, DEFAULT_INITIAL_CONTEXT); |
| } |
| } |
| |
| // system properties |
| private static final String BROKER_LANGUAGE = "broker.language"; |
| protected static final String BROKER_TYPE = "broker.type"; |
| private static final String BROKER_COMMAND = "broker.command"; |
| private static final String BROKER_COMMAND_PLATFORM = "broker.command." + SystemUtils.getOSConfigSuffix(); |
| private static final String BROKER_CLEAN_BETWEEN_TESTS = "broker.clean.between.tests"; |
| private static final String BROKER_VERSION = "broker.version"; |
| protected static final String BROKER_READY = "broker.ready"; |
| private static final String BROKER_STOPPED = "broker.stopped"; |
| private static final String TEST_OUTPUT = "test.output"; |
| private static final String BROKER_LOG_INTERLEAVE = "broker.log.interleave"; |
| private static final String BROKER_LOG_PREFIX = "broker.log.prefix"; |
| private static final String BROKER_PERSITENT = "broker.persistent"; |
| public static final String PROFILE_USE_SSL = "profile.use_ssl"; |
| |
| public static final int DEFAULT_PORT_VALUE = 5672; |
| public static final int DEFAULT_SSL_PORT_VALUE = 5671; |
| public static final int DEFAULT_JMXPORT_REGISTRYSERVER = 8999; |
| public static final int JMXPORT_CONNECTORSERVER_OFFSET = 100; |
| public static final int DEFAULT_HTTP_MANAGEMENT_PORT_VALUE = 8080; |
| public static final int DEFAULT_HTTPS_MANAGEMENT_PORT_VALUE = 8443; |
| |
| public static final String TEST_AMQP_PORT_PROTOCOLS_PROPERTY="test.amqp_port_protocols"; |
| |
| // values |
| protected static final String JAVA = "java"; |
| protected static final String CPP = "cpp"; |
| |
| protected static final String QPID_HOME = "QPID_HOME"; |
| |
| public static final int DEFAULT_PORT = Integer.getInteger("test.port", DEFAULT_PORT_VALUE); |
| public static final int FAILING_PORT = Integer.parseInt(System.getProperty("test.port.alt")); |
| public static final int DEFAULT_MANAGEMENT_PORT = Integer.getInteger("test.mport", DEFAULT_JMXPORT_REGISTRYSERVER); |
| public static final int DEFAULT_HTTP_MANAGEMENT_PORT = Integer.getInteger("test.hport", DEFAULT_HTTP_MANAGEMENT_PORT_VALUE); |
| public static final int DEFAULT_HTTPS_MANAGEMENT_PORT = Integer.getInteger("test.hsport", DEFAULT_HTTPS_MANAGEMENT_PORT_VALUE); |
| public static final int DEFAULT_SSL_PORT = Integer.getInteger("test.port.ssl", DEFAULT_SSL_PORT_VALUE); |
| |
| protected String _brokerLanguage = System.getProperty(BROKER_LANGUAGE, JAVA); |
| protected BrokerType _brokerType = BrokerType.valueOf(System.getProperty(BROKER_TYPE, "").toUpperCase()); |
| |
| private static final String BROKER_COMMAND_TEMPLATE = System.getProperty(BROKER_COMMAND_PLATFORM, System.getProperty(BROKER_COMMAND)); |
| protected BrokerCommandHelper _brokerCommandHelper = new BrokerCommandHelper(BROKER_COMMAND_TEMPLATE); |
| |
| private Boolean _brokerCleanBetweenTests = Boolean.getBoolean(BROKER_CLEAN_BETWEEN_TESTS); |
| private final Protocol _brokerProtocol = Protocol.valueOf("AMQP_" + System.getProperty(BROKER_VERSION, " ").substring(1)); |
| protected String _output = System.getProperty(TEST_OUTPUT, System.getProperty("java.io.tmpdir")); |
| protected Boolean _brokerPersistent = Boolean.getBoolean(BROKER_PERSITENT); |
| |
| protected static String _brokerLogPrefix = System.getProperty(BROKER_LOG_PREFIX,"BROKER: "); |
| protected static boolean _interleaveBrokerLog = Boolean.valueOf(System.getProperty(BROKER_LOG_INTERLEAVE,"true")); |
| |
| protected File _outputFile; |
| |
| protected PrintStream _testcaseOutputStream; |
| |
| protected Map<Integer, BrokerHolder> _brokers = new HashMap<Integer, BrokerHolder>(); |
| |
| protected InitialContext _initialContext; |
| protected AMQConnectionFactory _connectionFactory; |
| |
| // the connections created for a given test |
| protected List<Connection> _connections = new ArrayList<Connection>(); |
| public static final String QUEUE = "queue"; |
| public static final String TOPIC = "topic"; |
| public static final String MANAGEMENT_MODE_PASSWORD = "mm_password"; |
| |
| /** Map to hold test defined environment properties */ |
| private Map<String, String> _env; |
| |
| /** Ensure our messages have some sort of size */ |
| protected static final int DEFAULT_MESSAGE_SIZE = 1024; |
| |
| /** Size to create our message*/ |
| private int _messageSize = DEFAULT_MESSAGE_SIZE; |
| /** Type of message*/ |
| protected enum MessageType |
| { |
| BYTES, |
| MAP, |
| OBJECT, |
| STREAM, |
| TEXT |
| } |
| private MessageType _messageType = MessageType.TEXT; |
| |
| public QpidBrokerTestCase() |
| { |
| super(); |
| _brokerConfigurations = new HashMap<Integer, TestBrokerConfiguration>(); |
| initialiseLogConfigFile(); |
| } |
| |
| public TestBrokerConfiguration getBrokerConfiguration(int port) |
| { |
| int actualPort = getPort(port); |
| |
| synchronized (_brokerConfigurations) |
| { |
| TestBrokerConfiguration configuration = _brokerConfigurations.get(actualPort); |
| if (configuration == null) |
| { |
| configuration = createBrokerConfiguration(actualPort); |
| } |
| return configuration; |
| } |
| } |
| |
| public TestBrokerConfiguration getBrokerConfiguration() |
| { |
| return getBrokerConfiguration(DEFAULT_PORT); |
| } |
| |
| public TestBrokerConfiguration createBrokerConfiguration(int port) |
| { |
| int actualPort = getPort(port); |
| if(_taskExecutor == null) |
| { |
| _taskExecutor = new TaskExecutorImpl(); |
| _taskExecutor.start(); |
| } |
| TestBrokerConfiguration configuration = new TestBrokerConfiguration(_brokerStoreType, _configFile.getAbsolutePath(), _taskExecutor); |
| synchronized (_brokerConfigurations) |
| { |
| _brokerConfigurations.put(actualPort, configuration); |
| } |
| if (actualPort != DEFAULT_PORT) |
| { |
| configuration.setObjectAttribute(Port.class, TestBrokerConfiguration.ENTRY_NAME_AMQP_PORT, Port.PORT, actualPort); |
| configuration.setObjectAttribute(Port.class, TestBrokerConfiguration.ENTRY_NAME_RMI_PORT, Port.PORT, getManagementPort(actualPort)); |
| configuration.setObjectAttribute(Port.class, TestBrokerConfiguration.ENTRY_NAME_JMX_PORT, Port.PORT, getManagementPort(actualPort) + JMXPORT_CONNECTORSERVER_OFFSET); |
| |
| String workDir = System.getProperty("QPID_WORK") + File.separator + TestBrokerConfiguration.ENTRY_NAME_VIRTUAL_HOST + File.separator + actualPort; |
| configuration.setObjectAttribute(VirtualHostNode.class, TestBrokerConfiguration.ENTRY_NAME_VIRTUAL_HOST, JsonVirtualHostNode.STORE_PATH, workDir); |
| } |
| |
| return configuration; |
| } |
| |
| private void initialiseLogConfigFile() |
| { |
| _logConfigFile = new File(LOG4J_CONFIG_FILE_PATH); |
| if(!_logConfigFile.exists()) |
| { |
| throw new RuntimeException("Log config file " + _logConfigFile.getAbsolutePath() + " does not exist"); |
| } |
| } |
| |
| public Logger getLogger() |
| { |
| return QpidBrokerTestCase._logger; |
| } |
| |
| @Override |
| public void runBare() throws Throwable |
| { |
| String qname = getClass().getName() + "." + getName(); |
| |
| // Initialize this for each test run |
| _env = new HashMap<String, String>(); |
| |
| PrintStream oldOut = System.out; |
| PrintStream oldErr = System.err; |
| PrintStream out = null; |
| PrintStream err = null; |
| |
| boolean redirected = _output != null && _output.length() > 0; |
| if (redirected) |
| { |
| _outputFile = new File(String.format("%s/TEST-%s.out", _output, qname)); |
| out = new PrintStream(new FileOutputStream(_outputFile), true); |
| err = new PrintStream(String.format("%s/TEST-%s.err", _output, qname)); |
| |
| System.setOut(out); |
| System.setErr(err); |
| |
| if (_interleaveBrokerLog) |
| { |
| _testcaseOutputStream = out; |
| } |
| else |
| { |
| _testcaseOutputStream = new PrintStream(new FileOutputStream(String |
| .format("%s/TEST-%s.broker.out", _output, qname)), true); |
| } |
| } |
| |
| try |
| { |
| super.runBare(); |
| } |
| catch (Exception e) |
| { |
| _logger.error("exception", e); |
| throw e; |
| } |
| finally |
| { |
| stopAllBrokers(); |
| |
| // reset properties used in the test |
| revertSystemProperties(); |
| revertLoggingLevels(); |
| |
| if(_brokerCleanBetweenTests) |
| { |
| final String qpidWork = System.getProperty("QPID_WORK"); |
| cleanBrokerWork(qpidWork); |
| createBrokerWork(qpidWork); |
| } |
| |
| _logger.info("========== stop " + getTestName() + " =========="); |
| |
| if (redirected) |
| { |
| System.setErr(oldErr); |
| System.setOut(oldOut); |
| err.close(); |
| out.close(); |
| if (!_interleaveBrokerLog) |
| { |
| _testcaseOutputStream.close(); |
| } |
| } |
| } |
| } |
| |
| @Override |
| protected void setUp() throws Exception |
| { |
| super.setUp(); |
| _taskExecutor = new TaskExecutorImpl(); |
| _taskExecutor.start(); |
| if (!_configFile.exists()) |
| { |
| fail("Unable to test without config file:" + _configFile); |
| } |
| |
| startBroker(); |
| } |
| |
| /** |
| * Return the management port in use by the broker on this main port |
| * |
| * @param mainPort the broker's main port. |
| * |
| * @return the management port that corresponds to the broker on the given port |
| */ |
| protected int getManagementPort(int mainPort) |
| { |
| return mainPort + (DEFAULT_MANAGEMENT_PORT - DEFAULT_PORT); |
| } |
| |
| /** |
| * The returned set of port numbers is only a guess because it assumes no ports have been overridden |
| * using system properties. |
| */ |
| protected Set<Integer> guessAllPortsUsedByBroker(int mainPort) |
| { |
| Set<Integer> ports = new HashSet<Integer>(); |
| int managementPort = getManagementPort(mainPort); |
| int connectorServerPort = managementPort + JMXPORT_CONNECTORSERVER_OFFSET; |
| |
| ports.add(mainPort); |
| ports.add(managementPort); |
| ports.add(connectorServerPort); |
| ports.add(DEFAULT_SSL_PORT); |
| |
| return ports; |
| } |
| |
| /** |
| * Get the Port that is use by the current broker |
| * |
| * @return the current port |
| */ |
| protected int getPort() |
| { |
| return getPort(0); |
| } |
| |
| protected int getPort(int port) |
| { |
| if (!_brokerType.equals(BrokerType.EXTERNAL)) |
| { |
| return port == 0 ? DEFAULT_PORT : port; |
| } |
| else |
| { |
| return port; |
| } |
| } |
| |
| public void startBroker() throws Exception |
| { |
| startBroker(0); |
| } |
| |
| public void startBroker(int port) throws Exception |
| { |
| startBroker(port, false); |
| } |
| |
| public void startBroker(int port, boolean managementMode) throws Exception |
| { |
| int actualPort = getPort(port); |
| TestBrokerConfiguration configuration = getBrokerConfiguration(actualPort); |
| startBroker(actualPort, configuration, managementMode); |
| } |
| |
| protected File getBrokerCommandLog4JFile() |
| { |
| return _logConfigFile; |
| } |
| |
| protected void setBrokerCommandLog4JFile(File file) |
| { |
| _logConfigFile = file; |
| _logger.info("Modified log config file to: " + file); |
| } |
| |
| public void startBroker(int port, TestBrokerConfiguration testConfiguration) throws Exception |
| { |
| startBroker(port, testConfiguration, false); |
| } |
| |
| public void startBroker(int port, TestBrokerConfiguration testConfiguration, boolean managementMode) throws Exception |
| { |
| port = getPort(port); |
| |
| if(_brokers.get(port) != null) |
| { |
| throw new IllegalStateException("There is already an existing broker running on port " + port); |
| } |
| |
| Set<Integer> portsUsedByBroker = guessAllPortsUsedByBroker(port); |
| String testConfig = saveTestConfiguration(port, testConfiguration); |
| |
| if (_brokerType.equals(BrokerType.INTERNAL) && !existingInternalBroker()) |
| { |
| setSystemProperty(BrokerProperties.PROPERTY_USE_CUSTOM_RMI_SOCKET_FACTORY, "false"); |
| BrokerOptions options = new BrokerOptions(); |
| |
| options.setConfigurationStoreType(_brokerStoreType); |
| options.setConfigurationStoreLocation(testConfig); |
| options.setManagementMode(managementMode); |
| if (managementMode) |
| { |
| options.setManagementModePassword(MANAGEMENT_MODE_PASSWORD); |
| } |
| |
| //Set the log config file, relying on the log4j.configuration system property |
| //set on the JVM by the JUnit runner task in module.xml. |
| options.setLogConfigFileLocation(_logConfigFile.getAbsolutePath()); |
| |
| Broker broker = new Broker(); |
| _logger.info("Starting internal broker (same JVM)"); |
| broker.startup(options); |
| |
| _brokers.put(port, new InternalBrokerHolder(broker, System.getProperty("QPID_WORK"), portsUsedByBroker)); |
| } |
| else if (!_brokerType.equals(BrokerType.EXTERNAL)) |
| { |
| // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests |
| final String qpidWork = getQpidWork(_brokerType, port); |
| |
| String[] cmd = _brokerCommandHelper.getBrokerCommand(port, testConfig, _brokerStoreType, _logConfigFile); |
| if (managementMode) |
| { |
| String[] newCmd = new String[cmd.length + 3]; |
| System.arraycopy(cmd, 0, newCmd, 0, cmd.length); |
| newCmd[cmd.length] = "-mm"; |
| newCmd[cmd.length + 1] = "-mmpass"; |
| newCmd[cmd.length + 2] = MANAGEMENT_MODE_PASSWORD; |
| cmd = newCmd; |
| } |
| _logger.info("Starting spawn broker using command: " + StringUtils.join(cmd, ' ')); |
| ProcessBuilder pb = new ProcessBuilder(cmd); |
| pb.redirectErrorStream(true); |
| Map<String, String> processEnv = pb.environment(); |
| String qpidHome = System.getProperty(QPID_HOME); |
| processEnv.put(QPID_HOME, qpidHome); |
| //Augment Path with bin directory in QPID_HOME. |
| boolean foundPath = false; |
| final String pathEntry = qpidHome + File.separator + "bin"; |
| for(Map.Entry<String,String> entry : processEnv.entrySet()) |
| { |
| if(entry.getKey().equalsIgnoreCase("path")) |
| { |
| entry.setValue(entry.getValue().concat(File.pathSeparator + pathEntry)); |
| foundPath = true; |
| } |
| } |
| if(!foundPath) |
| { |
| processEnv.put("PATH", pathEntry); |
| |
| } |
| //Add the test name to the broker run. |
| // DON'T change PNAME, qpid.stop needs this value. |
| processEnv.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + getTestName() + "\""); |
| processEnv.put("QPID_WORK", qpidWork); |
| |
| // Use the environment variable to set amqj.logging.level for the broker |
| // The value used is a 'server' value in the test configuration to |
| // allow a differentiation between the client and broker logging levels. |
| if (System.getProperty("amqj.server.logging.level") != null) |
| { |
| setBrokerEnvironment("AMQJ_LOGGING_LEVEL", System.getProperty("amqj.server.logging.level")); |
| } |
| |
| // Add all the environment settings the test requested |
| if (!_env.isEmpty()) |
| { |
| for (Map.Entry<String, String> entry : _env.entrySet()) |
| { |
| processEnv.put(entry.getKey(), entry.getValue()); |
| } |
| } |
| |
| String qpidOpts = ""; |
| |
| // a synchronized hack to avoid adding into QPID_OPTS the values |
| // of JVM properties "test.virtualhosts" and "test.config" set by a concurrent startup process |
| synchronized (_propertiesSetForBroker) |
| { |
| // Add default test logging levels that are used by the log4j-test |
| // Use the convenience methods to push the current logging setting |
| // in to the external broker's QPID_OPTS string. |
| setSystemProperty("amqj.protocol.logging.level"); |
| setSystemProperty("root.logging.level"); |
| |
| setSystemProperty("test.port"); |
| setSystemProperty("test.mport"); |
| setSystemProperty("test.cport"); |
| setSystemProperty("test.hport"); |
| setSystemProperty("test.hsport"); |
| setSystemProperty("test.port.ssl"); |
| setSystemProperty("test.port.alt"); |
| setSystemProperty("test.port.alt.ssl"); |
| setSystemProperty("test.amqp_port_protocols"); |
| |
| setSystemProperty("virtualhostnode.type"); |
| setSystemProperty("virtualhostnode.context.blueprint"); |
| |
| // Add all the specified system properties to QPID_OPTS |
| if (!_propertiesSetForBroker.isEmpty()) |
| { |
| for (String key : _propertiesSetForBroker.keySet()) |
| { |
| qpidOpts += " -D" + key + "=" + _propertiesSetForBroker.get(key); |
| } |
| } |
| } |
| if (processEnv.containsKey("QPID_OPTS")) |
| { |
| qpidOpts = processEnv.get("QPID_OPTS") + qpidOpts; |
| } |
| processEnv.put("QPID_OPTS", qpidOpts); |
| |
| // cpp broker requires that the work directory is created |
| createBrokerWork(qpidWork); |
| |
| Process process = pb.start(); |
| |
| Piper p = new Piper(process.getInputStream(), |
| _testcaseOutputStream, |
| System.getProperty(BROKER_READY), |
| System.getProperty(BROKER_STOPPED), |
| _interleaveBrokerLog ? _brokerLogPrefix : null); |
| |
| p.start(); |
| StringBuilder cmdLine = new StringBuilder(cmd[0]); |
| for(int i = 1; i< cmd.length; i++) |
| { |
| cmdLine.append(' '); |
| cmdLine.append(cmd[i]); |
| } |
| |
| SpawnedBrokerHolder holder = new SpawnedBrokerHolder(process, qpidWork, portsUsedByBroker, cmdLine.toString()); |
| if (!p.await(30, TimeUnit.SECONDS)) |
| { |
| _logger.info("broker failed to become ready (" + p.getReady() + "):" + p.getStopLine()); |
| String threadDump = holder.dumpThreads(); |
| if (!threadDump.isEmpty()) |
| { |
| _logger.info("the result of a try to capture thread dump:" + threadDump); |
| } |
| //Ensure broker has stopped |
| process.destroy(); |
| cleanBrokerWork(qpidWork); |
| throw new RuntimeException("broker failed to become ready:" |
| + p.getStopLine()); |
| } |
| |
| try |
| { |
| //test that the broker is still running and hasn't exited unexpectedly |
| int exit = process.exitValue(); |
| _logger.info("broker aborted: " + exit); |
| cleanBrokerWork(qpidWork); |
| throw new RuntimeException("broker aborted: " + exit); |
| } |
| catch (IllegalThreadStateException e) |
| { |
| // this is expect if the broker started successfully |
| } |
| |
| _brokers.put(port, holder); |
| } |
| } |
| |
| private boolean existingInternalBroker() |
| { |
| for(BrokerHolder holder : _brokers.values()) |
| { |
| if(holder instanceof InternalBrokerHolder) |
| { |
| return true; |
| } |
| } |
| |
| return false; |
| } |
| |
| private String getQpidWork(BrokerType broker, int port) |
| { |
| if (!broker.equals(BrokerType.EXTERNAL)) |
| { |
| return System.getProperty("QPID_WORK")+ "/" + port; |
| } |
| |
| return System.getProperty("QPID_WORK"); |
| } |
| |
| public String getTestConfigFile() |
| { |
| return getTestConfigFile(getPort()); |
| } |
| |
| public String getTestConfigFile(int port) |
| { |
| return _output + File.separator + getTestQueueName() + "-" + port + "-config"; |
| } |
| |
| protected String getPathRelativeToWorkingDirectory(String file) |
| { |
| File configLocation = new File(file); |
| File workingDirectory = new File(System.getProperty("user.dir")); |
| |
| _logger.debug("Converting path to be relative to working directory: " + file); |
| |
| try |
| { |
| String configPath = configLocation.getAbsolutePath(); |
| String workingDirectoryPath = workingDirectory.getCanonicalPath(); |
| if (SystemUtils.isWindows()) |
| { |
| configPath = configPath.toLowerCase(); |
| workingDirectoryPath = workingDirectoryPath.toLowerCase(); |
| } |
| if(!configPath.startsWith(workingDirectoryPath)) |
| { |
| throw new RuntimeException("Provided path is not a child of the working directory: " + workingDirectoryPath); |
| } |
| |
| String substring = configPath.replace(workingDirectoryPath, "").substring(1); |
| _logger.debug("Converted relative path: " + substring); |
| |
| return substring; |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException("Problem while converting to relative path", e); |
| } |
| } |
| |
| protected String saveTestConfiguration(int port, TestBrokerConfiguration testConfiguration) |
| { |
| String testConfig = getTestConfigFile(port); |
| String relative = getPathRelativeToWorkingDirectory(testConfig); |
| if (testConfiguration != null && !testConfiguration.isSaved()) |
| { |
| _logger.info("Saving test broker configuration at: " + testConfig); |
| testConfiguration.save(new File(testConfig)); |
| testConfiguration.setSaved(true); |
| } |
| return relative; |
| } |
| |
| protected void cleanBrokerWork(final String qpidWork) |
| { |
| if (qpidWork != null) |
| { |
| _logger.info("Cleaning broker work dir: " + qpidWork); |
| |
| File file = new File(qpidWork); |
| if (file.exists()) |
| { |
| final boolean success = FileUtils.delete(file, true); |
| if(!success) |
| { |
| throw new RuntimeException("Failed to recursively delete beneath : " + file); |
| } |
| } |
| } |
| } |
| |
| protected void createBrokerWork(final String qpidWork) |
| { |
| if (qpidWork != null) |
| { |
| final File dir = new File(qpidWork); |
| dir.mkdirs(); |
| if (!dir.isDirectory()) |
| { |
| throw new RuntimeException("Failed to created Qpid work directory : " + qpidWork); |
| } |
| } |
| } |
| |
| public void stopBroker() |
| { |
| stopBroker(0); |
| } |
| |
| public void stopAllBrokers() |
| { |
| boolean exceptionOccured = false; |
| Set<Integer> runningBrokerPorts = new HashSet<Integer>(getBrokerPortNumbers()); |
| for (int brokerPortNumber : runningBrokerPorts) |
| { |
| if (!stopBrokerSafely(brokerPortNumber)) |
| { |
| exceptionOccured = true; |
| } |
| } |
| if (exceptionOccured) |
| { |
| throw new RuntimeException("Exception occurred on stopping of test broker. Please, examine logs for details"); |
| } |
| } |
| |
| protected boolean stopBrokerSafely(int brokerPortNumber) |
| { |
| boolean success = true; |
| BrokerHolder broker = _brokers.get(brokerPortNumber); |
| try |
| { |
| stopBroker(brokerPortNumber); |
| } |
| catch(Exception e) |
| { |
| success = false; |
| _logger.error("Failed to stop broker " + broker + " at port " + brokerPortNumber, e); |
| if (broker != null) |
| { |
| // save the thread dump in case of dead locks |
| try |
| { |
| _logger.error("Broker " + broker + " thread dump:" + broker.dumpThreads()); |
| } |
| finally |
| { |
| // try to kill broker |
| try |
| { |
| broker.kill(); |
| } |
| catch(Exception killException) |
| { |
| // ignore |
| } |
| } |
| } |
| } |
| return success; |
| } |
| |
| public void stopBroker(int port) |
| { |
| if (isBrokerPresent(port)) |
| { |
| port = getPort(port); |
| |
| _logger.info("stopping broker on port : " + port); |
| BrokerHolder broker = _brokers.remove(port); |
| broker.shutdown(); |
| } |
| } |
| |
| public void killBroker() |
| { |
| killBroker(0); |
| } |
| |
| public void killBroker(int port) |
| { |
| if (isBrokerPresent(port)) |
| { |
| port = getPort(port); |
| |
| _logger.info("killing broker on port : " + port); |
| BrokerHolder broker = _brokers.remove(port); |
| broker.kill(); |
| } |
| } |
| |
| public boolean isBrokerPresent(int port) |
| { |
| port = getPort(port); |
| |
| return _brokers.containsKey(port); |
| } |
| |
| public BrokerHolder getBroker(int port) throws Exception |
| { |
| port = getPort(port); |
| return _brokers.get(port); |
| } |
| |
| public Set<Integer> getBrokerPortNumbers() |
| { |
| return new HashSet<Integer>(_brokers.keySet()); |
| } |
| |
| /** |
| * Creates a new virtual host node in broker configuration for given broker port |
| * @param brokerPort broker port |
| * @param virtualHostNodeName virtual host node name |
| */ |
| protected void createTestVirtualHostNode(int brokerPort, String virtualHostNodeName, boolean withBlueprint) |
| { |
| String storeType = getTestProfileVirtualHostNodeType(); |
| String storeDir = null; |
| |
| if (System.getProperty("profile", "").startsWith("java-dby-mem")) |
| { |
| storeDir = ":memory:"; |
| } |
| else if (!MemoryConfigurationStore.TYPE.equals(storeType)) |
| { |
| storeDir = "${QPID_WORK}" + File.separator + virtualHostNodeName + File.separator + brokerPort; |
| } |
| |
| // add new virtual host node with vhost blueprint configuration to the broker store |
| Map<String, Object> attributes = new HashMap<String, Object>(); |
| attributes.put(VirtualHostNode.NAME, virtualHostNodeName); |
| attributes.put(VirtualHostNode.TYPE, storeType); |
| if (storeDir != null) |
| { |
| attributes.put(JsonVirtualHostNode.STORE_PATH, storeDir); |
| } |
| |
| if (withBlueprint) |
| { |
| final String blueprint = getTestProfileVirtualHostNodeBlueprint(); |
| |
| attributes.put(ConfiguredObject.CONTEXT, |
| Collections.singletonMap(AbstractVirtualHostNode.VIRTUALHOST_BLUEPRINT_CONTEXT_VAR, |
| blueprint)); |
| } |
| |
| int port = getPort(brokerPort); |
| getBrokerConfiguration(port).addObjectConfiguration(VirtualHostNode.class, attributes); |
| } |
| |
| protected void createTestVirtualHostNode(int brokerPort, String virtualHostNodeName) |
| { |
| createTestVirtualHostNode(brokerPort, virtualHostNodeName, true); |
| } |
| |
| /** |
| * Set a System property that is to be applied only to the external test |
| * broker. |
| * |
| * This is a convenience method to enable the setting of a -Dproperty=value |
| * entry in QPID_OPTS |
| * |
| * This is only useful for the External Java Broker tests. |
| * |
| * @param property the property name |
| * @param value the value to set the property to |
| */ |
| protected void setBrokerOnlySystemProperty(String property, String value) |
| { |
| synchronized (_propertiesSetForBroker) |
| { |
| if (!_propertiesSetForBroker.containsKey(property)) |
| { |
| _propertiesSetForBroker.put(property, value); |
| } |
| } |
| } |
| |
| /** |
| * Set a System (-D) property for this test run. |
| * |
| * This convenience method copies the current VMs System Property |
| * for the external VM Broker. |
| * |
| * @param property the System property to set |
| */ |
| protected void setSystemProperty(String property) |
| { |
| String value = System.getProperty(property); |
| if (value != null) |
| { |
| setSystemProperty(property, value); |
| } |
| } |
| |
| /** |
| * Set a System property for the duration of this test. |
| * |
| * When the test run is complete the value will be reverted. |
| * |
| * The values set using this method will also be propagated to the external |
| * Java Broker via a -D value defined in QPID_OPTS. |
| * |
| * If the value should not be set on the broker then use |
| * setTestClientSystemProperty(). |
| * |
| * @param property the property to set |
| * @param value the new value to use |
| */ |
| protected void setSystemProperty(String property, String value) |
| { |
| synchronized(_propertiesSetForBroker) |
| { |
| // Record the value for the external broker |
| if (value == null) |
| { |
| _propertiesSetForBroker.remove(property); |
| } |
| else |
| { |
| _propertiesSetForBroker.put(property, value); |
| } |
| } |
| //Set the value for the test client vm aswell. |
| setTestClientSystemProperty(property, value); |
| } |
| |
| /** |
| * Set a System property for the client (and broker if using the same vm) of this test. |
| * |
| * @param property The property to set |
| * @param value the value to set it to. |
| */ |
| protected void setTestClientSystemProperty(String property, String value) |
| { |
| setTestSystemProperty(property, value); |
| } |
| |
| /** |
| * Restore the System property values that were set before this test run. |
| */ |
| protected void revertSystemProperties() |
| { |
| revertTestSystemProperties(); |
| |
| // We don't change the current VMs settings for Broker only properties |
| // so we can just clear this map |
| _propertiesSetForBroker.clear(); |
| } |
| |
| /** |
| * Add an environment variable for the external broker environment |
| * |
| * @param property the property to set |
| * @param value the value to set it to |
| */ |
| protected void setBrokerEnvironment(String property, String value) |
| { |
| _env.put(property, value); |
| } |
| |
| /** |
| * Check whether the broker is an 0.8 |
| * |
| * @return true if the broker is an 0_8 version, false otherwise. |
| */ |
| public boolean isBroker08() |
| { |
| return _brokerProtocol.equals(Protocol.AMQP_0_8); |
| } |
| |
| public boolean isBroker010() |
| { |
| return _brokerProtocol.equals(Protocol.AMQP_0_10); |
| } |
| |
| public Protocol getBrokerProtocol() |
| { |
| return _brokerProtocol; |
| } |
| |
| protected boolean isJavaBroker() |
| { |
| return _brokerLanguage.equals("java"); |
| } |
| |
| protected boolean isCppBroker() |
| { |
| return _brokerLanguage.equals("cpp"); |
| } |
| |
| protected boolean isExternalBroker() |
| { |
| return !isInternalBroker(); |
| } |
| |
| protected boolean isInternalBroker() |
| { |
| return _brokerType.equals(BrokerType.INTERNAL); |
| } |
| |
| protected boolean isBrokerStorePersistent() |
| { |
| return _brokerPersistent; |
| } |
| |
| public void restartBroker() throws Exception |
| { |
| restartBroker(0); |
| } |
| |
| public void restartBroker(int port) throws Exception |
| { |
| stopBroker(port); |
| startBroker(port); |
| } |
| |
| /** |
| * we assume that the environment is correctly set |
| * i.e. -Djava.naming.provider.url="..//example010.properties" |
| * |
| * @return an initial context |
| * |
| * @throws NamingException if there is an error getting the context |
| */ |
| public InitialContext getInitialContext() throws NamingException |
| { |
| if (_initialContext == null) |
| { |
| _initialContext = new InitialContext(); |
| } |
| return _initialContext; |
| } |
| |
| /** |
| * Get the default connection factory for the currently used broker |
| * Default factory is "local" |
| * |
| * @return A connection factory |
| * |
| * @throws Exception if there is an error getting the factory |
| */ |
| public AMQConnectionFactory getConnectionFactory() throws NamingException |
| { |
| if (_connectionFactory == null) |
| { |
| if (Boolean.getBoolean(PROFILE_USE_SSL)) |
| { |
| _connectionFactory = getConnectionFactory("default.ssl"); |
| } |
| else |
| { |
| _connectionFactory = getConnectionFactory("default"); |
| } |
| } |
| return _connectionFactory; |
| } |
| |
| /** |
| * Get a connection factory for the currently used broker |
| * |
| * @param factoryName The factory name |
| * |
| * @return A connection factory |
| * |
| * @throws Exception if there is an error getting the factory |
| */ |
| public AMQConnectionFactory getConnectionFactory(String factoryName) throws NamingException |
| { |
| return (AMQConnectionFactory) getInitialContext().lookup(factoryName); |
| } |
| |
| public Connection getConnection() throws JMSException, NamingException |
| { |
| return getConnection(GUEST_USERNAME, GUEST_PASSWORD); |
| } |
| |
| public Connection getConnectionWithOptions(Map<String, String> options) |
| throws URLSyntaxException, NamingException, JMSException |
| { |
| ConnectionURL curl = new AMQConnectionURL(getConnectionFactory().getConnectionURLString()); |
| for(Map.Entry<String,String> entry : options.entrySet()) |
| { |
| curl.setOption(entry.getKey(), entry.getValue()); |
| } |
| curl = new AMQConnectionURL(curl.toString()); |
| |
| curl.setUsername(GUEST_USERNAME); |
| curl.setPassword(GUEST_PASSWORD); |
| return getConnection(curl); |
| } |
| |
| |
| public Connection getConnection(ConnectionURL url) throws JMSException |
| { |
| _logger.debug("get connection for " + url.getURL()); |
| Connection connection = new AMQConnectionFactory(url).createConnection(url.getUsername(), url.getPassword()); |
| |
| _connections.add(connection); |
| |
| return connection; |
| } |
| |
| /** |
| * Get a connection (remote or in-VM) |
| * |
| * @param username The user name |
| * @param password The user password |
| * |
| * @return a newly created connection |
| * |
| * @throws Exception if there is an error getting the connection |
| */ |
| public Connection getConnection(String username, String password) throws JMSException, NamingException |
| { |
| _logger.debug("get connection for username " + username); |
| Connection con = getConnectionFactory().createConnection(username, password); |
| //add the connection in the list of connections |
| _connections.add(con); |
| return con; |
| } |
| |
| protected Connection getClientConnection(String username, String password, String id) throws JMSException, URLSyntaxException, AMQException, NamingException |
| { |
| _logger.debug("get connection for id " + id); |
| Connection con = getConnectionFactory().createConnection(username, password, id); |
| //add the connection in the list of connections |
| _connections.add(con); |
| return con; |
| } |
| |
| /** |
| * Useful, for example, to avoid the connection being automatically closed in {@link #tearDown()} |
| * if it has deliberately been put into an error state already. |
| */ |
| protected void forgetConnection(Connection connection) |
| { |
| _logger.debug("Forgetting about connection " + connection); |
| boolean removed = _connections.remove(connection); |
| assertTrue( |
| "The supplied connection " + connection + " should have been one that I already know about", |
| removed); |
| } |
| |
| /** |
| * Return a uniqueName for this test. |
| * In this case it returns a queue Named by the TestCase and TestName |
| * |
| * @return String name for a queue |
| */ |
| protected String getTestQueueName() |
| { |
| return getClass().getSimpleName() + "-" + getName(); |
| } |
| |
| /** |
| * Return a Queue specific for this test. |
| * Uses getTestQueueName() as the name of the queue |
| * @return |
| */ |
| public Queue getTestQueue() |
| { |
| return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, getTestQueueName()); |
| } |
| |
| /** |
| * Return a Topic specific for this test. |
| * Uses getTestQueueName() as the name of the topic |
| * @return |
| */ |
| public Topic getTestTopic() |
| { |
| return new AMQTopic(AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_NAME), getTestQueueName()); |
| } |
| |
| @Override |
| protected void tearDown() throws java.lang.Exception |
| { |
| super.tearDown(); |
| |
| // close all the connections used by this test. |
| for (Connection c : _connections) |
| { |
| c.close(); |
| } |
| if(_taskExecutor != null) |
| { |
| _taskExecutor.stop(); |
| } |
| } |
| |
| /** |
| * Consume all the messages in the specified queue. Helper to ensure |
| * persistent tests don't leave data behind. |
| * |
| * @param queue the queue to purge |
| * |
| * @return the count of messages drained |
| * |
| * @throws Exception if a problem occurs |
| */ |
| protected int drainQueue(Queue queue) throws Exception |
| { |
| Connection connection = getConnection(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| MessageConsumer consumer = session.createConsumer(queue); |
| |
| connection.start(); |
| |
| int count = 0; |
| while (consumer.receive(1000) != null) |
| { |
| count++; |
| } |
| |
| connection.close(); |
| |
| return count; |
| } |
| |
| /** |
| * Send messages to the given destination. |
| * |
| * If session is transacted then messages will be committed before returning |
| * |
| * @param session the session to use for sending |
| * @param destination where to send them to |
| * @param count no. of messages to send |
| * |
| * @return the sent messages |
| * |
| * @throws Exception |
| */ |
| public List<Message> sendMessage(Session session, Destination destination, |
| int count) throws Exception |
| { |
| return sendMessage(session, destination, count, 0, 0); |
| } |
| |
| /** |
| * Send messages to the given destination. |
| * |
| * If session is transacted then messages will be committed before returning |
| * |
| * @param session the session to use for sending |
| * @param destination where to send them to |
| * @param count no. of messages to send |
| * |
| * @param batchSize the batchSize in which to commit, 0 means no batching, |
| * but a single commit at the end |
| * @return the sent message |
| * |
| * @throws Exception |
| */ |
| public List<Message> sendMessage(Session session, Destination destination, |
| int count, int batchSize) throws Exception |
| { |
| return sendMessage(session, destination, count, 0, batchSize); |
| } |
| |
| /** |
| * Send messages to the given destination. |
| * |
| * If session is transacted then messages will be committed before returning |
| * |
| * @param session the session to use for sending |
| * @param destination where to send them to |
| * @param count no. of messages to send |
| * |
| * @param offset offset allows the INDEX value of the message to be adjusted. |
| * @param batchSize the batchSize in which to commit, 0 means no batching, |
| * but a single commit at the end |
| * @return the sent message |
| * |
| * @throws Exception |
| */ |
| public List<Message> sendMessage(Session session, Destination destination, |
| int count, int offset, int batchSize) throws Exception |
| { |
| List<Message> messages = new ArrayList<Message>(count); |
| |
| MessageProducer producer = session.createProducer(destination); |
| |
| int i = offset; |
| for (; i < (count + offset); i++) |
| { |
| Message next = createNextMessage(session, i); |
| |
| producer.send(next); |
| |
| if (session.getTransacted() && batchSize > 0) |
| { |
| if (i % batchSize == 0) |
| { |
| session.commit(); |
| } |
| |
| } |
| |
| messages.add(next); |
| } |
| |
| // Ensure we commit the last messages |
| // Commit the session if we are transacted and |
| // we have no batchSize or |
| // our count is not divible by batchSize. |
| if (session.getTransacted() && |
| ( batchSize == 0 || (i-1) % batchSize != 0)) |
| { |
| session.commit(); |
| } |
| |
| return messages; |
| } |
| |
| public Message createNextMessage(Session session, int msgCount) throws JMSException |
| { |
| Message message = createMessage(session, _messageSize); |
| message.setIntProperty(INDEX, msgCount); |
| |
| return message; |
| |
| } |
| |
| public Message createMessage(Session session, int messageSize) throws JMSException |
| { |
| String payload = new String(new byte[messageSize]); |
| |
| Message message; |
| |
| switch (_messageType) |
| { |
| case BYTES: |
| message = session.createBytesMessage(); |
| ((BytesMessage) message).writeUTF(payload); |
| break; |
| case MAP: |
| message = session.createMapMessage(); |
| ((MapMessage) message).setString(CONTENT, payload); |
| break; |
| default: // To keep the compiler happy |
| case TEXT: |
| message = session.createTextMessage(); |
| ((TextMessage) message).setText(payload); |
| break; |
| case OBJECT: |
| message = session.createObjectMessage(); |
| ((ObjectMessage) message).setObject(payload); |
| break; |
| case STREAM: |
| message = session.createStreamMessage(); |
| ((StreamMessage) message).writeString(payload); |
| break; |
| } |
| |
| return message; |
| } |
| |
| protected int getMessageSize() |
| { |
| return _messageSize; |
| } |
| |
| protected void setMessageSize(int byteSize) |
| { |
| _messageSize = byteSize; |
| } |
| |
| public BrokerDetails getBroker() |
| { |
| try |
| { |
| if (getConnectionFactory().getConnectionURL().getBrokerCount() > 0) |
| { |
| return getConnectionFactory().getConnectionURL().getBrokerDetails(0); |
| } |
| else |
| { |
| fail("No broker details are available."); |
| } |
| } |
| catch (NamingException e) |
| { |
| fail(e.getMessage()); |
| } |
| |
| //keep compiler happy |
| return null; |
| } |
| |
| protected int getFailingPort() |
| { |
| return FAILING_PORT; |
| } |
| |
| public int getHttpManagementPort(int mainPort) |
| { |
| return mainPort + (DEFAULT_HTTP_MANAGEMENT_PORT - DEFAULT_PORT); |
| } |
| |
| public void assertProducingConsuming(final Connection connection) throws Exception |
| { |
| Session session = connection.createSession(true, Session.SESSION_TRANSACTED); |
| Destination destination = session.createQueue(getTestQueueName()); |
| MessageConsumer consumer = session.createConsumer(destination); |
| sendMessage(session, destination, 1); |
| session.commit(); |
| connection.start(); |
| Message m1 = consumer.receive(RECEIVE_TIMEOUT); |
| assertNotNull("Message 1 is not received", m1); |
| assertEquals("Unexpected first message received", 0, m1.getIntProperty(INDEX)); |
| session.commit(); |
| session.close(); |
| } |
| } |
