| /* |
| * 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.systest.core.brokerj; |
| |
| import static java.nio.charset.StandardCharsets.UTF_8; |
| |
| import java.io.BufferedReader; |
| import java.io.ByteArrayOutputStream; |
| import java.io.File; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.InputStreamReader; |
| import java.io.OutputStream; |
| import java.lang.reflect.Method; |
| import java.net.InetSocketAddress; |
| import java.nio.file.Files; |
| import java.text.SimpleDateFormat; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.Hashtable; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| |
| import javax.jms.Connection; |
| import javax.jms.ConnectionFactory; |
| import javax.jms.JMSException; |
| import javax.jms.Session; |
| import javax.naming.Context; |
| import javax.naming.InitialContext; |
| import javax.naming.NamingException; |
| |
| import ch.qos.logback.classic.LoggerContext; |
| import com.fasterxml.jackson.core.JsonProcessingException; |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import com.google.common.io.ByteStreams; |
| import com.google.common.util.concurrent.Futures; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.systest.core.BrokerAdmin; |
| import org.apache.qpid.systest.core.BrokerAdminException; |
| import org.apache.qpid.systest.core.dependency.ClasspathQuery; |
| import org.apache.qpid.systest.core.logback.LogbackPropertyValueDiscriminator; |
| import org.apache.qpid.systest.core.util.FileUtils; |
| import org.apache.qpid.systest.core.logback.LogbackSocketPortNumberDefiner; |
| import org.apache.qpid.systest.core.util.SystemUtils; |
| |
| public class SpawnQpidBrokerAdmin implements BrokerAdmin |
| { |
| private static final Logger LOGGER = LoggerFactory.getLogger(SpawnQpidBrokerAdmin.class); |
| private static final String BROKER_LOG_PREFIX = "BROKER"; |
| private static final String SYSTEST_PROPERTY_PREFIX = "qpid.systest."; |
| private static final String SYSTEST_PROPERTY_BROKER_READY = "qpid.systest.broker.ready"; |
| private static final String SYSTEST_PROPERTY_BROKER_STOPPED = "qpid.systest.broker.stopped"; |
| private static final String SYSTEST_PROPERTY_BROKER_LISTENING = "qpid.systest.broker.listening"; |
| private static final String SYSTEST_PROPERTY_BROKER_PROCESS = "qpid.systest.broker.process"; |
| private static final String SYSTEST_PROPERTY_SPAWN_BROKER_STARTUP_TIME = "qpid.systest.broker_startup_time"; |
| private static final String SYSTEST_PROPERTY_BROKER_CLEAN_BETWEEN_TESTS = "qpid.systest.broker.clean.between.tests"; |
| |
| static final String SYSTEST_PROPERTY_VIRTUALHOSTNODE_TYPE = "qpid.systest.virtualhostnode.type"; |
| static final String SYSTEST_PROPERTY_VIRTUALHOST_BLUEPRINT = "qpid.systest.virtualhost.blueprint"; |
| static final String SYSTEST_PROPERTY_INITIAL_CONFIGURATION_LOCATION = "qpid.systest.initialConfigurationLocation"; |
| static final String SYSTEST_PROPERTY_BUILD_CLASSPATH_FILE = "qpid.systest.build.classpath.file"; |
| static final String SYSTEST_PROPERTY_JAVA8_EXECUTABLE = "qpid.systest.java8.executable"; |
| static final String SYSTEST_PROPERTY_BROKERJ_DEPENDECIES = "qpid.systest.brokerj.dependencies"; |
| |
| private final static AtomicLong BROKER_INSTANCE_COUNTER = new AtomicLong(); |
| |
| private volatile List<ListeningPort> _ports; |
| private volatile Process _process; |
| private volatile Integer _pid; |
| private volatile String _currentWorkDirectory; |
| private volatile boolean _isPersistentStore; |
| private volatile String _virtualHostNodeName; |
| |
| @Override |
| public void create(final Class testClass) |
| { |
| setClassQualifiedTestName(testClass.getName()); |
| LOGGER.info("========================= starting broker for test class : {}", testClass.getSimpleName()); |
| startBroker(testClass); |
| } |
| |
| @Override |
| public void start(final Class testClass, final Method method) |
| { |
| LOGGER.info("========================= prepare test environment for test : {}#{}", |
| testClass.getSimpleName(), |
| method.getName()); |
| String virtualHostNodeName = getVirtualHostNodeName(testClass, method); |
| createVirtualHost(virtualHostNodeName); |
| _virtualHostNodeName = virtualHostNodeName; |
| LOGGER.info("========================= executing test : {}#{}", testClass.getSimpleName(), method.getName()); |
| setClassQualifiedTestName(testClass.getName() + "." + method.getName()); |
| LOGGER.info("========================= start executing test : {}#{}", |
| testClass.getSimpleName(), |
| method.getName()); |
| } |
| |
| |
| @Override |
| public void stop(final Class testClass, final Method method) |
| { |
| LOGGER.info("========================= stop executing test : {}#{}", |
| testClass.getSimpleName(), |
| method.getName()); |
| setClassQualifiedTestName(testClass.getName()); |
| LOGGER.info("========================= cleaning up test environment for test : {}#{}", |
| testClass.getSimpleName(), |
| method.getName()); |
| deleteVirtualHost(getVirtualHostNodeName(testClass, method)); |
| _virtualHostNodeName = null; |
| LOGGER.info("========================= cleaning done for test : {}#{}", |
| testClass.getSimpleName(), |
| method.getName()); |
| } |
| |
| @Override |
| public void destroy(final Class testClass) |
| { |
| LOGGER.info("========================= stopping broker for test class: {}", testClass.getSimpleName()); |
| shutdown(); |
| _ports.clear(); |
| if (Boolean.getBoolean(SYSTEST_PROPERTY_BROKER_CLEAN_BETWEEN_TESTS)) |
| { |
| FileUtils.delete(new File(_currentWorkDirectory), true); |
| } |
| _isPersistentStore = false; |
| LOGGER.info("========================= stopping broker done for test class : {}", testClass.getSimpleName()); |
| setClassQualifiedTestName(null); |
| } |
| |
| @Override |
| public InetSocketAddress getBrokerAddress(final PortType portType) |
| { |
| Integer port = null; |
| switch (portType) |
| { |
| case AMQP: |
| for (ListeningPort p : _ports) |
| { |
| if (p.getProtocol() == null |
| && (p.getTransport().contains("TCP") /*|| p.getTransport().contains("SSL") */)) |
| { |
| port = p.getPort(); |
| break; |
| } |
| } |
| break; |
| default: |
| throw new IllegalArgumentException(String.format("Unknown port type '%s'", portType)); |
| } |
| if (port == null) |
| { |
| throw new IllegalArgumentException(String.format("Cannot find port of type '%s'", portType)); |
| } |
| return new InetSocketAddress(port); |
| } |
| |
| @Override |
| public boolean supportsPersistence() |
| { |
| return _isPersistentStore; |
| } |
| |
| @Override |
| public ListenableFuture<Void> restart() |
| { |
| if (_virtualHostNodeName == null) |
| { |
| throw new BrokerAdminException("Virtual host is not started"); |
| } |
| return restartVirtualHost(_virtualHostNodeName); |
| } |
| |
| @Override |
| public String getValidUsername() |
| { |
| return "guest"; |
| } |
| |
| @Override |
| public String getValidPassword() |
| { |
| return "guest"; |
| } |
| |
| @Override |
| public String getType() |
| { |
| return SpawnQpidBrokerAdmin.class.getSimpleName(); |
| } |
| |
| @Override |
| public BrokerType getBrokerType() |
| { |
| return BrokerType.BROKERJ; |
| } |
| |
| @Override |
| public Connection getConnection() throws JMSException |
| { |
| return createConnection(_virtualHostNodeName); |
| } |
| |
| @Override |
| public Connection getConnection(final Map<String, String> options) throws JMSException |
| { |
| return createConnection(_virtualHostNodeName, options); |
| } |
| |
| private void startBroker(final Class testClass) |
| { |
| try |
| { |
| start(testClass); |
| } |
| catch (Exception e) |
| { |
| if (e instanceof RuntimeException) |
| { |
| throw (RuntimeException) e; |
| } |
| else |
| { |
| throw new BrokerAdminException("Unexpected exception on broker startup", e); |
| } |
| } |
| } |
| |
| void start(final Class testClass) throws Exception |
| { |
| String timestamp = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis())); |
| _currentWorkDirectory = |
| Files.createTempDirectory(String.format("qpid-work-%s-%s-", timestamp, testClass.getSimpleName())) |
| .toString(); |
| |
| String initialConfiguration = System.getProperty(SYSTEST_PROPERTY_INITIAL_CONFIGURATION_LOCATION); |
| if (initialConfiguration == null) |
| { |
| throw new BrokerAdminException( |
| String.format("No initial configuration is found: JVM property '%s' is not set.", |
| SYSTEST_PROPERTY_INITIAL_CONFIGURATION_LOCATION)); |
| } |
| |
| File testInitialConfiguration = new File(_currentWorkDirectory, "initial-configuration.json"); |
| if (!testInitialConfiguration.createNewFile()) |
| { |
| throw new BrokerAdminException("Failed to create a file for a copy of initial configuration"); |
| } |
| if (initialConfiguration.startsWith("classpath:")) |
| { |
| String config = initialConfiguration.substring("classpath:".length()); |
| try (InputStream is = getClass().getClassLoader().getResourceAsStream(config); |
| OutputStream os = new FileOutputStream(testInitialConfiguration)) |
| { |
| ByteStreams.copy(is, os); |
| } |
| } |
| else |
| { |
| Files.copy(new File(initialConfiguration).toPath(), testInitialConfiguration.toPath()); |
| } |
| |
| String classpath; |
| File file = new File(System.getProperty(SYSTEST_PROPERTY_BUILD_CLASSPATH_FILE)); |
| if (!file.exists()) |
| { |
| String dependencies = System.getProperty(SYSTEST_PROPERTY_BROKERJ_DEPENDECIES); |
| final ClasspathQuery classpathQuery = new ClasspathQuery(SpawnQpidBrokerAdmin.class, |
| Arrays.asList(dependencies.split(","))); |
| classpath = classpathQuery.getClasspath(); |
| Files.write(file.toPath(), Collections.singleton(classpath), UTF_8); |
| } |
| else |
| { |
| classpath = new String(Files.readAllBytes(file.toPath()), UTF_8); |
| } |
| |
| // grab Qpid related JVM settings |
| List<String> jvmArguments = new ArrayList<>(); |
| Properties jvmProperties = System.getProperties(); |
| for (String jvmProperty : jvmProperties.stringPropertyNames()) |
| { |
| if (jvmProperty.startsWith(SYSTEST_PROPERTY_PREFIX) |
| || jvmProperty.equalsIgnoreCase("java.io.tmpdir")) |
| { |
| jvmArguments.add("-D" + jvmProperty + "=" + jvmProperties.getProperty(jvmProperty)); |
| } |
| } |
| |
| jvmArguments.add(0, System.getProperty(SYSTEST_PROPERTY_JAVA8_EXECUTABLE, "/usr/bin/java")); |
| jvmArguments.add(1, "-cp"); |
| jvmArguments.add(2, classpath); |
| jvmArguments.add("-Dqpid.systest.logback.socket.port=" |
| + LogbackSocketPortNumberDefiner.getLogbackSocketPortNumber()); |
| jvmArguments.add("-Dqpid.systest.logback.logs_dir=" + System.getProperty("qpid.systest.logback.logs_dir", |
| "${qpid.work_dir}")); |
| jvmArguments.add(String.format("-Dqpid.systest.logback.origin=%s-%d", |
| BROKER_LOG_PREFIX, |
| BROKER_INSTANCE_COUNTER.getAndIncrement())); |
| jvmArguments.add("-Dqpid.systest.logback.context=" + testClass.getName()); |
| if (System.getProperty("qpid.systest.remote_debugger") != null) |
| { |
| jvmArguments.add(System.getProperty("qpid.systest.remote_debugger")); |
| } |
| jvmArguments.add("org.apache.qpid.server.Main"); |
| jvmArguments.add("-prop"); |
| jvmArguments.add(String.format("qpid.work_dir=%s", escapePath(_currentWorkDirectory))); |
| jvmArguments.add("--store-type"); |
| jvmArguments.add("JSON"); |
| jvmArguments.add("--initial-config-path"); |
| jvmArguments.add(escapePath(testInitialConfiguration.toString())); |
| |
| LOGGER.debug("Spawning broker JVM :", jvmArguments); |
| |
| String ready = System.getProperty(SYSTEST_PROPERTY_BROKER_READY, "BRK-1004 : Qpid Broker Ready"); |
| String stopped = System.getProperty(SYSTEST_PROPERTY_BROKER_STOPPED, "BRK-1005 : Stopped"); |
| String amqpListening = System.getProperty(SYSTEST_PROPERTY_BROKER_LISTENING, |
| "BRK-1002 : Starting( : \\w*)? : Listening on (\\w*) port ([0-9]+)"); |
| String process = System.getProperty(SYSTEST_PROPERTY_BROKER_PROCESS, "BRK-1017 : Process : PID : ([0-9]+)"); |
| int startUpTime = Integer.getInteger(SYSTEST_PROPERTY_SPAWN_BROKER_STARTUP_TIME, 30000); |
| |
| LOGGER.debug("Spawning broker permitted start-up time: {}", startUpTime); |
| |
| String[] cmd = jvmArguments.toArray(new String[jvmArguments.size()]); |
| |
| ProcessBuilder processBuilder = new ProcessBuilder(cmd); |
| processBuilder.redirectErrorStream(true); |
| |
| Map<String, String> processEnvironment = processBuilder.environment(); |
| processEnvironment.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + testClass.getName() + "\""); |
| |
| long startTime = System.currentTimeMillis(); |
| _process = processBuilder.start(); |
| |
| BrokerSystemOutpuHandler brokerSystemOutpuHandler = new BrokerSystemOutpuHandler(_process.getInputStream(), |
| ready, |
| stopped, |
| process, |
| amqpListening, |
| getClass().getName()); |
| |
| boolean brokerStarted = false; |
| ExecutorService executorService = Executors.newFixedThreadPool(1); |
| try |
| { |
| Future<?> result = executorService.submit(brokerSystemOutpuHandler); |
| result.get(startUpTime, TimeUnit.MILLISECONDS); |
| |
| _pid = brokerSystemOutpuHandler.getPID(); |
| _ports = brokerSystemOutpuHandler.getAmqpPorts(); |
| |
| if (_pid == -1) |
| { |
| throw new BrokerAdminException("Broker PID is not detected"); |
| } |
| |
| if (_ports.size() == 0) |
| { |
| throw new BrokerAdminException("Broker port is not detected"); |
| } |
| |
| try |
| { |
| //test that the broker is still running and hasn't exited unexpectedly |
| int exit = _process.exitValue(); |
| LOGGER.info("broker aborted: {}", exit); |
| throw new BrokerAdminException("broker aborted: " + exit); |
| } |
| catch (IllegalThreadStateException e) |
| { |
| // this is expect if the broker started successfully |
| } |
| |
| LOGGER.info("Broker was started successfully within {} milliseconds, broker PID {}", |
| System.currentTimeMillis() - startTime, |
| _pid); |
| LOGGER.info("Broker ports: {}", _ports); |
| brokerStarted = true; |
| } |
| catch (RuntimeException e) |
| { |
| throw e; |
| } |
| catch (TimeoutException e) |
| { |
| LOGGER.warn("Spawned broker failed to become ready within {} ms. Ready line '{}'", |
| startUpTime, brokerSystemOutpuHandler.getReady()); |
| String threadDump = dumpThreads(); |
| if (!threadDump.isEmpty()) |
| { |
| LOGGER.warn("the result of a try to capture thread dump:" + threadDump); |
| } |
| throw new BrokerAdminException(String.format("Broker failed to become ready within %d ms. Stop line : %s", |
| startUpTime, |
| brokerSystemOutpuHandler.getStopLine())); |
| } |
| catch (ExecutionException e) |
| { |
| throw new BrokerAdminException(String.format("Broker startup failed due to %s", e.getCause()), |
| e.getCause()); |
| } |
| catch (Exception e) |
| { |
| throw new BrokerAdminException(String.format("Unexpected exception on broker startup: %s", e), e); |
| } |
| finally |
| { |
| if (!brokerStarted) |
| { |
| LOGGER.warn("Broker failed to start"); |
| _process.destroy(); |
| } |
| executorService.shutdown(); |
| } |
| } |
| |
| void createVirtualHost(final String virtualHostNodeName) |
| { |
| final String nodeType = System.getProperty(SYSTEST_PROPERTY_VIRTUALHOSTNODE_TYPE); |
| _isPersistentStore = !"Memory".equals(nodeType); |
| |
| String storeDir = null; |
| if (System.getProperty("profile", "").startsWith("java-dby-mem")) |
| { |
| storeDir = ":memory:"; |
| } |
| else if (!"Memory".equals(nodeType)) |
| { |
| storeDir = "${qpid.work_dir}" + File.separator + virtualHostNodeName; |
| } |
| |
| String blueprint = System.getProperty(SYSTEST_PROPERTY_VIRTUALHOST_BLUEPRINT); |
| LOGGER.debug("Creating Virtual host from blueprint: {}", blueprint); |
| |
| Map<String, Object> attributes = new HashMap<>(); |
| attributes.put("name", virtualHostNodeName); |
| attributes.put("type", nodeType); |
| attributes.put("qpid-type", nodeType); |
| String contextAsString; |
| try |
| { |
| contextAsString = |
| new ObjectMapper().writeValueAsString(Collections.singletonMap("virtualhostBlueprint", blueprint)); |
| } |
| catch (JsonProcessingException e) |
| { |
| throw new BrokerAdminException("Cannot create virtual host as context serialization failed", e); |
| } |
| attributes.put("context", contextAsString); |
| attributes.put("defaultVirtualHostNode", true); |
| attributes.put("virtualHostInitialConfiguration", blueprint); |
| if (storeDir != null) |
| { |
| attributes.put("storePath", storeDir); |
| } |
| |
| try |
| { |
| Connection connection = createConnection("$management"); |
| try |
| { |
| connection.start(); |
| final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| try |
| { |
| new AmqpManagementFacade().createEntityUsingAmqpManagement(virtualHostNodeName, |
| "org.apache.qpid.VirtualHostNode", |
| attributes, |
| session); |
| } |
| catch (AmqpManagementFacade.OperationUnsuccessfulException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot create test virtual host '%s'", |
| virtualHostNodeName), e); |
| } |
| finally |
| { |
| session.close(); |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| catch (JMSException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot create virtual host '%s'", virtualHostNodeName), e); |
| } |
| } |
| |
| void deleteVirtualHost(final String virtualHostNodeName) |
| { |
| try |
| { |
| Connection connection = createConnection("$management"); |
| try |
| { |
| connection.start(); |
| final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| try |
| { |
| new AmqpManagementFacade().deleteEntityUsingAmqpManagement(virtualHostNodeName, |
| "org.apache.qpid.VirtualHostNode", |
| session); |
| } |
| catch (AmqpManagementFacade.OperationUnsuccessfulException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot delete test virtual host '%s'", |
| virtualHostNodeName), e); |
| } |
| finally |
| { |
| session.close(); |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| catch (JMSException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot delete virtual host '%s'", virtualHostNodeName), e); |
| } |
| } |
| |
| ListenableFuture<Void> restartVirtualHost(final String virtualHostNodeName) |
| { |
| try |
| { |
| Connection connection = createConnection("$management"); |
| try |
| { |
| connection.start(); |
| updateVirtualHostNode(virtualHostNodeName, |
| Collections.<String, Object>singletonMap("desiredState", "STOPPED"), connection); |
| updateVirtualHostNode(virtualHostNodeName, |
| Collections.<String, Object>singletonMap("desiredState", "ACTIVE"), connection); |
| return Futures.immediateFuture(null); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| catch (JMSException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot create virtual host %s", virtualHostNodeName), e); |
| } |
| } |
| |
| private void updateVirtualHostNode(final String virtualHostNodeName, |
| final Map<String, Object> attributes, |
| final Connection connection) throws JMSException |
| { |
| final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| try |
| { |
| new AmqpManagementFacade().updateEntityUsingAmqpManagement(virtualHostNodeName, |
| "org.apache.qpid.VirtualHostNode", |
| attributes, |
| session); |
| } |
| catch (AmqpManagementFacade.OperationUnsuccessfulException e) |
| { |
| throw new BrokerAdminException(String.format("Cannot create test virtual host '%s'", virtualHostNodeName), |
| e); |
| } |
| finally |
| { |
| session.close(); |
| } |
| } |
| |
| void shutdown() |
| { |
| if (SystemUtils.isWindows()) |
| { |
| doWindowsKill(); |
| } |
| |
| if (_process != null) |
| { |
| LOGGER.info("Destroying broker process"); |
| _process.destroy(); |
| |
| reapChildProcess(); |
| } |
| } |
| |
| private String escapePath(String value) |
| { |
| if (SystemUtils.isWindows() && value.contains("\"") && !value.startsWith("\"")) |
| { |
| return "\"" + value.replaceAll("\"", "\"\"") + "\""; |
| } |
| else |
| { |
| return value; |
| } |
| } |
| |
| private Connection createConnection(String virtualHostName) throws JMSException |
| { |
| return createConnection(virtualHostName, null); |
| } |
| |
| private Connection createConnection(String virtualHostName, |
| final Map<String, String> options) throws JMSException |
| { |
| final Hashtable<Object, Object> initialContextEnvironment = new Hashtable<>(); |
| initialContextEnvironment.put(Context.INITIAL_CONTEXT_FACTORY, |
| "org.apache.qpid.jndi.PropertiesFileInitialContextFactory"); |
| final String factoryName = "connectionFactory"; |
| String urlTemplate = "amqp://:@%s/%s?brokerlist='tcp://localhost:%d?failover='nofailover''"; |
| StringBuilder url = new StringBuilder(String.format(urlTemplate, |
| "spawn_broker_admin", |
| virtualHostName, |
| getBrokerAddress(PortType.AMQP).getPort())); |
| if (options != null) |
| { |
| for (Map.Entry<String, String> option : options.entrySet()) |
| { |
| url.append("&").append(option.getKey()).append("='").append(option.getValue()).append("'"); |
| } |
| } |
| initialContextEnvironment.put("connectionfactory." + factoryName, url.toString()); |
| try |
| { |
| InitialContext initialContext = new InitialContext(initialContextEnvironment); |
| try |
| { |
| ConnectionFactory factory = (ConnectionFactory) initialContext.lookup(factoryName); |
| return factory.createConnection(getValidUsername(), getValidPassword()); |
| } |
| finally |
| { |
| initialContext.close(); |
| } |
| } |
| catch (NamingException e) |
| { |
| throw new BrokerAdminException("Unexpected exception on connection lookup", e); |
| } |
| } |
| |
| private void setClassQualifiedTestName(final String name) |
| { |
| final LoggerContext loggerContext = ((ch.qos.logback.classic.Logger) LOGGER).getLoggerContext(); |
| loggerContext.putProperty(LogbackPropertyValueDiscriminator.CLASS_QUALIFIED_TEST_NAME, name); |
| } |
| |
| |
| private String getVirtualHostNodeName(final Class testClass, final Method method) |
| { |
| return testClass.getSimpleName() + "_" + method.getName(); |
| } |
| |
| |
| private void doWindowsKill() |
| { |
| try |
| { |
| |
| Process p; |
| p = Runtime.getRuntime().exec(new String[]{"taskkill", "/PID", Integer.toString(_pid), "/T", "/F"}); |
| consumeAllOutput(p); |
| } |
| catch (IOException e) |
| { |
| LOGGER.error("Error whilst killing process " + _pid, e); |
| } |
| } |
| |
| private static void consumeAllOutput(Process p) throws IOException |
| { |
| try (InputStreamReader inputStreamReader = new InputStreamReader(p.getInputStream())) |
| { |
| try (BufferedReader reader = new BufferedReader(inputStreamReader)) |
| { |
| String line; |
| while ((line = reader.readLine()) != null) |
| { |
| LOGGER.debug("Consuming output: {}", line); |
| } |
| } |
| } |
| } |
| |
| private void reapChildProcess() |
| { |
| try |
| { |
| _process.waitFor(); |
| LOGGER.info("broker exited: " + _process.exitValue()); |
| } |
| catch (InterruptedException e) |
| { |
| LOGGER.error("Interrupted whilst waiting for process shutdown"); |
| Thread.currentThread().interrupt(); |
| } |
| finally |
| { |
| try |
| { |
| _process.getInputStream().close(); |
| _process.getErrorStream().close(); |
| _process.getOutputStream().close(); |
| } |
| catch (IOException ignored) |
| { |
| } |
| } |
| } |
| |
| private String dumpThreads() |
| { |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| try |
| { |
| Process process = Runtime.getRuntime().exec("jstack " + _pid); |
| InputStream is = process.getInputStream(); |
| byte[] buffer = new byte[1024]; |
| int length; |
| while ((length = is.read(buffer)) != -1) |
| { |
| baos.write(buffer, 0, length); |
| } |
| } |
| catch (Exception e) |
| { |
| LOGGER.error("Error whilst collecting thread dump for " + _pid, e); |
| } |
| return new String(baos.toByteArray()); |
| } |
| |
| |
| public final class BrokerSystemOutpuHandler implements Runnable |
| { |
| |
| private final BufferedReader _in; |
| private final Logger _out; |
| private final String _ready; |
| private final String _stopped; |
| private final List<ListeningPort> _amqpPorts; |
| private final Pattern _pidPattern; |
| private final Pattern _amqpPortPattern; |
| private volatile boolean _seenReady; |
| private volatile String _stopLine; |
| private volatile int _pid; |
| |
| private BrokerSystemOutpuHandler(InputStream in, |
| String ready, |
| String stopped, |
| String pidRegExp, |
| String amqpPortRegExp, |
| String loggerName) |
| { |
| _amqpPorts = new ArrayList<>(); |
| _in = new BufferedReader(new InputStreamReader(in)); |
| _out = LoggerFactory.getLogger(loggerName); |
| _ready = ready; |
| _stopped = stopped; |
| _seenReady = false; |
| _amqpPortPattern = Pattern.compile(amqpPortRegExp); |
| _pidPattern = Pattern.compile(pidRegExp); |
| } |
| |
| @Override |
| public void run() |
| { |
| try |
| { |
| String line; |
| while ((line = _in.readLine()) != null) |
| { |
| _out.info(line); |
| |
| checkPortListeningLog(line, _amqpPortPattern, _amqpPorts); |
| |
| Matcher pidMatcher = _pidPattern.matcher(line); |
| if (pidMatcher.find()) |
| { |
| if (pidMatcher.groupCount() > 1) |
| { |
| _pid = Integer.parseInt(pidMatcher.group(1)); |
| } |
| } |
| |
| if (line.contains(_ready)) |
| { |
| _seenReady = true; |
| break; |
| } |
| |
| if (!_seenReady && line.contains(_stopped)) |
| { |
| _stopLine = line; |
| } |
| } |
| } |
| catch (IOException e) |
| { |
| LOGGER.warn(e.getMessage() |
| + " : Broker stream from unexpectedly closed; last log lines written by Broker may be lost."); |
| } |
| } |
| |
| private void checkPortListeningLog(final String line, |
| final Pattern portPattern, |
| final List<ListeningPort> ports) |
| { |
| Matcher portMatcher = portPattern.matcher(line); |
| if (portMatcher.find()) |
| { |
| ports.add(new ListeningPort(portMatcher.group(1), |
| portMatcher.group(2), |
| Integer.parseInt(portMatcher.group(3)))); |
| } |
| } |
| |
| String getStopLine() |
| { |
| return _stopLine; |
| } |
| |
| String getReady() |
| { |
| return _ready; |
| } |
| |
| int getPID() |
| { |
| return _pid; |
| } |
| |
| List<ListeningPort> getAmqpPorts() |
| { |
| return _amqpPorts; |
| } |
| } |
| |
| private static class ListeningPort |
| { |
| private String _protocol; |
| private String _transport; |
| private int _port; |
| |
| ListeningPort(final String protocol, final String transport, final int port) |
| { |
| _transport = transport; |
| _port = port; |
| _protocol = protocol; |
| } |
| |
| String getTransport() |
| { |
| return _transport; |
| } |
| |
| int getPort() |
| { |
| return _port; |
| } |
| |
| String getProtocol() |
| { |
| return _protocol; |
| } |
| |
| @Override |
| public String toString() |
| { |
| return "ListeningPort{" + |
| "_protocol='" + _protocol + '\'' + |
| ", _transport='" + _transport + '\'' + |
| ", _port=" + _port + |
| '}'; |
| } |
| } |
| } |
