Author: rgemmell Date: 2010-01-11 10:00:27 -0500 (Mon, 11 Jan 2010) New Revision: 3778 Added: store/branches/java/0.5-release/bin/bindingsWorkaround.sh store/branches/java/0.5-release/src/tools/java/BDBStoreBindingsWorkaround-log4j.xml store/branches/java/0.5-release/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreBindingsWorkaround.java Log: Add initial version of tool to workaround broker bindings issue. Use the broker config to load the VirtualHosts and then insert required durable Exchange entries into the store Added: store/branches/java/0.5-release/bin/bindingsWorkaround.sh =================================================================== --- store/branches/java/0.5-release/bin/bindingsWorkaround.sh (rev 0) +++ store/branches/java/0.5-release/bin/bindingsWorkaround.sh 2010-01-11 15:00:27 UTC (rev 3778) @@ -0,0 +1,26 @@ +#!/bin/bash + +QPID_VERSION=0.5 + +if [ -z "$QPID_HOME" ]; then + echo "QPID_HOME not set. Exiting" + exit 1 +fi + +if [ -z "$QPID_WORK" ]; then + echo "QPID_WORK not set. Exiting" + exit 1 +fi + +QPID_SYS_PROPS="-DQPID_HOME=$QPID_HOME -DQPID_WORK=$QPID_WORK" + +LIBS=$QPID_HOME/lib/qpid-bdbtools-$QPID_VERSION.jar:$QPID_HOME/lib/je-3.3.62.jar:$QPID_HOME/lib/qpid-bdbstore-$QPID_VERSION.jar:$QPID_HOME/lib/qpid-all.jar + +java -Dlog4j.configuration=BDBStoreBindingsWorkaround-log4j.xml $QPID_SYS_PROPS -cp $LIBS org.apache.qpid.server.store.berkeleydb.BDBStoreBindingsWorkaround "$@" +exitValue=$? + +if [ $exitValue != 0 ] +then + exit $exitValue +fi + Added: store/branches/java/0.5-release/src/tools/java/BDBStoreBindingsWorkaround-log4j.xml =================================================================== --- store/branches/java/0.5-release/src/tools/java/BDBStoreBindingsWorkaround-log4j.xml (rev 0) +++ store/branches/java/0.5-release/src/tools/java/BDBStoreBindingsWorkaround-log4j.xml 2010-01-11 15:00:27 UTC (rev 3778) @@ -0,0 +1,45 @@ +<?xml version="1.0"?> +<!-- + - + - 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. + - + --> +<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd"> +<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/"> + + <appender name="STDOUT" class="org.apache.log4j.ConsoleAppender"> + <layout class="org.apache.log4j.PatternLayout"> + <param name="ConversionPattern" value="%d %-5p - %m%n"/> + </layout> + </appender> + + <category name="org.apache.qpid.server.store.berkeleydb.BDBStoreBindingsWorkaround"> + <priority value="info"/> + </category> + + <!-- Only show errors --> + <category name="org.apache"> + <priority value="error"/> + </category> + + <root> + <priority value="error"/> + <appender-ref ref="STDOUT"/> + </root> + +</log4j:configuration> Added: store/branches/java/0.5-release/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreBindingsWorkaround.java =================================================================== --- store/branches/java/0.5-release/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreBindingsWorkaround.java (rev 0) +++ store/branches/java/0.5-release/src/tools/java/org/apache/qpid/server/store/berkeleydb/BDBStoreBindingsWorkaround.java 2010-01-11 15:00:27 UTC (rev 3778) @@ -0,0 +1,310 @@ +/* + * 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.store.berkeleydb; + +import org.apache.qpid.server.store.berkeleydb.BDBMessageStore; + +import org.apache.commons.cli.Option; +import org.apache.commons.cli.OptionBuilder; +import org.apache.commons.configuration.ConfigurationException; +import org.apache.qpid.AMQException; +import org.apache.qpid.configuration.Configuration; +import org.apache.qpid.configuration.Configuration.InitException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.configuration.ServerConfiguration; +import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.exchange.ExchangeRegistry; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry; +import org.apache.qpid.server.store.berkeleydb.DatabaseVisitor; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.sleepycat.bind.tuple.TupleBinding; +import com.sleepycat.je.DatabaseEntry; +import com.sleepycat.je.DatabaseException; + +import java.io.File; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +public class BDBStoreBindingsWorkaround +{ + private static final Logger _logger = LoggerFactory.getLogger(BDBStoreBindingsWorkaround.class); + + private Configuration _config; + private boolean _initialised = false; + + public static void main(String[] args) throws Configuration.InitException + { + BDBStoreBindingsWorkaround tool = new BDBStoreBindingsWorkaround(args); + + tool.start(); + + //Shut down the JVM gracefully, the ShutdownHook will stop the VirtualHosts. + System.exit(0); + } + + public BDBStoreBindingsWorkaround(String[] args) throws InitException + { + Runtime.getRuntime().addShutdownHook(new Thread(new ShutdownHook(this))); + + loadConfig(args); + } + + @SuppressWarnings("static-access") + private void loadConfig(String args[]) throws InitException + { + _config = new Configuration(); + + Option configFile = + OptionBuilder.withArgName("file").hasArg() + .withDescription("use given configuration file By " + + "default looks for a file named " + + Configuration.DEFAULT_CONFIG_FILE + " in " + Configuration.QPID_HOME) + .withLongOpt("config") + .create("c"); + + _config.setOption(configFile); + + _config.processCommandline(args); + } + + /** + * Simple ShutdownHook to cleanly shutdown the VirtualHosts on JVM shut down + */ + protected class ShutdownHook implements Runnable + { + BDBStoreBindingsWorkaround _tool; + + ShutdownHook(BDBStoreBindingsWorkaround bindingsTool) + { + _tool = bindingsTool; + } + + public void run() + { + _tool.shutdown(); + } + } + + protected void shutdown() + { + if (_initialised) + { + ApplicationRegistry.remove(1); + } + } + + protected void start() + { + _initialised = false; + + _logger.info("BDBStore BindingsWorkaround process commencing"); + + loadVirtualHosts(); + + if (!_initialised) + { + System.exit(1); + } + + addDurableExchangesToStoreIfRequired(); + _logger.info("Workaround process complete"); + } + + private void loadVirtualHosts() + { + final File configFile = _config.getConfigFile(); + + if (!configFile.exists()) + { + _logger.error("Config file not found:" + configFile.getAbsolutePath()); + _logger.error("Options: [-c <broker config file>] : Defaults to \"$QPID_HOME/etc/config.xml\""); + return; + } + else + { + _logger.info("Using config file :" + configFile.getAbsolutePath()); + } + + try + { + _logger.info("Starting the VirtualHosts"); + + ConfigurationFileApplicationRegistry registry = new ConfigurationFileApplicationRegistry(configFile); + + disableManagementStartup(registry.getConfiguration()); + + ApplicationRegistry.remove(1); + ApplicationRegistry.initialise(registry); + + checkMessageStores(); + _initialised = true; + } + catch (ConfigurationException e) + { + _logger.error("Unable to load configuration due to configuration error: " + e.getMessage()); + e.printStackTrace(); + System.exit(1); + } + catch (Exception e) + { + _logger.error("Unable to load configuration due to: " + e.getMessage()); + e.printStackTrace(); + System.exit(1); + } + } + + private void disableManagementStartup(ServerConfiguration config) + { + /* + * Update the server config to indicate management is disabled, which + * should prompt usage of the NoopManagedObjectRegistry instead of + * the JMXManagedObjectRegistry. + */ + config.setManagementEnabled(false); + + /* + * Set the com.sun.management.jmxremote property (to any value) + * if it isnt already set. + * + * The JMXManagedObjectRegistry checks for this system property before + * starting its own ConnectorServer programatically as this (usually) + * implies that the JVM was started explicitly with properties to enable + * its out-of-the-box JMX agent, to which JMXManagedObjectRegistry defers. + */ + if(System.getProperty("com.sun.management.jmxremote") == null) + { + System.setProperty("com.sun.management.jmxremote", "8999"); + } + } + + private void checkMessageStores() + { + Collection<VirtualHost> vhosts = ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHosts(); + + for (VirtualHost vhost : vhosts) + { + if(! (vhost.getMessageStore() instanceof BDBMessageStore) ) + { + _logger.warn("Virtualhost '" + vhost.getName() + "' is not using a BDBMessageStore. " + + "No changes will be made to it."); + } + } + } + + private static List<AMQShortString> getBDBStoreExchangeNames(final VirtualHost vhost, final BDBMessageStore store) + { + final List<AMQShortString> exchanges = new ArrayList<AMQShortString>(); + + //Create a visitor to visit the Exchange entries and gather the names + DatabaseVisitor exchangeVisitor = new DatabaseVisitor() + { + @SuppressWarnings("unchecked") + public void visit(DatabaseEntry key, DatabaseEntry value) throws DatabaseException + { + _logger.debug("Visiting new Exchange store entry"); + + TupleBinding binding = new ExchangeTB(vhost); + Exchange exchange = (Exchange) binding.entryToObject(value); + + if(exchange != null) + { + AMQShortString name = exchange.getName(); + if(name != null) + { + _logger.debug("Visited store entry for Exchange: " + name); + exchanges.add(name); + } + } + } + }; + + try + { + store.visitExchanges(exchangeVisitor); + } + catch (Exception e) + { + _logger.error("Error retrieving exiting Exchange names from the store: " + e.getMessage()); + e.printStackTrace(); + } + + return exchanges; + } + + private static void addDurableExchangesToStoreIfRequired() + { + Collection<VirtualHost> vhosts = ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHosts(); + + //For each active VHost, add all durable exchanges in the ExchangeRegistry to the store if they arent already in it. + for(VirtualHost vhost : vhosts) + { + _logger.info("Beginning process for VirtualHost: " + vhost.getName()); + + BDBMessageStore store; + + if(!(vhost.getMessageStore() instanceof BDBMessageStore) ) + { + _logger.info("Store is not a BDBMessageStore, skipping"); + continue; + } + else + { + store = (BDBMessageStore) vhost.getMessageStore(); + } + + List<AMQShortString> bdbExchangeNames = getBDBStoreExchangeNames(vhost, store); + + ExchangeRegistry registry = vhost.getExchangeRegistry(); + Collection<AMQShortString> exchangeNames = registry.getExchangeNames(); + + for(AMQShortString exchangeName : exchangeNames) + { + Exchange exchange = registry.getExchange(exchangeName); + if(exchange.isDurable()) + { + if(bdbExchangeNames.contains(exchangeName)) + { + _logger.info("Store already contains entry for Exchange: " + exchangeName); + } + else + { + try + { + _logger.info("Adding store entry for Exchange: " + exchangeName); + store.createExchange(exchange); + } + catch (AMQException e) + { + _logger.error("Error adding entry to store for Exchange '" + exchangeName + "':" + e.getMessage()); + e.printStackTrace(); + } + } + } + } + } + } + +}