You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gr...@apache.org on 2010/10/13 17:06:27 UTC
svn commit: r1022127 [2/15] - in
/qpid/branches/grkvlt-network-20101013/qpid/java: ./
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/
broker-plugins/access-control/src/test/java/org/apache/qpid/server/securit...
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallConfigurationTest.java Wed Oct 13 15:05:29 2010
@@ -45,15 +45,15 @@ public class FirewallConfigurationTest e
ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
try
{
- ApplicationRegistry.initialise(reg, 1);
+ ApplicationRegistry.initialise(reg);
// Test config
- assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
- assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.1.2.3", 65535)));
+ assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
+ assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.1.2.3", 65535).toString()));
}
finally
{
- ApplicationRegistry.remove(1);
+ ApplicationRegistry.remove();
}
}
@@ -118,14 +118,14 @@ public class FirewallConfigurationTest e
ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
try
{
- ApplicationRegistry.initialise(reg, 1);
+ ApplicationRegistry.initialise(reg);
// Test config
- assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
}
finally
{
- ApplicationRegistry.remove(1);
+ ApplicationRegistry.remove();
}
}
@@ -141,21 +141,21 @@ public class FirewallConfigurationTest e
ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
try
{
- ApplicationRegistry.initialise(reg, 1);
+ ApplicationRegistry.initialise(reg);
// Test config
- assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
// Switch to deny the connection
writeConfigFile(mainFile, true);
reg.getConfiguration().reparseConfigFileSecuritySections();
- assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
}
finally
{
- ApplicationRegistry.remove(1);
+ ApplicationRegistry.remove();
}
}
@@ -219,10 +219,10 @@ public class FirewallConfigurationTest e
ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
try
{
- ApplicationRegistry.initialise(reg, 1);
+ ApplicationRegistry.initialise(reg);
// Test config
- assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
RandomAccessFile fileBRandom = new RandomAccessFile(fileB, "rw");
fileBRandom.setLength(0);
@@ -237,7 +237,7 @@ public class FirewallConfigurationTest e
reg.getConfiguration().reparseConfigFileSecuritySections();
- assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertTrue(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
fileBRandom = new RandomAccessFile(fileB, "rw");
fileBRandom.setLength(0);
@@ -252,11 +252,11 @@ public class FirewallConfigurationTest e
reg.getConfiguration().reparseConfigFileSecuritySections();
- assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535)));
+ assertFalse(reg.getSecurityManager().accessVirtualhost("test", new InetSocketAddress("127.0.0.1", 65535).toString()));
}
finally
{
- ApplicationRegistry.remove(1);
+ ApplicationRegistry.remove();
}
}
@@ -346,7 +346,7 @@ public class FirewallConfigurationTest e
// Load config
ApplicationRegistry reg = new ConfigurationFileApplicationRegistry(mainFile);
- ApplicationRegistry.initialise(reg, 1);
+ ApplicationRegistry.initialise(reg);
// Test config
VirtualHostRegistry virtualHostRegistry = reg.getVirtualHostRegistry();
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker-plugins/firewall/src/test/java/org/apache/qpid/server/security/access/FirewallPluginTest.java Wed Oct 13 15:05:29 2010
@@ -28,6 +28,7 @@ import java.net.SocketAddress;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.qpid.server.security.Result;
+import org.apache.qpid.server.security.access.ObjectProperties.Property;
import org.apache.qpid.server.security.access.plugins.Firewall;
import org.apache.qpid.server.security.access.plugins.FirewallConfiguration;
import org.apache.qpid.server.util.InternalBrokerBaseCase;
@@ -122,15 +123,23 @@ public class FirewallPluginTest extends
return initialisePlugin(string, null);
}
+ private Result getResult(Firewall plugin, SocketAddress address)
+ {
+ ObjectProperties properties = new ObjectProperties();
+ properties.put(Property.REMOTE_ADDRESS, address.toString());
+
+ return plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, properties);
+ }
+
public void testDefaultAction() throws Exception
{
// Test simple deny
Firewall plugin = initialisePlugin("deny");
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Test simple allow
plugin = initialisePlugin("allow");
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
@@ -142,11 +151,11 @@ public class FirewallPluginTest extends
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{rule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testSingleNetworkRule() throws Exception
@@ -157,11 +166,11 @@ public class FirewallPluginTest extends
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{rule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testSingleHostRule() throws Exception
@@ -174,7 +183,7 @@ public class FirewallPluginTest extends
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("127.0.0.1", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testSingleHostWilcardRule() throws Exception
@@ -187,7 +196,7 @@ public class FirewallPluginTest extends
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("127.0.0.1", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testSeveralFirstAllowsAccess() throws Exception
@@ -206,11 +215,11 @@ public class FirewallPluginTest extends
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{firstRule, secondRule, thirdRule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testSeveralLastAllowsAccess() throws Exception
@@ -229,11 +238,11 @@ public class FirewallPluginTest extends
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{firstRule, secondRule, thirdRule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testNetmask() throws Exception
@@ -243,11 +252,11 @@ public class FirewallPluginTest extends
firstRule.setNetwork("192.168.23.0/24");
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{firstRule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testCommaSeperatedNetmask() throws Exception
@@ -257,11 +266,11 @@ public class FirewallPluginTest extends
firstRule.setNetwork("10.1.1.1/8, 192.168.23.0/24");
Firewall plugin = initialisePlugin("deny", new RuleInfo[]{firstRule});
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("192.168.23.23", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
public void testCommaSeperatedHostnames() throws Exception
@@ -273,10 +282,10 @@ public class FirewallPluginTest extends
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("10.0.0.1", 65535);
- assertEquals(Result.DENIED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.DENIED, getResult(plugin, _address));
// Set IP so that we're connected from the right address
_address = new InetSocketAddress("127.0.0.1", 65535);
- assertEquals(Result.ALLOWED, plugin.access(ObjectType.VIRTUALHOST, _address));
+ assertEquals(Result.ALLOWED, getResult(plugin, _address));
}
}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/etc/log4j.xml
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/etc/log4j.xml?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/etc/log4j.xml (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/etc/log4j.xml Wed Oct 13 15:05:29 2010
@@ -106,9 +106,18 @@
<!-- Examples of additional logging settings -->
<!-- Used to generate extra debug. See debug.log4j.xml -->
- <!--<category additivity="true" name="org.apache.qpid.server.store">
+ <!-- Transaction Logging -->
+ <category additivity="true" name="org.apache.qpid.server.store">
<priority value="debug"/>
- </category-->
+ </category>
+
+ <category additivity="true" name="org.apache.qpid.server.txn">
+ <priority value="debug"/>
+ </category>
+
+ <category additivity="true" name="com.sleepycat.je">
+ <priority value="debug"/>
+ </category>
<!-- Set the commons logging that the XML parser uses to WARN, it is very chatty at debug -->
<logger name="org.apache.commons">
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/qmf/ManagementExchange.java Wed Oct 13 15:05:29 2010
@@ -156,10 +156,11 @@ public class ManagementExchange implemen
AMQShortString name,
boolean durable,
int ticket,
- boolean autoDelete) throws AMQException
+ boolean autoDelete,
+ Map<String, Object> arguments) throws AMQException
{
ManagementExchange exch = new ManagementExchange();
- exch.initialise(host, name, durable, ticket, autoDelete);
+ exch.initialise(host, name, durable, ticket, autoDelete, arguments);
return exch;
}
@@ -180,7 +181,7 @@ public class ManagementExchange implemen
return QPID_MANAGEMENT_TYPE;
}
- public void initialise(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete)
+ public void initialise(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete, Map<String, Object> arguments)
throws AMQException
{
if(!QPID_MANAGEMENT.equals(name))
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java Wed Oct 13 15:05:29 2010
@@ -85,6 +85,7 @@ public class AMQBrokerManagerMBean exten
/**
* Returns an array of the exchange types available for creation.
+ *
* @since Qpid JMX API 1.3
* @throws IOException
*/
@@ -101,6 +102,7 @@ public class AMQBrokerManagerMBean exten
/**
* Returns a list containing the names of the attributes available for the Queue mbeans.
+ *
* @since Qpid JMX API 1.3
* @throws IOException
*/
@@ -111,7 +113,8 @@ public class AMQBrokerManagerMBean exten
/**
* Returns a List of Object Lists containing the requested attribute values (in the same sequence requested) for each queue in the virtualhost.
- * If a particular attribute cant be found or raises an mbean/reflection exception whilst being gathered its value is substituted with the String "-".
+ * If a particular attribute cant be found or raises an mbean/reflection exception whilst being gathered its value is substituted with the String "-"
+ *
* @since Qpid JMX API 1.3
* @throws IOException
*/
@@ -175,7 +178,7 @@ public class AMQBrokerManagerMBean exten
if (exchange == null)
{
exchange = _exchangeFactory.createExchange(new AMQShortString(exchangeName), new AMQShortString(type),
- durable, false, 0);
+ durable, false, Collections.<String, Object>emptyMap(), 0);
_exchangeRegistry.registerExchange(exchange);
if (durable)
{
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Wed Oct 13 15:05:29 2010
@@ -22,6 +22,7 @@ package org.apache.qpid.server;
import org.apache.log4j.Logger;
+import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.framing.AMQMethodBody;
@@ -55,10 +56,10 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.output.ProtocolOutputConverter;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQProtocolEngine;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.queue.IncomingMessage;
@@ -141,6 +142,7 @@ public class AMQChannel implements Sessi
private final AtomicLong _txnCommits = new AtomicLong(0);
private final AtomicLong _txnRejects = new AtomicLong(0);
private final AtomicLong _txnCount = new AtomicLong(0);
+ private final AtomicLong _txnUpdateTime = new AtomicLong(0);
private final AMQProtocolSession _session;
private AtomicBoolean _closing = new AtomicBoolean(false);
@@ -200,6 +202,11 @@ public class AMQChannel implements Sessi
return !(_transaction instanceof AutoCommitTransaction);
}
+ public boolean inTransaction()
+ {
+ return isTransactional() && _txnUpdateTime.get() > 0 && _transaction.getTransactionStartTime() > 0;
+ }
+
private void incrementOutstandingTxnsIfNecessary()
{
if(isTransactional())
@@ -295,7 +302,6 @@ public class AMQChannel implements Sessi
});
deliverCurrentMessageIfComplete();
-
}
}
@@ -333,6 +339,7 @@ public class AMQChannel implements Sessi
{
_transaction.enqueue(destinationQueues, _currentMessage, new MessageDeliveryAction(_currentMessage, destinationQueues, isTransactional()));
incrementOutstandingTxnsIfNecessary();
+ updateTransactionalActivity();
}
}
}
@@ -794,6 +801,7 @@ public class AMQChannel implements Sessi
{
Collection<QueueEntry> ackedMessages = getAckedMessages(deliveryTag, multiple);
_transaction.dequeue(ackedMessages, new MessageAcknowledgeAction(ackedMessages));
+ updateTransactionalActivity();
}
private Collection<QueueEntry> getAckedMessages(long deliveryTag, boolean multiple)
@@ -968,6 +976,17 @@ public class AMQChannel implements Sessi
}
+ /**
+ * Update last transaction activity timestamp
+ */
+ private void updateTransactionalActivity()
+ {
+ if (isTransactional())
+ {
+ _txnUpdateTime.set(System.currentTimeMillis());
+ }
+ }
+
public String toString()
{
return "["+_session.toString()+":"+_channelId+"]";
@@ -1407,4 +1426,36 @@ public class AMQChannel implements Sessi
{
_session.mgmtCloseChannel(_channelId);
}
+
+ public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
+ {
+ if (inTransaction())
+ {
+ long currentTime = System.currentTimeMillis();
+ long openTime = currentTime - _transaction.getTransactionStartTime();
+ long idleTime = currentTime - _txnUpdateTime.get();
+
+ // Log a warning on idle or open transactions
+ if (idleWarn > 0L && idleTime > idleWarn)
+ {
+ CurrentActor.get().message(_logSubject, ChannelMessages.IDLE_TXN(idleTime));
+ _logger.warn("IDLE TRANSACTION ALERT " + _logSubject.toString() + " " + idleTime + " ms");
+ }
+ else if (openWarn > 0L && openTime > openWarn)
+ {
+ CurrentActor.get().message(_logSubject, ChannelMessages.OPEN_TXN(openTime));
+ _logger.warn("OPEN TRANSACTION ALERT " + _logSubject.toString() + " " + openTime + " ms");
+ }
+
+ // Close connection for idle or open transactions that have timed out
+ if (idleClose > 0L && idleTime > idleClose)
+ {
+ getConnectionModel().closeSession(this, AMQConstant.RESOURCE_ERROR, "Idle transaction timed out");
+ }
+ else if (openClose > 0L && openTime > openClose)
+ {
+ getConnectionModel().closeSession(this, AMQConstant.RESOURCE_ERROR, "Open transaction timed out");
+ }
+ }
+ }
}
Added: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerActivator.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerActivator.java?rev=1022127&view=auto
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerActivator.java (added)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerActivator.java Wed Oct 13 15:05:29 2010
@@ -0,0 +1,61 @@
+/*
+ * 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 org.apache.log4j.Logger;
+import org.apache.qpid.BrokerOptions;
+import org.osgi.framework.BundleActivator;
+import org.osgi.framework.BundleContext;
+
+public class BrokerActivator implements BundleActivator
+{
+ private static final Logger _logger = Logger.getLogger(BrokerActivator.class);
+
+ private BundleContext _context = null;
+ private BrokerInstance _instance = null;
+
+ public void start(BundleContext ctx) throws Exception
+ {
+ _context = ctx;
+ _logger.info("Starting broker: " + _context.getBundle().getSymbolicName());
+
+ BrokerOptions options = new BrokerOptions();
+ options.setBind("*");
+ options.setPorts(BrokerOptions.DEFAULT_PORT);
+
+ _instance = new BrokerInstance();
+ _instance.startup(options);
+
+ _context.registerService(BrokerInstance.class.getName(), _instance, null);
+ }
+
+ public void stop(BundleContext ctx) throws Exception
+ {
+ _logger.info("Stopping broker: " + _context.getBundle().getSymbolicName());
+ _instance.shutdown();
+
+ _context = null;
+ _instance = null;
+ }
+
+ public BundleContext getContext()
+ {
+ return _context;
+ }
+}
Added: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerInstance.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerInstance.java?rev=1022127&view=auto
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerInstance.java (added)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/BrokerInstance.java Wed Oct 13 15:05:29 2010
@@ -0,0 +1,335 @@
+/*
+ *
+ * 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.io.File;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.log4j.Logger;
+import org.apache.log4j.PropertyConfigurator;
+import org.apache.log4j.xml.QpidLog4JConfigurator;
+import org.apache.qpid.BrokerOptions;
+import org.apache.qpid.common.QpidProperties;
+import org.apache.qpid.protocol.ReceiverFactory;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+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.BrokerReceiverFactory;
+import org.apache.qpid.server.protocol.BrokerReceiverFactory.VERSION;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
+import org.apache.qpid.ssl.SSLContextFactory;
+import org.apache.qpid.transport.ConnectionSettings;
+import org.apache.qpid.transport.network.IncomingNetworkTransport;
+import org.apache.qpid.transport.network.Transport;
+
+public class BrokerInstance
+{
+ private static Logger _logger;
+
+ public void shutdown()
+ {
+ ApplicationRegistry.remove();
+ }
+
+ public void startup(BrokerOptions options) throws Exception
+ {
+ //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");
+ }
+ _logger = Logger.getLogger(BrokerInstance.class);
+ CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
+
+ String home = System.getProperty(BrokerOptions.QPID_HOME);
+ File defaultConfigFile = new File(home, BrokerOptions.DEFAULT_CONFIG_FILE);
+ File configFile = new File(options.getValue(BrokerOptions.CONFIG, defaultConfigFile.getPath()));
+ if (!configFile.exists())
+ {
+ String error = "File " + configFile + " could not be found. Check the file exists and is readable.";
+ if (home == null)
+ {
+ error = error + "\nNote: " + BrokerOptions.QPID_HOME + " is not set.";
+ }
+
+ throw new StartupException(error);
+ }
+ else
+ {
+ CurrentActor.get().message(BrokerMessages.CONFIG(configFile.getAbsolutePath()));
+ }
+
+
+ String watch = options.getValue(BrokerOptions.WATCH);
+ int logWatchTime = 0;
+ try
+ {
+ logWatchTime = Integer.parseInt(watch);
+ }
+ catch (NumberFormatException e)
+ {
+ System.err.println("Log watch configuration value of " + watch + " is invalid. Must be "
+ + "a non-negative integer. Using default of zero (no watching configured");
+ }
+
+ String log4j = options.getValue(BrokerOptions.LOG4J, System.getProperty("log4j.configuration"));
+ File logConfigFile;
+ if (log4j != null)
+ {
+ logConfigFile = new File(log4j);
+ configureLogging(logConfigFile, logWatchTime);
+ }
+ else
+ {
+ File configFileDirectory = configFile.getParentFile();
+ logConfigFile = new File(configFileDirectory, BrokerOptions.DEFAULT_LOG_CONFIG_FILENAME);
+ configureLogging(logConfigFile, logWatchTime);
+ }
+
+ ConfigurationFileApplicationRegistry config = new ConfigurationFileApplicationRegistry(configFile);
+ ServerConfiguration serverConfig = config.getConfiguration();
+
+ String management = options.getValue(BrokerOptions.MANAGEMENT);
+ updateManagementPort(serverConfig, management);
+
+ // Initialise application registry
+ ApplicationRegistry.initialise(config);
+
+ // We have already loaded the BrokerMessages class by this point so we
+ // need to refresh the locale setting in case 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(ApplicationRegistry.getInstance().getRootMessageLogger());
+
+ try
+ {
+ configureLoggingManagementMBean(logConfigFile, logWatchTime);
+
+ ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean();
+ configMBean.register();
+
+ ServerInformationMBean sysInfoMBean =
+ new ServerInformationMBean(QpidProperties.getBuildVersion(), QpidProperties.getReleaseVersion());
+ sysInfoMBean.register();
+
+ Set<Integer> ports = new HashSet<Integer>();
+ Set<Integer> exclude_0_10 = new HashSet<Integer>();
+ Set<Integer> exclude_0_9_1 = new HashSet<Integer>();
+ Set<Integer> exclude_0_9 = new HashSet<Integer>();
+ Set<Integer> exclude_0_8 = new HashSet<Integer>();
+ parsePortList(ports, options.get(BrokerOptions.PORTS, serverConfig.getPorts()));
+ parsePortList(exclude_0_10, options.get(BrokerOptions.EXCLUDE_0_10, serverConfig.getPortExclude010()));
+ parsePortList(exclude_0_9_1, options.get(BrokerOptions.EXCLUDE_0_9_1, serverConfig.getPortExclude091()));
+ parsePortList(exclude_0_9, options.get(BrokerOptions.EXCLUDE_0_9, serverConfig.getPortExclude09()));
+ parsePortList(exclude_0_8, options.get(BrokerOptions.EXCLUDE_0_8, serverConfig.getPortExclude08()));
+
+ String protocol = options.getValue(BrokerOptions.PROTOCOL, "tcp");
+ String bind = options.getValue(BrokerOptions.BIND);
+ if (bind == null)
+ {
+ bind = serverConfig.getBind();
+ }
+ InetAddress address = null;
+
+ if (bind.equals("*"))
+ {
+ address = new InetSocketAddress(0).getAddress();
+ }
+ else
+ {
+ address = InetAddress.getByName(bind);
+ }
+ String host = address.getCanonicalHostName();
+
+ ConnectionSettings settings = new ConnectionSettings();
+ settings.setProtocol(protocol);
+ settings.setHost(bind);
+
+ String keystorePath = serverConfig.getKeystorePath();
+ String keystorePassword = serverConfig.getKeystorePassword();
+ String certType = serverConfig.getCertType();
+ SSLContextFactory sslFactory = null;
+
+ if (!serverConfig.getSSLOnly())
+ {
+ for (int port : ports)
+ {
+ IncomingNetworkTransport transport = Transport.getIncomingTransport();
+
+ Set<VERSION> supported = BrokerReceiverFactory.ALL_VERSIONS;
+ 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);
+ }
+
+ settings.setPort(port);
+
+ ReceiverFactory factory = new BrokerReceiverFactory(host, supported);
+ transport.accept(settings, factory, sslFactory);
+
+ ApplicationRegistry.getInstance().registerTransport(port, transport);
+ CurrentActor.get().message(BrokerMessages.LISTENING(protocol.toUpperCase(), port));
+ }
+ }
+
+ if (serverConfig.getEnableSSL())
+ {
+ IncomingNetworkTransport transport = Transport.getIncomingTransport();
+ sslFactory = new SSLContextFactory(keystorePath, keystorePassword, certType);
+ settings.setPort(serverConfig.getSSLPort());
+
+ ReceiverFactory factory = new BrokerReceiverFactory(host, BrokerReceiverFactory.ALL_VERSIONS);
+ transport.accept(settings, factory, sslFactory);
+
+ ApplicationRegistry.getInstance().registerTransport(serverConfig.getSSLPort(), transport);
+ CurrentActor.get().message(BrokerMessages.LISTENING(protocol.toUpperCase() + "/SSL", serverConfig.getSSLPort()));
+ }
+
+ CurrentActor.get().message(BrokerMessages.READY());
+ }
+ finally
+ {
+ // Startup is complete so remove the AR initialised Startup actor
+ CurrentActor.remove();
+ }
+ }
+
+ private void configureLogging(File logConfigFile, int logWatchTime) throws Exception
+ {
+ if (logConfigFile.exists() && logConfigFile.canRead())
+ {
+ CurrentActor.get().message(BrokerMessages.LOG_CONFIG(logConfigFile.getAbsolutePath()));
+
+ try
+ {
+ if (logWatchTime > 0)
+ {
+ _logger.info("log file " + logConfigFile.getAbsolutePath() +
+ " will be checked for changes every " + logWatchTime + " seconds");
+ // log4j expects the watch interval in milliseconds
+ QpidLog4JConfigurator.configureAndWatch(logConfigFile.getPath(), logWatchTime * 1000);
+ }
+ else
+ {
+ QpidLog4JConfigurator.configure(logConfigFile.getPath());
+ }
+ }
+ catch (Exception e)
+ {
+ throw new StartupException(e.getMessage(), e);
+ }
+ }
+ else
+ {
+ _logger.info("Logging configuration error: unable to read file " + logConfigFile.getAbsolutePath());
+ _logger.info("Using the fallback internal log4j.properties configuration");
+
+ InputStream propsFile = this.getClass().getResourceAsStream("/log4j.properties");
+ if (propsFile == null)
+ {
+ throw new StartupException("Unable to load the fallback internal log4j.properties configuration file");
+ }
+ else
+ {
+ Properties fallbackProps = new Properties();
+ fallbackProps.load(propsFile);
+ PropertyConfigurator.configure(fallbackProps);
+ }
+ }
+ }
+
+ private void parsePortList(Set<Integer> output, List<String> input) throws StartupException
+ {
+ if (input != null)
+ {
+ for (String port : input)
+ {
+ try
+ {
+ output.add(Integer.parseInt(String.valueOf(port)));
+ }
+ catch (NumberFormatException e)
+ {
+ throw new StartupException("Invalid port: " + port, 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);
+ }
+ }
+ }
+
+ private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception
+ {
+ LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime);
+
+ blm.register();
+ }
+}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java Wed Oct 13 15:05:29 2010
@@ -20,17 +20,6 @@
*/
package org.apache.qpid.server;
-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;
@@ -38,140 +27,74 @@ import org.apache.commons.cli.OptionBuil
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.BrokerOptions;
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.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.NetworkDriver;
-import org.apache.qpid.transport.network.mina.MINANetworkDriver;
/**
- * Main entry point for AMQPD.
- *
+ * Main entry point for Qpid broker.
*/
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 static class InitException extends Exception
- {
- InitException(String msg, Throwable cause)
- {
- super(msg, cause);
- }
- }
-
- protected final Options options = new Options();
- protected CommandLine commandLine;
-
- protected Main(String[] args)
- {
- setOptions(options);
- if (parseCommandline(args))
- {
- execute();
- }
- }
-
- protected boolean parseCommandline(String[] args)
- {
- try
- {
- commandLine = new PosixParser().parse(options, args);
-
- return true;
- }
- catch (ParseException e)
- {
- System.err.println("Error: " + e.getMessage());
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("Qpid", options, true);
-
- return false;
- }
- }
-
@SuppressWarnings("static-access")
- protected void setOptions(Options options)
+ private static Options getOptions()
{
Option help = new Option("h", "help", false, "print this message");
+
Option version = new Option("v", "version", false, "print the version information and exit");
- Option configFile =
- OptionBuilder.withArgName("file").hasArg().withDescription("use given configuration file").withLongOpt("config")
- .create("c");
- Option port =
- OptionBuilder.withArgName("port").hasArg()
+
+ Option configFile = OptionBuilder.hasArg().withArgName("file")
+ .withDescription("use given configuration file")
+ .withLongOpt("config").create("c");
+
+ Option port = OptionBuilder.hasArg().withArgName("port")
.withDescription("listen on the specified port. Overrides any value in the config file")
.withLongOpt("port").create("p");
- Option exclude0_10 =
- OptionBuilder.withArgName("exclude-0-10").hasArg()
- .withDescription("when listening on the specified port do not accept AMQP0-10 connections. The specified port must be one specified on the command line")
+ Option exclude0_10 = OptionBuilder.hasArg().withArgName("exclude-0-10")
+ .withDescription("when listening on the specified port do not accept AMQP0-10 connections. " +
+ "The specified port must be one specified on the command line")
.withLongOpt("exclude-0-10").create();
- Option exclude0_9_1 =
- OptionBuilder.withArgName("exclude-0-9-1").hasArg()
- .withDescription("when listening on the specified port do not accept AMQP0-9-1 connections. The specified port must be one specified on the command line")
+ Option exclude0_9_1 = OptionBuilder.hasArg().withArgName("exclude-0-9-1")
+ .withDescription("when listening on the specified port do not accept AMQP0-9-1 connections. " +
+ "The specified port must be one specified on the command line")
.withLongOpt("exclude-0-9-1").create();
-
- Option exclude0_9 =
- OptionBuilder.withArgName("exclude-0-9").hasArg()
- .withDescription("when listening on the specified port do not accept AMQP0-9 connections. The specified port must be one specified on the command line")
+ Option exclude0_9 = OptionBuilder.hasArg().withArgName("exclude-0-9")
+ .withDescription("when listening on the specified port do not accept AMQP0-9 connections. " +
+ "The specified port must be one specified on the command line")
.withLongOpt("exclude-0-9").create();
-
- Option exclude0_8 =
- OptionBuilder.withArgName("exclude-0-8").hasArg()
- .withDescription("when listening on the specified port do not accept AMQP0-8 connections. The specified port must be one specified on the command line")
+ Option exclude0_8 = OptionBuilder.hasArg().withArgName("exclude-0-8")
+ .withDescription("when listening on the specified port do not accept AMQP0-8 connections. " +
+ "The specified port must be one specified on the command line")
.withLongOpt("exclude-0-8").create();
-
- Option mport =
- OptionBuilder.withArgName("mport").hasArg()
+ Option mport = OptionBuilder.hasArg().withArgName("mport")
.withDescription("listen on the specified management port. Overrides any value in the config file")
.withLongOpt("mport").create("m");
-
- Option bind =
- OptionBuilder.withArgName("bind").hasArg()
+ Option bind = OptionBuilder.hasArg().withArgName("bind")
.withDescription("bind to the specified address. Overrides any value in the config file")
.withLongOpt("bind").create("b");
- 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
- + " in the same directory as the configuration file").withLongOpt("logconfig").create("l");
- Option logwatchconfig =
- OptionBuilder.withArgName("logwatch").hasArg()
- .withDescription("monitor the log file configuration file for changes. Units are seconds. "
- + "Zero means do not check for changes.").withLongOpt("logwatch").create("w");
+
+ Option logconfig = OptionBuilder.hasArg().withArgName("logconfig")
+ .withDescription("use the specified log4j xml configuration file. By " +
+ "default looks for a file named " + BrokerOptions.DEFAULT_LOG_CONFIG_FILENAME +
+ " in the same directory as the configuration file")
+ .withLongOpt("logconfig").create("l");
+
+ Option logwatchconfig = OptionBuilder.hasArg().withArgName("logwatch")
+ .withDescription("monitor the log file configuration file for changes. Units are seconds. " +
+ "Zero means do not check for changes.")
+ .withLongOpt("logwatch").create("w");
+
+ Option protocol = OptionBuilder.hasArg().withArgName("protocol")
+ .withDescription("Change the transport protocol.")
+ .withLongOpt("transport").create("t");
+ Options options = new Options();
options.addOption(help);
options.addOption(version);
options.addOption(configFile);
@@ -184,435 +107,75 @@ public class Main
options.addOption(exclude0_8);
options.addOption(mport);
options.addOption(bind);
+ options.addOption(protocol);
+ return options;
}
- protected void execute()
- {
- // 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"))
- {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("Qpid", options, true);
- }
- 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
- {
- 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);
- }
- }
- }
-
- protected void shutdown(int status)
- {
- ApplicationRegistry.removeAll();
- System.exit(status);
- }
-
- protected void startup() throws Exception
+ public static void main(String[] args)
{
- 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 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()));
- }
-
- String logConfig = commandLine.getOptionValue("l");
- String logWatchConfig = commandLine.getOptionValue("w", "0");
-
- int logWatchTime = 0;
+ Options options = getOptions();
try
{
- logWatchTime = Integer.parseInt(logWatchConfig);
- }
- catch (NumberFormatException e)
- {
- System.err.println("Log watch configuration value of " + logWatchConfig + " is invalid. Must be "
- + "a non-negative integer. Using default of zero (no watching configured");
- }
-
- File logConfigFile;
- if (logConfig != null)
- {
- logConfigFile = new File(logConfig);
- configureLogging(logConfigFile, logWatchTime);
- }
- else
- {
- File configFileDirectory = configFile.getParentFile();
- logConfigFile = new File(configFileDirectory, DEFAULT_LOG_CONFIG_FILENAME);
- configureLogging(logConfigFile, logWatchTime);
- }
-
- 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());
+ CommandLine commandLine = new PosixParser().parse(options, args);
-
- try
- {
- configureLoggingManagementMBean(logConfigFile, logWatchTime);
-
- ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean();
- configMBean.register();
-
- ServerInformationMBean sysInfoMBean =
- new ServerInformationMBean(QpidProperties.getBuildVersion(), QpidProperties.getReleaseVersion());
- sysInfoMBean.register();
-
-
- String[] portStr = commandLine.getOptionValues("p");
-
- Set<Integer> ports = new HashSet<Integer>();
- Set<Integer> exclude_0_10 = new HashSet<Integer>();
- Set<Integer> exclude_0_9_1 = new HashSet<Integer>();
- Set<Integer> exclude_0_9 = new HashSet<Integer>();
- Set<Integer> exclude_0_8 = new HashSet<Integer>();
-
- 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(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"))
- {
- 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)
+ // 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"))
+ {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("Qpid", options, true);
+ }
+ else if (commandLine.hasOption("v"))
+ {
+ String ver = QpidProperties.getVersionString();
+
+ StringBuilder protocol = new StringBuilder();
+ for (ProtocolVersion pv : ProtocolVersion.getSupportedProtocolVersions())
{
-
- NetworkDriver driver = new MINANetworkDriver();
-
- Set<VERSION> 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))
+ if (protocol.length() > 0)
{
- supported.remove(VERSION.v0_9);
+ protocol.append(", ");
}
- if(exclude_0_8.contains(port))
- {
- supported.remove(VERSION.v0_8);
- }
-
- MultiVersionProtocolEngineFactory protocolEngineFactory =
- new MultiVersionProtocolEngineFactory(hostName, supported);
-
-
-
- driver.bind(port, new InetAddress[]{bindAddress}, protocolEngineFactory,
- serverConfig.getNetworkConfiguration(), null);
- ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, port),
- new QpidAcceptor(driver,"TCP"));
- CurrentActor.get().message(BrokerMessages.LISTENING("TCP", port));
-
+ protocol.append(pv.getMajorVersion()).append('-').append(pv.getMinorVersion());
}
-
- }
-
- if (serverConfig.getEnableSSL())
- {
- sslFactory = new SSLContextFactory(keystorePath, keystorePassword, certType);
- NetworkDriver driver = new MINANetworkDriver();
- driver.bind(serverConfig.getSSLPort(), new InetAddress[]{bindAddress},
- new AMQProtocolEngineFactory(), serverConfig.getNetworkConfiguration(), sslFactory);
- ApplicationRegistry.getInstance().addAcceptor(new InetSocketAddress(bindAddress, serverConfig.getSSLPort()),
- new QpidAcceptor(driver,"TCP"));
- CurrentActor.get().message(BrokerMessages.LISTENING("TCP/SSL", serverConfig.getSSLPort()));
+
+ System.out.println(ver + " AMQP version(s) [major.minor]: (" + protocol + ")");
}
-
- CurrentActor.get().message(BrokerMessages.READY());
-
- }
- finally
- {
- // Startup is complete so remove the AR initialised Startup actor
- CurrentActor.remove();
- }
-
-
-
- }
-
- private void parsePortArray(Set<Integer> ports, String[] portStr)
- throws InitException
- {
- if(portStr != null)
- {
- for(int i = 0; i < portStr.length; i++)
+ else
{
- try
+ BrokerOptions brokerOptions = new BrokerOptions();
+ for (String option : BrokerOptions.COMMAND_LINE_OPTIONS)
{
- ports.add(Integer.parseInt(portStr[i]));
+ brokerOptions.put(option, commandLine.getOptionValues(option));
}
- catch (NumberFormatException e)
- {
- throw new InitException("Invalid port: " + portStr[i], e);
- }
- }
- }
- }
-
- private void parsePortList(Set<Integer> output, List input)
- throws InitException
- {
- if(input != null)
- {
- for(Object port : input)
- {
+ BrokerInstance broker = new BrokerInstance();
try
{
- output.add(Integer.parseInt(String.valueOf(port)));
+ broker.startup(brokerOptions);
}
- catch (NumberFormatException e)
+ catch (StartupException e)
{
- throw new InitException("Invalid port: " + port, e);
+ System.out.println("Error initialising message broker: " + e);
+ e.printStackTrace();
+ broker.shutdown();
+ System.exit(1);
}
- }
- }
- }
-
- /**
- * 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
- {
- 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 * 1000);
- }
- catch (Exception e)
- {
- throw new InitException(e.getMessage(),e);
- }
- }
- else
- {
- try
- {
- QpidLog4JConfigurator.configure(logConfigFile.getPath());
- }
- catch (Exception e)
+ catch (Throwable e)
{
- throw new InitException(e.getMessage(),e);
+ System.out.println("Error running message broker: " + e);
+ e.printStackTrace();
+ broker.shutdown();
+ System.exit(1);
}
}
}
- else
+ catch (ParseException pe)
{
- 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();
- }
- }
+ System.err.println("Error: " + pe.getMessage());
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("Qpid", options, true);
+ System.exit(1);
}
}
-
- private void configureLoggingManagementMBean(File logConfigFile, int logWatchTime) throws Exception
- {
- LoggingManagementMBean blm = new LoggingManagementMBean(logConfigFile.getPath(),logWatchTime);
-
- blm.register();
- }
}
Added: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/StartupException.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/StartupException.java?rev=1022127&view=auto
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/StartupException.java (added)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/StartupException.java Wed Oct 13 15:05:29 2010
@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+public class StartupException extends Exception {
+
+ public StartupException(String msg, Throwable cause)
+ {
+ super(msg, cause);
+ }
+
+ public StartupException(String msg)
+ {
+ super(msg);
+ }
+}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/Binding.java Wed Oct 13 15:05:29 2010
@@ -43,7 +43,7 @@ public class Binding
_bindingKey = bindingKey;
_queue = queue;
_exchange = exchange;
- _arguments = arguments == null ? Collections.EMPTY_MAP : Collections.unmodifiableMap(arguments);
+ _arguments = arguments == null ? Collections.<String, Object>emptyMap() : Collections.unmodifiableMap(arguments);
}
public UUID getId()
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/binding/BindingFactory.java Wed Oct 13 15:05:29 2010
@@ -33,8 +33,6 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.configuration.BindingConfigType;
import org.apache.qpid.server.configuration.ConfigStore;
import org.apache.qpid.server.configuration.ConfiguredObject;
-import org.apache.qpid.server.configuration.QueueConfiguration;
-import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.BindingMessages;
@@ -51,7 +49,6 @@ public class BindingFactory
private final ConcurrentHashMap<BindingImpl, BindingImpl> _bindings = new ConcurrentHashMap<BindingImpl, BindingImpl>();
-
public BindingFactory(final VirtualHost vhost)
{
this(vhost, vhost.getExchangeRegistry().getDefaultExchange());
@@ -76,8 +73,6 @@ public class BindingFactory
return _virtualHost;
}
-
-
private final class BindingImpl extends Binding implements AMQQueue.Task, Exchange.Task, BindingConfig
{
private final BindingLogSubject _logSubject;
@@ -88,7 +83,6 @@ public class BindingFactory
{
super(queue.getVirtualHost().getConfigStore().createId(), bindingKey, queue, exchange, arguments);
_logSubject = new BindingLogSubject(bindingKey,exchange,queue);
-
}
@@ -157,7 +151,10 @@ public class BindingFactory
private boolean makeBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments, boolean restore, boolean force) throws AMQSecurityException, AMQInternalException
{
- assert queue != null;
+ if (queue == null)
+ {
+ throw new AMQInternalException("Queue cannot be null");
+ }
if (bindingKey == null)
{
bindingKey = "";
@@ -224,7 +221,10 @@ public class BindingFactory
public Binding removeBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments) throws AMQSecurityException, AMQInternalException
{
- assert queue != null;
+ if (queue == null)
+ {
+ throw new AMQInternalException("Queue cannot be null");
+ }
if (bindingKey == null)
{
bindingKey = "";
@@ -269,16 +269,19 @@ public class BindingFactory
public Binding getBinding(String bindingKey, AMQQueue queue, Exchange exchange, Map<String, Object> arguments)
{
- assert queue != null;
- if(bindingKey == null)
+ if (queue == null)
+ {
+ throw new RuntimeException("Queue cannot be null"); // FIXME
+ }
+ if (bindingKey == null)
{
bindingKey = "";
}
- if(exchange == null)
+ if (exchange == null)
{
exchange = _defaultExchange;
}
- if(arguments == null)
+ if (arguments == null)
{
arguments = Collections.emptyMap();
}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java Wed Oct 13 15:05:29 2010
@@ -42,7 +42,6 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
-import org.apache.qpid.transport.NetworkDriverConfiguration;
import sun.misc.Signal;
import sun.misc.SignalHandler;
@@ -654,34 +653,34 @@ public class ServerConfiguration extends
return getIntValue("connector.processors", 4);
}
- public List getPorts()
+ public List<String> getPorts()
{
- return getListValue("connector.port", Collections.singletonList(DEFAULT_PORT));
+ return getListValue("connector.port", Collections.<String>singletonList(Integer.toString(DEFAULT_PORT)));
}
- public List getPortExclude010()
+ public List<String> getPortExclude010()
{
return getListValue("connector.non010port");
}
- public List getPortExclude091()
+ public List<String> getPortExclude091()
{
return getListValue("connector.non091port");
}
- public List getPortExclude09()
+ public List<String> getPortExclude09()
{
return getListValue("connector.non09port");
}
- public List getPortExclude08()
+ public List<String> getPortExclude08()
{
return getListValue("connector.non08port");
}
public String getBind()
{
- return getStringValue("connector.bind", "wildcard");
+ return getStringValue("connector.bind", "*");
}
public int getReceiveBufferSize()
@@ -775,57 +774,4 @@ public class ServerConfiguration extends
getLongValue("housekeeping.expiredMessageCheckPeriod",
DEFAULT_HOUSEKEEPING_PERIOD));
}
-
- public NetworkDriverConfiguration getNetworkConfiguration()
- {
- return new NetworkDriverConfiguration()
- {
-
- public Integer getTrafficClass()
- {
- return null;
- }
-
- public Boolean getTcpNoDelay()
- {
- // Can't call parent getTcpNoDelay since it just calls this one
- return getBooleanValue("connector.tcpNoDelay", true);
- }
-
- public Integer getSoTimeout()
- {
- return null;
- }
-
- public Integer getSoLinger()
- {
- return null;
- }
-
- public Integer getSendBufferSize()
- {
- return getBufferWriteLimit();
- }
-
- public Boolean getReuseAddress()
- {
- return null;
- }
-
- public Integer getReceiveBufferSize()
- {
- return getBufferReadLimit();
- }
-
- public Boolean getOOBInline()
- {
- return null;
- }
-
- public Boolean getKeepAlive()
- {
- return null;
- }
- };
- }
}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Wed Oct 13 15:05:29 2010
@@ -313,4 +313,24 @@ public class VirtualHostConfiguration ex
{
return getIntValue("housekeeping.poolSize", Runtime.getRuntime().availableProcessors());
}
+
+ public long getTransactionTimeoutOpenWarn()
+ {
+ return getLongValue("transactionTimeout.openWarn", 0L);
+ }
+
+ public long getTransactionTimeoutOpenClose()
+ {
+ return getLongValue("transactionTimeout.openClose", 0L);
+ }
+
+ public long getTransactionTimeoutIdleWarn()
+ {
+ return getLongValue("transactionTimeout.idleWarn", 0L);
+ }
+
+ public long getTransactionTimeoutIdleClose()
+ {
+ return getLongValue("transactionTimeout.idleClose", 0L);
+ }
}
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/plugins/ConfigurationPlugin.java Wed Oct 13 15:05:29 2010
@@ -67,6 +67,7 @@ public abstract class ConfigurationPlugi
return _configuration;
}
+ @SuppressWarnings("unchecked")
public <C extends ConfigurationPlugin> C getConfiguration(String plugin)
{
return (C) _pluginConfiguration.get(plugin);
@@ -126,18 +127,6 @@ public abstract class ConfigurationPlugi
elements.remove(tag);
}
- if (_logger.isInfoEnabled())
- {
- if (!elements.isEmpty())
- {
- _logger.info("Elements to lookup:" + path);
- for (String tag : elements)
- {
- _logger.info("Tag:'" + tag + "'");
- }
- }
- }
-
// Process the elements in the configuration
for (String element : elements)
{
@@ -151,12 +140,6 @@ public abstract class ConfigurationPlugi
}
List<ConfigurationPlugin> handlers = configurationManager.getConfigurationPlugins(configurationElement, handled);
-
- if(_logger.isDebugEnabled())
- {
- _logger.debug("For '" + element + "' found handlers (" + handlers.size() + "):" + handlers);
- }
-
for (ConfigurationPlugin plugin : handlers)
{
_pluginConfiguration.put(plugin.getClass().getName(), plugin);
@@ -241,14 +224,15 @@ public abstract class ConfigurationPlugi
return _configuration.getBoolean(property, defaultValue);
}
- protected List getListValue(String property)
+ protected List<String> getListValue(String property)
{
- return getListValue(property, Collections.EMPTY_LIST);
+ return getListValue(property, Collections.<String>emptyList());
}
- protected List getListValue(String property, List defaultValue)
+ @SuppressWarnings("unchecked")
+ protected List<String> getListValue(String property, List<String> defaultValue)
{
- return _configuration.getList(property, defaultValue);
+ return (List<String>) _configuration.getList(property, defaultValue);
}
/// Validation Helpers
Modified: qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java?rev=1022127&r1=1022126&r2=1022127&view=diff
==============================================================================
--- qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java (original)
+++ qpid/branches/grkvlt-network-20101013/qpid/java/broker/src/main/java/org/apache/qpid/server/connection/ConnectionRegistry.java Wed Oct 13 15:05:29 2010
@@ -20,19 +20,19 @@
*/
package org.apache.qpid.server.connection;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.log4j.Logger;
-import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQException;
import org.apache.qpid.common.Closeable;
import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
public class ConnectionRegistry implements IConnectionRegistry, Closeable
{
- private List<AMQProtocolSession> _registry = new CopyOnWriteArrayList<AMQProtocolSession>();
+ private List<AMQConnectionModel> _registry = new CopyOnWriteArrayList<AMQConnectionModel>();
private Logger _logger = Logger.getLogger(ConnectionRegistry.class);
@@ -40,44 +40,42 @@ public class ConnectionRegistry implemen
{
// None required
}
-
- public void expireClosedChannels()
- {
- for (AMQProtocolSession connection : _registry)
- {
- connection.closeIfLingeringClosedChannels();
- }
- }
/** Close all of the currently open connections. */
public void close()
{
while (!_registry.isEmpty())
{
- AMQProtocolSession connection = _registry.get(0);
-
- try
- {
- connection.closeConnection(0, new AMQConnectionException(AMQConstant.INTERNAL_ERROR, "Broker is shutting down",
- 0, 0,
- connection.getProtocolOutputConverter().getProtocolMajorVersion(),
- connection.getProtocolOutputConverter().getProtocolMinorVersion(),
- (Throwable) null), true);
- }
- catch (AMQException e)
- {
- _logger.warn("Error closing connection:" + e.getMessage());
- }
+ AMQConnectionModel connection = _registry.get(0);
+ closeConnection(connection, AMQConstant.INTERNAL_ERROR, "Broker is shutting down");
+ }
+ }
+
+ public void closeConnection(AMQConnectionModel connection, AMQConstant cause, String message)
+ {
+ try
+ {
+ connection.close(cause, message);
+ }
+ catch (AMQException e)
+ {
+ _logger.warn("Error closing connection:" + e.getMessage());
}
}
- public void registerConnection(AMQProtocolSession connnection)
+ public void registerConnection(AMQConnectionModel connnection)
{
_registry.add(connnection);
}
- public void deregisterConnection(AMQProtocolSession connnection)
+ public void deregisterConnection(AMQConnectionModel connnection)
{
_registry.remove(connnection);
}
+
+ @Override
+ public List<AMQConnectionModel> getConnections()
+ {
+ return new ArrayList<AMQConnectionModel>(_registry);
+ }
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org