From 155ed9d858367d218e43144061b9c1225dd30e2b Mon Sep 17 00:00:00 2001 From: Robert Gemmell Date: Thu, 7 Jul 2011 15:12:26 +0000 Subject: QPID-2815: refactor broker startup to present a clean interface interface for starting the broker within an existing application Applied patch by Keith Wall and myself git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1143870 13f79535-47bb-0310-9956-ffa450edef68 --- .../main/java/org/apache/qpid/server/Broker.java | 420 ++++++++++++++++ .../java/org/apache/qpid/server/BrokerOptions.java | 160 +++++++ .../src/main/java/org/apache/qpid/server/Main.java | 527 +++------------------ .../org/apache/qpid/server/ProtocolExclusion.java | 73 +++ .../server/configuration/ServerConfiguration.java | 6 +- 5 files changed, 735 insertions(+), 451 deletions(-) create mode 100644 java/broker/src/main/java/org/apache/qpid/server/Broker.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java (limited to 'java/broker/src/main') diff --git a/java/broker/src/main/java/org/apache/qpid/server/Broker.java b/java/broker/src/main/java/org/apache/qpid/server/Broker.java new file mode 100644 index 0000000000..124c2a7d08 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/Broker.java @@ -0,0 +1,420 @@ +/* + * + * 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.server; + +import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.EnumSet; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; + +import org.apache.log4j.PropertyConfigurator; +import org.apache.log4j.xml.QpidLog4JConfigurator; +import org.apache.qpid.server.configuration.ServerConfiguration; +import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration; +import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean; +import org.apache.qpid.server.information.management.ServerInformationMBean; +import org.apache.qpid.server.logging.SystemOutMessageLogger; +import org.apache.qpid.server.logging.actors.BrokerActor; +import org.apache.qpid.server.logging.actors.CurrentActor; +import org.apache.qpid.server.logging.actors.GenericActor; +import org.apache.qpid.server.logging.management.LoggingManagementMBean; +import org.apache.qpid.server.logging.messages.BrokerMessages; +import org.apache.qpid.server.protocol.AMQProtocolEngineFactory; +import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory; +import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory.VERSION; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; +import org.apache.qpid.server.transport.QpidAcceptor; +import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.transport.NetworkTransportConfiguration; +import org.apache.qpid.transport.network.IncomingNetworkTransport; +import org.apache.qpid.transport.network.Transport; +import org.apache.qpid.transport.network.mina.MinaNetworkTransport; + +public class Broker +{ + private static final int IPV4_ADDRESS_LENGTH = 4; + private static final char IPV4_LITERAL_SEPARATOR = '.'; + + protected static class InitException extends RuntimeException + { + private static final long serialVersionUID = 1L; + + InitException(String msg, Throwable cause) + { + super(msg, cause); + } + } + + public void shutdown() + { + ApplicationRegistry.remove(); + } + + public void startup() throws Exception + { + startup(new BrokerOptions()); + } + + public void startup(BrokerOptions options) throws Exception + { + try + { + CurrentActor.set(new BrokerActor(new SystemOutMessageLogger())); + startupImpl(options); + } + finally + { + CurrentActor.remove(); + } + } + + private void startupImpl(final BrokerOptions options) throws Exception + { + final String qpidHome = options.getQpidHome(); + final File configFile = getConfigFile(options.getConfigFile(), + BrokerOptions.DEFAULT_CONFIG_FILE, qpidHome, true); + + CurrentActor.get().message(BrokerMessages.CONFIG(configFile.getAbsolutePath())); + + File logConfigFile = getConfigFile(options.getLogConfigFile(), + BrokerOptions.DEFAULT_LOG_CONFIG_FILE, qpidHome, false); + + configureLogging(logConfigFile, options.getLogWatchFrequency()); + + ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configFile); + ServerConfiguration serverConfig = config.getConfiguration(); + updateManagementPort(serverConfig, options.getJmxPort()); + + ApplicationRegistry.initialise(config); + + // We have already loaded the BrokerMessages class by this point so we + // need to refresh the locale setting incase we had a different value in + // the configuration. + BrokerMessages.reload(); + + // AR.initialise() sets and removes its own actor so we now need to set the actor + // for the remainder of the startup, and the default actor if the stack is empty + CurrentActor.set(new BrokerActor(config.getCompositeStartupMessageLogger())); + CurrentActor.setDefault(new BrokerActor(config.getRootMessageLogger())); + GenericActor.setDefaultMessageLogger(config.getRootMessageLogger()); + + try + { + configureLoggingManagementMBean(logConfigFile, options.getLogWatchFrequency()); + + ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean(); + configMBean.register(); + + ServerInformationMBean sysInfoMBean = new ServerInformationMBean(config); + sysInfoMBean.register(); + + Set ports = new HashSet(options.getPorts()); + if(ports.isEmpty()) + { + parsePortList(ports, serverConfig.getPorts()); + } + + Set sslPorts = new HashSet(options.getSSLPorts()); + if(sslPorts.isEmpty()) + { + parsePortList(sslPorts, serverConfig.getSSLPorts()); + } + + Set exclude_0_10 = new HashSet(options.getExcludedPorts(ProtocolExclusion.v0_10)); + if(exclude_0_10.isEmpty()) + { + parsePortList(exclude_0_10, serverConfig.getPortExclude010()); + } + + Set exclude_0_9_1 = new HashSet(options.getExcludedPorts(ProtocolExclusion.v0_9_1)); + if(exclude_0_9_1.isEmpty()) + { + parsePortList(exclude_0_9_1, serverConfig.getPortExclude091()); + } + + Set exclude_0_9 = new HashSet(options.getExcludedPorts(ProtocolExclusion.v0_9)); + if(exclude_0_9.isEmpty()) + { + parsePortList(exclude_0_9, serverConfig.getPortExclude09()); + } + + Set exclude_0_8 = new HashSet(options.getExcludedPorts(ProtocolExclusion.v0_8)); + if(exclude_0_8.isEmpty()) + { + parsePortList(exclude_0_8, serverConfig.getPortExclude08()); + } + + String bindAddr = options.getBind(); + if (bindAddr == null) + { + bindAddr = serverConfig.getBind(); + } + + InetAddress bindAddress = null; + if (bindAddr.equals(WILDCARD_ADDRESS)) + { + bindAddress = new InetSocketAddress(0).getAddress(); + } + else + { + bindAddress = InetAddress.getByAddress(parseIP(bindAddr)); + } + String hostName = bindAddress.getCanonicalHostName(); + + if (!serverConfig.getSSLOnly()) + { + for(int port : ports) + { + Set supported = EnumSet.allOf(VERSION.class); + + if(exclude_0_10.contains(port)) + { + supported.remove(VERSION.v0_10); + } + + if(exclude_0_9_1.contains(port)) + { + supported.remove(VERSION.v0_9_1); + } + if(exclude_0_9.contains(port)) + { + supported.remove(VERSION.v0_9); + } + if(exclude_0_8.contains(port)) + { + supported.remove(VERSION.v0_8); + } + + NetworkTransportConfiguration settings = + new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP); + + IncomingNetworkTransport transport = new MinaNetworkTransport(); + MultiVersionProtocolEngineFactory protocolEngineFactory = + new MultiVersionProtocolEngineFactory(hostName, supported); + + transport.accept(settings, protocolEngineFactory, null); + ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port), + new QpidAcceptor(transport,"TCP")); + CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port)); + } + } + + if (serverConfig.getEnableSSL()) + { + String keystorePath = serverConfig.getKeystorePath(); + String keystorePassword = serverConfig.getKeystorePassword(); + String certType = serverConfig.getCertType(); + SSLContextFactory sslFactory = + new SSLContextFactory(keystorePath, keystorePassword, certType); + + for(int sslPort : sslPorts) + { + NetworkTransportConfiguration settings = + new ServerNetworkTransportConfiguration(serverConfig, sslPort, bindAddress.getHostName(), Transport.TCP); + + IncomingNetworkTransport transport = new MinaNetworkTransport(); + + transport.accept(settings, new AMQProtocolEngineFactory(), sslFactory); + + ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, sslPort), + new QpidAcceptor(transport,"TCP")); + CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", sslPort)); + } + } + + CurrentActor.get().message(BrokerMessages.READY()); + } + finally + { + // Startup is complete so remove the AR initialised Startup actor + CurrentActor.remove(); + } + } + + private File getConfigFile(final String fileName, + final String defaultFileName, + final String qpidHome, boolean throwOnFileNotFound) throws InitException + { + File configFile = null; + if (fileName != null) + { + configFile = new File(fileName); + } + else + { + configFile = new File(qpidHome, defaultFileName); + } + + if (!configFile.exists() && throwOnFileNotFound) + { + String error = "File " + fileName + " could not be found. Check the file exists and is readable."; + + if (qpidHome == null) + { + error = error + "\nNote: " + BrokerOptions.QPID_HOME + " is not set."; + } + + throw new InitException(error, null); + } + + return configFile; + } + + public static void parsePortList(Set output, List ports) throws InitException + { + if(ports != null) + { + for(Object o : ports) + { + try + { + output.add(Integer.parseInt(String.valueOf(o))); + } + catch (NumberFormatException e) + { + throw new InitException("Invalid port: " + o, e); + } + } + } + } + + /** + * Update the configuration data with the management port. + * @param configuration + * @param managementPort The string from the command line + */ + private void updateManagementPort(ServerConfiguration configuration, Integer managementPort) + { + if (managementPort != null) + { + try + { + configuration.setJMXManagementPort(managementPort); + } + catch (NumberFormatException e) + { + throw new InitException("Invalid management port: " + managementPort, null); + } + } + } + + private byte[] parseIP(String address) throws Exception + { + char[] literalBuffer = address.toCharArray(); + int byteCount = 0; + int currByte = 0; + byte[] ip = new byte[IPV4_ADDRESS_LENGTH]; + for (int i = 0; i < literalBuffer.length; i++) + { + char currChar = literalBuffer[i]; + if ((currChar >= '0') && (currChar <= '9')) + { + currByte = (currByte * 10) + (Character.digit(currChar, 10) & 0xFF); + } + + if (currChar == IPV4_LITERAL_SEPARATOR || (i + 1 == literalBuffer.length)) + { + ip[byteCount++] = (byte) currByte; + currByte = 0; + } + } + + if (byteCount != 4) + { + throw new Exception("Invalid IP address: " + address); + } + return ip; + } + + private void configureLogging(File logConfigFile, long logWatchTime) throws InitException, IOException + { + if (logConfigFile.exists() && logConfigFile.canRead()) + { + CurrentActor.get().message(BrokerMessages.LOG_CONFIG(logConfigFile.getAbsolutePath())); + + if (logWatchTime > 0) + { + System.out.println("log file " + logConfigFile.getAbsolutePath() + " will be checked for changes every " + + logWatchTime + " seconds"); + // log4j expects the watch interval in milliseconds + try + { + QpidLog4JConfigurator.configureAndWatch(logConfigFile.getPath(), logWatchTime); + } + catch (Exception e) + { + throw new InitException(e.getMessage(),e); + } + } + else + { + try + { + QpidLog4JConfigurator.configure(logConfigFile.getPath()); + } + catch (Exception e) + { + throw new InitException(e.getMessage(),e); + } + } + } + else + { + System.err.println("Logging configuration error: unable to read file " + logConfigFile.getAbsolutePath()); + System.err.println("Using the fallback internal log4j.properties configuration"); + + InputStream propsFile = this.getClass().getResourceAsStream("/log4j.properties"); + if(propsFile == null) + { + throw new IOException("Unable to load the fallback internal log4j.properties configuration file"); + } + else + { + try + { + Properties fallbackProps = new Properties(); + fallbackProps.load(propsFile); + PropertyConfigurator.configure(fallbackProps); + } + finally + { + propsFile.close(); + } + } + } + } + + private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception + { + LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime); + + blm.register(); + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java b/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java new file mode 100644 index 0000000000..bf554c526f --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/BrokerOptions.java @@ -0,0 +1,160 @@ +/* + * + * 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.server; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +public class BrokerOptions +{ + /** serialVersionUID */ + private static final long serialVersionUID = 8051825964945442234L; + + public static final String DEFAULT_CONFIG_FILE = "etc/config.xml"; + public static final String DEFAULT_LOG_CONFIG_FILE = "etc/log4j.xml"; + public static final String QPID_HOME = "QPID_HOME"; + + public static final String PORTS = "p"; + public static final String SSL_PORTS = "s"; + public static final String BIND = "b"; + public static final String MANAGEMENT = "m"; + public static final String LOG4J = "l"; + public static final String WATCH = "w"; + public static final String CONFIG = "c"; + public static final String PROTOCOL = "protocol"; + + private final Set _ports = new HashSet(); + private final Set _sslPorts = new HashSet(); + private final Map> _exclusionMap = new HashMap>(); + + private String _configFile; + private String _logConfigFile; + private String _bind; + private String _transport = Transport.TCP; + private Integer _jmxPort; + + private Integer _logWatchFrequency = 0; + + public void addPort(final int port) + { + _ports.add(port); + } + + public void addSSLPort(final int sslPort) + { + _sslPorts.add(sslPort); + } + + public Set getPorts() + { + return Collections.unmodifiableSet(_ports); + } + + public Set getSSLPorts() + { + return Collections.unmodifiableSet(_sslPorts); + } + + public String getConfigFile() + { + return _configFile; + } + + public void setConfigFile(final String configFile) + { + _configFile = configFile; + } + + public String getLogConfigFile() + { + return _logConfigFile; + } + + public void setLogConfigFile(final String logConfigFile) + { + _logConfigFile = logConfigFile; + } + + public Integer getJmxPort() + { + return _jmxPort; + } + + public void setJmxPort(final int jmxPort) + { + _jmxPort = jmxPort; + } + + public String getQpidHome() + { + return System.getProperty(QPID_HOME); + } + + public Set getExcludedPorts(final ProtocolExclusion excludeProtocol) + { + final Set excludedPorts = _exclusionMap.get(excludeProtocol); + return excludedPorts == null ? Collections.emptySet() : excludedPorts; + } + + public void addExcludedPort(final ProtocolExclusion excludeProtocol, final int port) + { + if (!_exclusionMap.containsKey(excludeProtocol)) + { + _exclusionMap.put(excludeProtocol, new HashSet()); + } + + Set ports = _exclusionMap.get(excludeProtocol); + ports.add(port); + } + + public String getBind() + { + return _bind; + } + + public void setBind(final String bind) + { + _bind = bind; + } + + public int getLogWatchFrequency() + { + return _logWatchFrequency; + } + + public void setLogWatchFrequency(final int logWatchFrequency) + { + _logWatchFrequency = logWatchFrequency; + } + + public String getTransport() + { + return _transport; + } + + public void setTransport(final String transport) + { + _transport = transport; + } +} \ No newline at end of file diff --git a/java/broker/src/main/java/org/apache/qpid/server/Main.java b/java/broker/src/main/java/org/apache/qpid/server/Main.java index 2925db69de..317459942a 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/Main.java +++ b/java/broker/src/main/java/org/apache/qpid/server/Main.java @@ -20,19 +20,6 @@ */ package org.apache.qpid.server; -import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS; - -import java.io.File; -import java.io.IOException; -import java.io.InputStream; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.util.EnumSet; -import java.util.HashSet; -import java.util.List; -import java.util.Properties; -import java.util.Set; - import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.HelpFormatter; import org.apache.commons.cli.Option; @@ -41,32 +28,9 @@ import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; import org.apache.commons.cli.PosixParser; import org.apache.log4j.Logger; -import org.apache.log4j.PropertyConfigurator; -import org.apache.log4j.xml.QpidLog4JConfigurator; -import org.apache.qpid.common.QpidProperties; -import org.apache.qpid.framing.ProtocolVersion; -import org.apache.qpid.server.configuration.ServerConfiguration; -import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration; -import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean; -import org.apache.qpid.server.information.management.ServerInformationMBean; -import org.apache.qpid.server.logging.SystemOutMessageLogger; -import org.apache.qpid.server.logging.actors.BrokerActor; -import org.apache.qpid.server.logging.actors.CurrentActor; -import org.apache.qpid.server.logging.actors.GenericActor; -import org.apache.qpid.server.logging.management.LoggingManagementMBean; -import org.apache.qpid.server.logging.messages.BrokerMessages; -import org.apache.qpid.server.protocol.AMQProtocolEngineFactory; -import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory; -import org.apache.qpid.server.protocol.MultiVersionProtocolEngineFactory.VERSION; +import org.apache.qpid.server.Broker.InitException; import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; -import org.apache.qpid.server.transport.QpidAcceptor; -import org.apache.qpid.ssl.SSLContextFactory; -import org.apache.qpid.transport.ConnectionSettings; -import org.apache.qpid.transport.NetworkTransportConfiguration; -import org.apache.qpid.transport.network.IncomingNetworkTransport; -import org.apache.qpid.transport.network.Transport; -import org.apache.qpid.transport.network.mina.MinaNetworkTransport; + /** * Main entry point for AMQPD. @@ -76,35 +40,38 @@ public class Main { private static Logger _logger; - private static final String DEFAULT_CONFIG_FILE = "etc/config.xml"; - - public static final String DEFAULT_LOG_CONFIG_FILENAME = "log4j.xml"; - public static final String QPID_HOME = "QPID_HOME"; - private static final int IPV4_ADDRESS_LENGTH = 4; - - private static final char IPV4_LITERAL_SEPARATOR = '.'; + protected final static Options options = new Options(); + protected static CommandLine commandLine; - protected static class InitException extends Exception + public static void main(String[] args) { - InitException(String msg, Throwable cause) + //if the -Dlog4j.configuration property has not been set, enable the init override + //to stop Log4J wondering off and picking up the first log4j.xml/properties file it + //finds from the classpath when we get the first Loggers + if(System.getProperty("log4j.configuration") == null) { - super(msg, cause); + System.setProperty("log4j.defaultInitOverride", "true"); } - } - protected final Options options = new Options(); - protected CommandLine commandLine; - - protected Main(String[] args) - { + //now that the override status is know, we can instantiate the Loggers + _logger = Logger.getLogger(Main.class); setOptions(options); if (parseCommandline(args)) { - execute(); + try + { + execute(); + } + catch(Exception e) + { + System.err.println("Exception during startup: " + e); + e.printStackTrace(); + shutdown(1); + } } } - protected boolean parseCommandline(String[] args) + protected static boolean parseCommandline(String[] args) { try { @@ -122,8 +89,7 @@ public class Main } } - @SuppressWarnings("static-access") - protected void setOptions(Options options) + protected static void setOptions(Options options) { Option help = new Option("h", "help", false, "print this message"); Option version = new Option("v", "version", false, "print the version information and exit"); @@ -171,7 +137,7 @@ public class Main Option logconfig = OptionBuilder.withArgName("logconfig").hasArg() .withDescription("use the specified log4j xml configuration file. By " - + "default looks for a file named " + DEFAULT_LOG_CONFIG_FILENAME + + "default looks for a file named " + BrokerOptions.DEFAULT_LOG_CONFIG_FILE + " in the same directory as the configuration file").withLongOpt("logconfig").create("l"); Option logwatchconfig = OptionBuilder.withArgName("logwatch").hasArg() @@ -198,445 +164,110 @@ public class Main options.addOption(sslport); } - protected void execute() + protected static void execute() throws Exception { - // note this understands either --help or -h. If an option only has a long name you can use that but if - // an option has a short name and a long name you must use the short name here. - if (commandLine.hasOption("h")) + BrokerOptions options = new BrokerOptions(); + String configFile = commandLine.getOptionValue(BrokerOptions.CONFIG); + if(configFile != null) { - HelpFormatter formatter = new HelpFormatter(); - formatter.printHelp("Qpid", options, true); + options.setConfigFile(configFile); } - else if (commandLine.hasOption("v")) - { - String ver = QpidProperties.getVersionString(); - StringBuilder protocol = new StringBuilder("AMQP version(s) [major.minor]: "); - - boolean first = true; - for (ProtocolVersion pv : ProtocolVersion.getSupportedProtocolVersions()) - { - if (first) - { - first = false; - } - else - { - protocol.append(", "); - } - - protocol.append(pv.getMajorVersion()).append('-').append(pv.getMinorVersion()); - - } - - System.out.println(ver + " (" + protocol + ")"); - } - else + String logWatchConfig = commandLine.getOptionValue(BrokerOptions.WATCH); + if(logWatchConfig != null) { - try - { - CurrentActor.set(new BrokerActor(new SystemOutMessageLogger())); - startup(); - CurrentActor.remove(); - } - catch (InitException e) - { - System.out.println("Initialisation Error : " + e.getMessage()); - shutdown(1); - } - catch (Throwable e) - { - System.out.println("Error initialising message broker: " + e); - e.printStackTrace(); - shutdown(1); - } + options.setLogWatchFrequency(Integer.parseInt(logWatchConfig) * 1000); } - } - protected void shutdown(int status) - { - ApplicationRegistry.remove(); - System.exit(status); - } - - protected void startup() throws Exception - { - final String QpidHome = System.getProperty(QPID_HOME); - final File defaultConfigFile = new File(QpidHome, DEFAULT_CONFIG_FILE); - final File configFile = new File(commandLine.getOptionValue("c", defaultConfigFile.getPath())); - if (!configFile.exists()) + String logConfig = commandLine.getOptionValue(BrokerOptions.LOG4J); + if(logConfig != null) { - String error = "File " + configFile + " could not be found. Check the file exists and is readable."; - - if (QpidHome == null) - { - error = error + "\nNote: " + QPID_HOME + " is not set."; - } - - throw new InitException(error, null); - } - else - { - CurrentActor.get().message(BrokerMessages.CONFIG(configFile.getAbsolutePath())); + options.setLogConfigFile(logConfig); } - String logConfig = commandLine.getOptionValue("l"); - String logWatchConfig = commandLine.getOptionValue("w", "0"); - - int logWatchTime = 0; - try - { - logWatchTime = Integer.parseInt(logWatchConfig); - } - catch (NumberFormatException e) + String jmxPort = commandLine.getOptionValue(BrokerOptions.MANAGEMENT); + if(jmxPort != null) { - System.err.println("Log watch configuration value of " + logWatchConfig + " is invalid. Must be " - + "a non-negative integer. Using default of zero (no watching configured"); + options.setJmxPort(Integer.parseInt(jmxPort)); } - File logConfigFile; - if (logConfig != null) - { - logConfigFile = new File(logConfig); - configureLogging(logConfigFile, logWatchTime); - } - else + String bindAddr = commandLine.getOptionValue(BrokerOptions.BIND); + if (bindAddr != null) { - File configFileDirectory = configFile.getParentFile(); - logConfigFile = new File(configFileDirectory, DEFAULT_LOG_CONFIG_FILENAME); - configureLogging(logConfigFile, logWatchTime); + options.setBind(bindAddr); } - ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configFile); - ServerConfiguration serverConfig = config.getConfiguration(); - updateManagementPort(serverConfig, commandLine.getOptionValue("m")); - - ApplicationRegistry.initialise(config); - - // We have already loaded the BrokerMessages class by this point so we - // need to refresh the locale setting incase we had a different value in - // the configuration. - BrokerMessages.reload(); - - // AR.initialise() sets and removes its own actor so we now need to set the actor - // for the remainder of the startup, and the default actor if the stack is empty - CurrentActor.set(new BrokerActor(config.getCompositeStartupMessageLogger())); - CurrentActor.setDefault(new BrokerActor(config.getRootMessageLogger())); - GenericActor.setDefaultMessageLogger(config.getRootMessageLogger()); - - - try + String[] portStr = commandLine.getOptionValues(BrokerOptions.PORTS); + if(portStr != null) { - configureLoggingManagementMBean(logConfigFile, logWatchTime); - - ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean(); - configMBean.register(); - - ServerInformationMBean sysInfoMBean = new ServerInformationMBean(config); - sysInfoMBean.register(); - - - String[] portStr = commandLine.getOptionValues("p"); - - Set ports = new HashSet(); - Set exclude_0_10 = new HashSet(); - Set exclude_0_9_1 = new HashSet(); - Set exclude_0_9 = new HashSet(); - Set exclude_0_8 = new HashSet(); - - if(portStr == null || portStr.length == 0) - { - - parsePortList(ports, serverConfig.getPorts()); - parsePortList(exclude_0_10, serverConfig.getPortExclude010()); - parsePortList(exclude_0_9_1, serverConfig.getPortExclude091()); - parsePortList(exclude_0_9, serverConfig.getPortExclude09()); - parsePortList(exclude_0_8, serverConfig.getPortExclude08()); - - } - else + parsePortArray(options, portStr, false); + for(ProtocolExclusion pe : ProtocolExclusion.values()) { - parsePortArray(ports, portStr); - parsePortArray(exclude_0_10, commandLine.getOptionValues("exclude-0-10")); - parsePortArray(exclude_0_9_1, commandLine.getOptionValues("exclude-0-9-1")); - parsePortArray(exclude_0_9, commandLine.getOptionValues("exclude-0-9")); - parsePortArray(exclude_0_8, commandLine.getOptionValues("exclude-0-8")); - - } - - - - - String bindAddr = commandLine.getOptionValue("b"); - if (bindAddr == null) - { - bindAddr = serverConfig.getBind(); - } - InetAddress bindAddress = null; - - - - if (bindAddr.equals(WILDCARD_ADDRESS)) - { - bindAddress = new InetSocketAddress(0).getAddress(); - } - else - { - bindAddress = InetAddress.getByAddress(parseIP(bindAddr)); - } - - String hostName = bindAddress.getCanonicalHostName(); - - - String keystorePath = serverConfig.getKeystorePath(); - String keystorePassword = serverConfig.getKeystorePassword(); - String certType = serverConfig.getCertType(); - SSLContextFactory sslFactory = null; - - if (!serverConfig.getSSLOnly()) - { - - for(int port : ports) - { - Set supported = EnumSet.allOf(VERSION.class); - - if(exclude_0_10.contains(port)) - { - supported.remove(VERSION.v0_10); - } - - if(exclude_0_9_1.contains(port)) - { - supported.remove(VERSION.v0_9_1); - } - if(exclude_0_9.contains(port)) - { - supported.remove(VERSION.v0_9); - } - if(exclude_0_8.contains(port)) - { - supported.remove(VERSION.v0_8); - } - - NetworkTransportConfiguration settings = - new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP); - - IncomingNetworkTransport transport = new MinaNetworkTransport(); - MultiVersionProtocolEngineFactory protocolEngineFactory = - new MultiVersionProtocolEngineFactory(hostName, supported); - - transport.accept(settings, protocolEngineFactory, sslFactory); - ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port), - new QpidAcceptor(transport, Transport.TCP)); - CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port)); - } - + parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe); } + } - if (serverConfig.getEnableSSL()) + String[] sslPortStr = commandLine.getOptionValues(BrokerOptions.SSL_PORTS); + if(sslPortStr != null) + { + parsePortArray(options, sslPortStr, true); + for(ProtocolExclusion pe : ProtocolExclusion.values()) { - String sslPort = commandLine.getOptionValue("s"); - int port = 0; - if (null != sslPort) - { - port = Integer.parseInt(sslPort); - } - else - { - port = serverConfig.getSSLPort(); - } - - NetworkTransportConfiguration settings = - new ServerNetworkTransportConfiguration(serverConfig, port, bindAddress.getHostName(), Transport.TCP); - - sslFactory = new SSLContextFactory(keystorePath, keystorePassword, certType); - - IncomingNetworkTransport transport = new MinaNetworkTransport(); - - transport.accept(settings, new AMQProtocolEngineFactory(), sslFactory); - - ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port), - new QpidAcceptor(transport,"TCP")); - CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", port)); + parsePortArray(options, commandLine.getOptionValues(pe.getExcludeName()), pe); } - - CurrentActor.get().message(BrokerMessages.READY()); - - } - finally - { - // Startup is complete so remove the AR initialised Startup actor - CurrentActor.remove(); } - - - + + Broker broker = new Broker(); + broker.startup(options); } - private void parsePortArray(Set ports, String[] portStr) - throws InitException + protected static void shutdown(int status) { - if(portStr != null) - { - for(int i = 0; i < portStr.length; i++) - { - try - { - ports.add(Integer.parseInt(portStr[i])); - } - catch (NumberFormatException e) - { - throw new InitException("Invalid port: " + portStr[i], e); - } - } - } + ApplicationRegistry.remove(); + System.exit(status); } - private void parsePortList(Set output, List input) - throws InitException + private static void parsePortArray(BrokerOptions options, Object[] ports, boolean ssl) throws InitException { - if(input != null) + if(ports != null) { - for(Object port : input) + for(int i = 0; i < ports.length; i++) { try { - output.add(Integer.parseInt(String.valueOf(port))); + if(ssl) + { + options.addSSLPort(Integer.parseInt(String.valueOf(ports[i]))); + } + else + { + options.addPort(Integer.parseInt(String.valueOf(ports[i]))); + } } catch (NumberFormatException e) { - throw new InitException("Invalid port: " + port, e); + throw new InitException("Invalid port: " + ports[i], e); } } } } - /** - * Update the configuration data with the management port. - * @param configuration - * @param managementPort The string from the command line - */ - private void updateManagementPort(ServerConfiguration configuration, String managementPort) - { - if (managementPort != null) - { - try - { - configuration.setJMXManagementPort(Integer.parseInt(managementPort)); - } - catch (NumberFormatException e) - { - _logger.warn("Invalid management port: " + managementPort + " will use:" + configuration.getJMXManagementPort(), e); - } - } - } - - public static void main(String[] args) - { - //if the -Dlog4j.configuration property has not been set, enable the init override - //to stop Log4J wondering off and picking up the first log4j.xml/properties file it - //finds from the classpath when we get the first Loggers - if(System.getProperty("log4j.configuration") == null) - { - System.setProperty("log4j.defaultInitOverride", "true"); - } - - //now that the override status is know, we can instantiate the Loggers - _logger = Logger.getLogger(Main.class); - - new Main(args); - } - - private byte[] parseIP(String address) throws Exception - { - char[] literalBuffer = address.toCharArray(); - int byteCount = 0; - int currByte = 0; - byte[] ip = new byte[IPV4_ADDRESS_LENGTH]; - for (int i = 0; i < literalBuffer.length; i++) - { - char currChar = literalBuffer[i]; - if ((currChar >= '0') && (currChar <= '9')) - { - currByte = (currByte * 10) + (Character.digit(currChar, 10) & 0xFF); - } - - if (currChar == IPV4_LITERAL_SEPARATOR || (i + 1 == literalBuffer.length)) - { - ip[byteCount++] = (byte) currByte; - currByte = 0; - } - } - - if (byteCount != 4) - { - throw new Exception("Invalid IP address: " + address); - } - return ip; - } - - private void configureLogging(File logConfigFile, int logWatchTime) throws InitException, IOException + private static void parsePortArray(BrokerOptions options, Object[] ports, ProtocolExclusion excludedProtocol) throws InitException { - if (logConfigFile.exists() && logConfigFile.canRead()) + if(ports != null) { - CurrentActor.get().message(BrokerMessages.LOG_CONFIG(logConfigFile.getAbsolutePath())); - - if (logWatchTime > 0) - { - System.out.println("log file " + logConfigFile.getAbsolutePath() + " will be checked for changes every " - + logWatchTime + " seconds"); - // log4j expects the watch interval in milliseconds - try - { - QpidLog4JConfigurator.configureAndWatch(logConfigFile.getPath(), logWatchTime * 1000); - } - catch (Exception e) - { - throw new InitException(e.getMessage(),e); - } - } - else + for(int i = 0; i < ports.length; i++) { try { - QpidLog4JConfigurator.configure(logConfigFile.getPath()); + options.addExcludedPort(excludedProtocol, + Integer.parseInt(String.valueOf(ports[i]))); } - catch (Exception e) - { - throw new InitException(e.getMessage(),e); - } - } - } - else - { - System.err.println("Logging configuration error: unable to read file " + logConfigFile.getAbsolutePath()); - System.err.println("Using the fallback internal log4j.properties configuration"); - - InputStream propsFile = this.getClass().getResourceAsStream("/log4j.properties"); - if(propsFile == null) - { - throw new IOException("Unable to load the fallback internal log4j.properties configuration file"); - } - else - { - try - { - Properties fallbackProps = new Properties(); - fallbackProps.load(propsFile); - PropertyConfigurator.configure(fallbackProps); - } - finally + catch (NumberFormatException e) { - propsFile.close(); + throw new InitException("Invalid port for exclusion: " + ports[i], e); } } } } - - private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception - { - LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime); - - blm.register(); - } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java b/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java new file mode 100644 index 0000000000..22d97d36dd --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/ProtocolExclusion.java @@ -0,0 +1,73 @@ +/* + * + * 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.server; + +import java.util.HashMap; +import java.util.Map; + +public enum ProtocolExclusion +{ + v0_8("exclude-0-8","--exclude-0-8"), + v0_9("exclude-0-9", "--exclude-0-9"), + v0_9_1("exclude-0-9-1", "--exclude-0-9-1"), + v0_10("exclude-0-10", "--exclude-0-10"); + + private static final Map MAP = new HashMap(); + + static + { + for(ProtocolExclusion pe : ProtocolExclusion.values()) + { + MAP.put(pe.getArg(), pe); + } + } + + private String _arg; + private String _excludeName; + + private ProtocolExclusion(final String excludeName, final String arg) + { + _excludeName = excludeName; + _arg = arg; + } + + public String getArg() + { + return _arg; + } + + public String getExcludeName() + { + return _excludeName; + } + + public static ProtocolExclusion lookup(final String arg) + { + ProtocolExclusion ex = MAP.get(arg); + + if(ex == null) + { + throw new IllegalArgumentException(arg + " is not a valid protocol exclusion"); + } + + return ex; + } +} diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java index 23ab5e8222..14de7c1723 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java +++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java @@ -638,7 +638,7 @@ public class ServerConfiguration extends ConfigurationPlugin implements SignalHa public List getPorts() { - return getListValue("connector.port", Collections.singletonList(DEFAULT_PORT)); + return getListValue("connector.port", Collections.singletonList(DEFAULT_PORT)); } public List getPortExclude010() @@ -696,9 +696,9 @@ public class ServerConfiguration extends ConfigurationPlugin implements SignalHa return getBooleanValue("connector.ssl.sslOnly"); } - public int getSSLPort() + public List getSSLPorts() { - return getIntValue("connector.ssl.port", DEFAULT_SSL_PORT); + return getListValue("connector.ssl.port", Collections.singletonList(DEFAULT_SSL_PORT)); } public String getKeystorePath() -- cgit v1.2.1