You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2011/08/15 13:26:57 UTC
svn commit: r1157780 [9/13] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: ./
broker-plugins/access-control/
broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/
broker-plugins/access-control/src/main/java/org/...
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java Mon Aug 15 11:26:46 2011
@@ -52,9 +52,7 @@ public interface BrokerDetails
public static final int DEFAULT_PORT = 5672;
- public static final String SOCKET = "socket";
public static final String TCP = "tcp";
- public static final String VM = "vm";
public static final String DEFAULT_TRANSPORT = TCP;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java Mon Aug 15 11:26:46 2011
@@ -27,7 +27,7 @@ import java.util.List;
/**
Connection URL format
- amqp://[user:pass@][clientid]/virtualhost?brokerlist='tcp://host:port?option=\'value\'&option=\'value\';vm://:3/virtualpath?option=\'value\''&failover='method?option=\'value\'&option='value''"
+ amqp://[user:pass@][clientid]/virtualhost?brokerlist='tcp://host:port?option=\'value\'&option=\'value\';tcp://host:port/virtualpath?option=\'value\''&failover='method?option=\'value\'&option='value''"
Options are of course optional except for requiring a single broker in the broker list.
The option seperator is defined to be either '&' or ','
*/
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java Mon Aug 15 11:26:46 2011
@@ -20,7 +20,6 @@
*/
package org.apache.qpid.jms;
-import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.jms.failover.FailoverExchangeMethod;
import org.apache.qpid.jms.failover.FailoverMethod;
import org.apache.qpid.jms.failover.FailoverRoundRobinServers;
@@ -51,7 +50,7 @@ public class FailoverPolicy
private long _lastMethodTime;
private long _lastFailTime;
- public FailoverPolicy(ConnectionURL connectionDetails, AMQConnection conn)
+ public FailoverPolicy(ConnectionURL connectionDetails, Connection conn)
{
FailoverMethod method;
@@ -83,7 +82,7 @@ public class FailoverPolicy
*/
if (failoverMethod.equals(FailoverMethod.SINGLE_BROKER))
{
- method = new FailoverRoundRobinServers(connectionDetails);
+ method = new FailoverSingleServer(connectionDetails);
}
else
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java Mon Aug 15 11:26:46 2011
@@ -32,9 +32,9 @@ import javax.jms.Session;
import org.apache.qpid.client.AMQAnyDestination;
import org.apache.qpid.client.AMQBrokerDetails;
-import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.jms.BrokerDetails;
+import org.apache.qpid.jms.Connection;
import org.apache.qpid.jms.ConnectionURL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +58,7 @@ public class FailoverExchangeMethod impl
private static final Logger _logger = LoggerFactory.getLogger(FailoverExchangeMethod.class);
/** This is not safe to use until attainConnection is called */
- private AMQConnection _conn;
+ private Connection _conn;
/** Protects the broker list when modifications happens */
private Object _brokerListLock = new Object();
@@ -80,7 +80,7 @@ public class FailoverExchangeMethod impl
/** Denotes the number of failed attempts **/
private int _failedAttemps = 0;
- public FailoverExchangeMethod(ConnectionURL connectionDetails, AMQConnection conn)
+ public FailoverExchangeMethod(ConnectionURL connectionDetails, Connection conn)
{
_connectionDetails = connectionDetails;
_originalBrokerDetail = _connectionDetails.getBrokerDetails(0);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java Mon Aug 15 11:26:46 2011
@@ -79,11 +79,6 @@ public class MockAMQConnection extends A
super(connectionURL, sslConfig);
}
- protected MockAMQConnection(String username, String password, String clientName, String virtualHost)
- {
- super(username, password, clientName, virtualHost);
- }
-
@Override
public ProtocolVersion makeBrokerConnection(BrokerDetails brokerDetail) throws IOException
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/protocol/AMQProtocolHandlerTest.java Mon Aug 15 11:26:46 2011
@@ -20,23 +20,24 @@
*/
package org.apache.qpid.client.protocol;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
import junit.framework.TestCase;
-import org.apache.qpid.framing.AMQFrame;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQAuthenticationException;
+import org.apache.qpid.client.MockAMQConnection;
+import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.framing.AMQBody;
+import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.framing.amqp_8_0.BasicRecoverOkBodyImpl;
-import org.apache.qpid.AMQException;
import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.transport.TestNetworkDriver;
-import org.apache.qpid.client.MockAMQConnection;
-import org.apache.qpid.client.AMQAuthenticationException;
-import org.apache.qpid.client.state.AMQState;
+import org.apache.qpid.transport.TestNetworkConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
/**
* This is a test address QPID-1431 where frame listeners would fail to be notified of an incomming exception.
*
@@ -72,8 +73,8 @@ public class AMQProtocolHandlerTest exte
public void setUp() throws Exception
{
//Create a new ProtocolHandler with a fake connection.
- _handler = new AMQProtocolHandler(new MockAMQConnection("amqp://guest:guest@client/test?brokerlist='vm://:1'"));
- _handler.setNetworkDriver(new TestNetworkDriver());
+ _handler = new AMQProtocolHandler(new MockAMQConnection("amqp://guest:guest@client/test?brokerlist='tcp://localhost:1'"));
+ _handler.setNetworkConnection(new TestNetworkConnection());
AMQBody body = BasicRecoverOkBodyImpl.getFactory().newInstance(null, 1);
_blockFrame = new AMQFrame(0, body);
Copied: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java (from r1140000, qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java)
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java?p2=qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java&p1=qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java&r1=1140000&r2=1157780&rev=1157780&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandlerTest.java Mon Aug 15 11:26:46 2011
@@ -50,7 +50,7 @@ public class UsernameHashedPasswordCallb
{
super.setUp();
- final String url = "amqp://username:password@client/test?brokerlist='vm://:1'";
+ final String url = "amqp://username:password@client/test?brokerlist='tcp://localhost:1'";
_callbackHandler.initialise(new AMQConnectionURL(url));
}
Copied: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java (from r1140000, qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java)
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java?p2=qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java&p1=qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java&r1=1140000&r2=1157780&rev=1157780&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/client/security/UsernamePasswordCallbackHandlerTest.java Mon Aug 15 11:26:46 2011
@@ -46,7 +46,7 @@ public class UsernamePasswordCallbackHan
{
super.setUp();
- final String url = "amqp://username:password@client/test?brokerlist='vm://:1'";
+ final String url = "amqp://username:password@client/test?brokerlist='tcp://localhost:1'";
_callbackHandler.initialise(new AMQConnectionURL(url));
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/BrokerDetails/BrokerDetailsTest.java Mon Aug 15 11:26:46 2011
@@ -43,15 +43,6 @@ public class BrokerDetailsTest extends T
assertTrue(broker.getProperty("immediatedelivery").equals("true"));
}
- public void testVMBroker() throws URLSyntaxException
- {
- String url = "vm://:2";
-
- AMQBrokerDetails broker = new AMQBrokerDetails(url);
- assertTrue(broker.getTransport().equals("vm"));
- assertEquals(broker.getPort(), 2);
- }
-
public void testTransportsDefaultToTCP() throws URLSyntaxException
{
String url = "localhost:5672";
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java Mon Aug 15 11:26:46 2011
@@ -300,53 +300,6 @@ public class ConnectionURLTest extends T
assertTrue(connectionurl.getOption("immediatedelivery").equals("true"));
}
- public void testSinglevmURL() throws URLSyntaxException
- {
- String url = "amqp://guest:guest@/test?brokerlist='vm://:2'";
-
- ConnectionURL connectionurl = new AMQConnectionURL(url);
-
- assertTrue(connectionurl.getFailoverMethod() == null);
- assertTrue(connectionurl.getUsername().equals("guest"));
- assertTrue(connectionurl.getPassword().equals("guest"));
- assertTrue(connectionurl.getVirtualHost().equals("/test"));
-
- assertTrue(connectionurl.getBrokerCount() == 1);
-
- BrokerDetails service = connectionurl.getBrokerDetails(0);
-
- assertTrue(service.getTransport().equals("vm"));
- assertTrue(service.getHost().equals(""));
- assertTrue(service.getPort() == 2);
-
- }
-
- public void testFailoverVMURL() throws URLSyntaxException
- {
- String url = "amqp://ritchiem:bob@/test?brokerlist='vm://:2;vm://:3',failover='roundrobin'";
-
- ConnectionURL connectionurl = new AMQConnectionURL(url);
-
- assertTrue(connectionurl.getFailoverMethod().equals("roundrobin"));
- assertTrue(connectionurl.getUsername().equals("ritchiem"));
- assertTrue(connectionurl.getPassword().equals("bob"));
- assertTrue(connectionurl.getVirtualHost().equals("/test"));
-
- assertTrue(connectionurl.getBrokerCount() == 2);
-
- BrokerDetails service = connectionurl.getBrokerDetails(0);
-
- assertTrue(service.getTransport().equals("vm"));
- assertTrue(service.getHost().equals(""));
- assertTrue(service.getPort() == 2);
-
- service = connectionurl.getBrokerDetails(1);
- assertTrue(service.getTransport().equals("vm"));
- assertTrue(service.getHost().equals(""));
- assertTrue(service.getPort() == 3);
- }
-
-
public void testNoVirtualHostURL()
{
String url = "amqp://user@?brokerlist='tcp://localhost:5672'";
@@ -487,27 +440,6 @@ public class ConnectionURLTest extends T
}
- public void testSocketProtocol() throws URLSyntaxException
- {
- String url = "amqp://guest:guest@id/test" + "?brokerlist='socket://VM-Unique-socketID'";
-
- try
- {
- AMQConnectionURL curl = new AMQConnectionURL(url);
- assertNotNull(curl);
- assertEquals(1, curl.getBrokerCount());
- assertNotNull(curl.getBrokerDetails(0));
- assertEquals(BrokerDetails.SOCKET, curl.getBrokerDetails(0).getTransport());
- assertEquals("VM-Unique-socketID", curl.getBrokerDetails(0).getHost());
- assertEquals("URL does not toString as expected",
- url.replace(":guest", ":********"), curl.toString());
- }
- catch (URLSyntaxException e)
- {
- fail(e.getMessage());
- }
- }
-
public void testSingleTransportMultiOptionOnBrokerURL() throws URLSyntaxException
{
String url = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672?foo='jim'&bar='bob'&fred='jimmy'',routingkey='jim',timeout='200',immediatedelivery='true'";
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java Mon Aug 15 11:26:46 2011
@@ -195,4 +195,10 @@ public class TestAMQSession extends AMQS
{
return false;
}
+
+ @Override
+ public AMQException getLastException()
+ {
+ return null;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common.xml
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common.xml?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common.xml (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common.xml Mon Aug 15 11:26:46 2011
@@ -23,7 +23,9 @@
<dirname property="project.root" file="${ant.file.common}"/>
<property name="project.name" value="qpid"/>
- <property name="project.version" value="0.11"/>
+ <property name="project.version" value="0.13"/>
+ <property name="project.url" value="http://qpid.apache.org"/>
+ <property name="project.groupid" value="org.apache.qpid"/>
<property name="project.namever" value="${project.name}-${project.version}"/>
<property name="resources" location="${project.root}/resources"/>
@@ -76,6 +78,7 @@
</path>
<property name="maven.local.repo" value="${build.scratch}/maven-local-repo"/>
+ <property name="maven.settings.xml" value="${project.root}/maven-settings.xml"/>
<property name="maven.unique.version" value="false"/>
<property name="maven.snapshot" value="true"/>
<condition property="maven.version.suffix" value="" else="-SNAPSHOT">
@@ -113,9 +116,10 @@
<attribute name="dir"/>
<attribute name="name"/>
<attribute name="excludes" default=""/>
+ <attribute name="erroronmissingdir" default="true"/>
<sequential>
- <dirset id="@{dir}.refid" dir="@{dir}" excludes="@{excludes}">
+ <dirset id="@{dir}.refid" dir="@{dir}" excludes="@{excludes}" erroronmissingdir="@{erroronmissingdir}">
<present targetdir="@{dir}">
<mapper type="glob" from="*" to="*/build.xml" />
</present>
@@ -129,8 +133,6 @@
</sequential>
</macrodef>
-
-
<macrodef name="jython">
<attribute name="path"/>
<element name="args"/>
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/common.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/common.bnd?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/common.bnd (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/common.bnd Mon Aug 15 11:26:46 2011
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.11.0
+ver: 0.13.0
Bundle-SymbolicName: qpid-common
Bundle-Version: ${ver}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java Mon Aug 15 11:26:46 2011
@@ -54,6 +54,7 @@ public class AMQChannelException extends
public AMQFrame getCloseFrame(int channel)
{
MethodRegistry reg = MethodRegistry.getMethodRegistry(new ProtocolVersion(major,minor));
- return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() == null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(), new AMQShortString(getMessage()),_classId,_methodId));
+ return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() == null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(), getMessageAsShortString(),_classId,_methodId));
}
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java Mon Aug 15 11:26:46 2011
@@ -62,9 +62,10 @@ public class AMQConnectionException exte
MethodRegistry reg = MethodRegistry.getMethodRegistry(new ProtocolVersion(major,minor));
return new AMQFrame(0,
reg.createConnectionCloseBody(getErrorCode().getCode(),
- new AMQShortString(getMessage()),
+ getMessageAsShortString(),
_classId,
_methodId));
}
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java Mon Aug 15 11:26:46 2011
@@ -20,6 +20,7 @@
*/
package org.apache.qpid;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.protocol.AMQConstant;
/**
@@ -121,4 +122,19 @@ public class AMQException extends Except
return newAMQE;
}
+
+ /**
+ * Truncates the exception message to 255 characters if its length exceeds 255.
+ *
+ * @return exception message
+ */
+ public AMQShortString getMessageAsShortString()
+ {
+ String message = getMessage();
+ if (message != null && message.length() > AMQShortString.MAX_LENGTH)
+ {
+ message = message.substring(0, AMQShortString.MAX_LENGTH - 3) + "...";
+ }
+ return new AMQShortString(message);
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java Mon Aug 15 11:26:46 2011
@@ -23,7 +23,7 @@ package org.apache.qpid.configuration;
*/
public class ClientProperties
{
-
+
/**
* Currently with Qpid it is not possible to change the client ID.
* If one is not specified upon connection construction, an id is generated automatically.
@@ -68,67 +68,40 @@ public class ClientProperties
* by the broker in TuneOK it will be used as the heartbeat interval.
* If not a warning will be printed and the max value specified for
* heartbeat in TuneOK will be used
- *
+ *
* The default idle timeout is set to 120 secs
*/
public static final String IDLE_TIMEOUT_PROP_NAME = "idle_timeout";
public static final long DEFAULT_IDLE_TIMEOUT = 120000;
-
+
public static final String HEARTBEAT = "qpid.heartbeat";
public static final int HEARTBEAT_DEFAULT = 120;
-
+
/**
* This value will be used to determine the default destination syntax type.
* Currently the two types are Binding URL (java only) and the Addressing format (used by
- * all clients).
+ * all clients).
*/
public static final String DEST_SYNTAX = "qpid.dest_syntax";
-
+
public static final String USE_LEGACY_MAP_MESSAGE_FORMAT = "qpid.use_legacy_map_message";
- /**
- * ==========================================================
- * Those properties are used when the io size should be bounded
- * ==========================================================
- */
+ public static final String AMQP_VERSION = "qpid.amqp.version";
- /**
- * When set to true the io layer throttle down producers and consumers
- * when written or read messages are accumulating and exceeding a certain size.
- * This is especially useful when a the producer rate is greater than the network
- * speed.
- * type: boolean
- */
- public static final String PROTECTIO_PROP_NAME = "protectio";
+ public static final String QPID_VERIFY_CLIENT_ID = "qpid.verify_client_id";
- //=== The following properties are only used when the previous one is true.
- /**
- * Max size of read messages that can be stored within the MINA layer
- * type: int
- */
- public static final String READ_BUFFER_LIMIT_PROP_NAME = "qpid.read.buffer.limit";
- public static final String READ_BUFFER_LIMIT_DEFAULT = "262144";
- /**
- * Max size of written messages that can be stored within the MINA layer
- * type: int
- */
- public static final String WRITE_BUFFER_LIMIT_PROP_NAME = "qpid.read.buffer.limit";
- public static final String WRITE_BUFFER_LIMIT_DEFAULT = "262144";
-
- public static final String AMQP_VERSION = "qpid.amqp.version";
-
private static ClientProperties _instance = new ClientProperties();
-
+
/*
- public static final QpidProperty<Boolean> IGNORE_SET_CLIENTID_PROP_NAME =
+ public static final QpidProperty<Boolean> IGNORE_SET_CLIENTID_PROP_NAME =
QpidProperty.booleanProperty(false,"qpid.ignore_set_client_id","ignore_setclientID");
-
+
public static final QpidProperty<Boolean> SYNC_PERSISTENT_PROP_NAME =
QpidProperty.booleanProperty(false,"qpid.sync_persistence","sync_persistence");
-
-
+
+
public static final QpidProperty<Integer> MAX_PREFETCH_PROP_NAME =
QpidProperty.intProperty(500,"qpid.max_prefetch","max_prefetch"); */
-
-
+
+
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java Mon Aug 15 11:26:46 2011
@@ -37,6 +37,10 @@ import java.lang.ref.WeakReference;
*/
public final class AMQShortString implements CharSequence, Comparable<AMQShortString>
{
+ /**
+ * The maximum number of octets in AMQ short string as defined in AMQP specification
+ */
+ public static final int MAX_LENGTH = 255;
private static final byte MINUS = (byte)'-';
private static final byte ZERO = (byte) '0';
@@ -118,22 +122,19 @@ public final class AMQShortString implem
public AMQShortString(byte[] data)
{
-
+ if (data == null)
+ {
+ throw new NullPointerException("Cannot create AMQShortString with null data[]");
+ }
+ if (data.length > MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
_data = data.clone();
_length = data.length;
_offset = 0;
}
- public AMQShortString(byte[] data, int pos)
- {
- final int size = data[pos++];
- final byte[] dataCopy = new byte[size];
- System.arraycopy(data,pos,dataCopy,0,size);
- _length = size;
- _data = dataCopy;
- _offset = 0;
- }
-
public AMQShortString(String data)
{
this((data == null) ? EMPTY_CHAR_ARRAY : data.toCharArray());
@@ -146,7 +147,12 @@ public final class AMQShortString implem
{
throw new NullPointerException("Cannot create AMQShortString with null char[]");
}
-
+ // the current implementation of 0.8/0.9.x short string encoding
+ // supports only ASCII characters
+ if (data.length> MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
final int length = data.length;
final byte[] stringBytes = new byte[length];
int hash = 0;
@@ -165,6 +171,17 @@ public final class AMQShortString implem
public AMQShortString(CharSequence charSequence)
{
+ if (charSequence == null)
+ {
+ // it should be possible to create short string for null data
+ charSequence = "";
+ }
+ // the current implementation of 0.8/0.9.x short string encoding
+ // supports only ASCII characters
+ if (charSequence.length() > MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
final int length = charSequence.length();
final byte[] stringBytes = new byte[length];
int hash = 0;
@@ -184,6 +201,10 @@ public final class AMQShortString implem
private AMQShortString(ByteBuffer data, final int length)
{
+ if (length > MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
if(data.isDirect() || data.isReadOnly())
{
byte[] dataBytes = new byte[length];
@@ -205,8 +226,17 @@ public final class AMQShortString implem
private AMQShortString(final byte[] data, final int from, final int to)
{
+ if (data == null)
+ {
+ throw new NullPointerException("Cannot create AMQShortString with null data[]");
+ }
+ int length = to - from;
+ if (length > MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
_offset = from;
- _length = to - from;
+ _length = length;
_data = data;
}
@@ -245,29 +275,6 @@ public final class AMQShortString implem
return new CharSubSequence(start, end);
}
- public int writeToByteArray(byte[] encoding, int pos)
- {
- final int size = length();
- encoding[pos++] = (byte) size;
- System.arraycopy(_data,_offset,encoding,pos,size);
- return pos+size;
- }
-
- public static AMQShortString readFromByteArray(byte[] byteEncodedDestination, int pos)
- {
-
-
- final AMQShortString shortString = new AMQShortString(byteEncodedDestination, pos);
- if(shortString.length() == 0)
- {
- return null;
- }
- else
- {
- return shortString;
- }
- }
-
public static AMQShortString readFromBuffer(ByteBuffer buffer)
{
final short length = buffer.getUnsigned();
@@ -690,6 +697,10 @@ public final class AMQShortString implem
size += term.length();
}
+ if (size > MAX_LENGTH)
+ {
+ throw new IllegalArgumentException("Cannot create AMQShortString with number of octets over 255!");
+ }
byte[] data = new byte[size];
int pos = 0;
final byte[] delimData = delim._data;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java Mon Aug 15 11:26:46 2011
@@ -22,8 +22,6 @@ package org.apache.qpid.protocol;
import java.net.SocketAddress;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.transport.NetworkDriver;
import org.apache.qpid.transport.Receiver;
/**
@@ -32,9 +30,6 @@ import org.apache.qpid.transport.Receive
*/
public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>
{
- // Sets the network driver providing data for this ProtocolEngine
- void setNetworkDriver (NetworkDriver driver);
-
// Returns the remote address of the NetworkDriver
SocketAddress getRemoteAddress();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngineFactory.java Mon Aug 15 11:26:46 2011
@@ -20,12 +20,12 @@
*/
package org.apache.qpid.protocol;
-import org.apache.qpid.transport.NetworkDriver;
+import org.apache.qpid.transport.network.NetworkConnection;
public interface ProtocolEngineFactory
{
// Returns a new instance of a ProtocolEngine
- ProtocolEngine newProtocolEngine(NetworkDriver networkDriver);
+ ProtocolEngine newProtocolEngine(NetworkConnection network);
}
\ No newline at end of file
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/thread/QpidThreadExecutor.java Mon Aug 15 11:26:46 2011
@@ -23,7 +23,7 @@ package org.apache.qpid.thread;
import org.apache.qpid.thread.Threading;
-import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+import java.util.concurrent.Executor;
public class QpidThreadExecutor implements Executor
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java Mon Aug 15 11:26:46 2011
@@ -247,7 +247,7 @@ public class ClientDelegate extends Conn
int i = heartbeat;
if (i == 0)
{
- log.warn("Idle timeout is zero. Heartbeats are disabled");
+ log.info("Idle timeout is 0 sec. Heartbeats are disabled.");
return 0; // heartbeats are disabled.
}
else if (i >= min && i <= max)
@@ -256,8 +256,8 @@ public class ClientDelegate extends Conn
}
else
{
- log.warn("Ignoring the idle timeout %s set by the connection," +
- " using the brokers max value %s", i,max);
+ log.info("The broker does not support the configured connection idle timeout of %s sec," +
+ " using the brokers max supported value of %s sec instead.", i,max);
return max;
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Mon Aug 15 11:26:46 2011
@@ -25,8 +25,8 @@ import static org.apache.qpid.transport.
import static org.apache.qpid.transport.Connection.State.NEW;
import static org.apache.qpid.transport.Connection.State.OPEN;
import static org.apache.qpid.transport.Connection.State.OPENING;
-import static org.apache.qpid.transport.Connection.State.RESUMING;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -40,6 +40,13 @@ import java.util.concurrent.atomic.Atomi
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslServer;
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.transport.network.Assembler;
+import org.apache.qpid.transport.network.Disassembler;
+import org.apache.qpid.transport.network.InputHandler;
+import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.OutgoingNetworkTransport;
+import org.apache.qpid.transport.network.Transport;
import org.apache.qpid.transport.network.security.SecurityLayer;
import org.apache.qpid.transport.util.Logger;
import org.apache.qpid.transport.util.Waiter;
@@ -113,15 +120,12 @@ public class Connection extends Connecti
private SaslServer saslServer;
private SaslClient saslClient;
private int idleTimeout = 0;
- private String _authorizationID;
private Map<String,Object> _serverProperties;
private String userID;
private ConnectionSettings conSettings;
private SecurityLayer securityLayer;
private String _clientId;
- private static final AtomicLong idGenerator = new AtomicLong(0);
- private final long _connectionId = idGenerator.incrementAndGet();
private final AtomicBoolean connectionLost = new AtomicBoolean(false);
public Connection() {}
@@ -235,13 +239,14 @@ public class Connection extends Connecti
state = OPENING;
userID = settings.getUsername();
delegate = new ClientDelegate(settings);
-
- TransportBuilder transport = new TransportBuilder();
- transport.init(this);
- this.sender = transport.buildSenderPipe();
- transport.buildReceiverPipe(this);
- this.securityLayer = transport.getSecurityLayer();
-
+
+ securityLayer = new SecurityLayer(this);
+
+ OutgoingNetworkTransport transport = Transport.getOutgoingTransportInstance(ProtocolVersion.v0_10);
+ Receiver<ByteBuffer> receiver = securityLayer.receiver(new InputHandler(new Assembler(this)));
+ NetworkConnection network = transport.connect(settings, receiver, null);
+ sender = new Disassembler(securityLayer.sender(network.getSender()), settings.getMaxFrameSize());
+
send(new ProtocolHeader(1, 0, 10));
Waiter w = new Waiter(lock, timeout);
@@ -353,11 +358,6 @@ public class Connection extends Connecti
_sessionFactory = sessionFactory;
}
- public long getConnectionId()
- {
- return _connectionId;
- }
-
public ConnectionDelegate getConnectionDelegate()
{
return delegate;
@@ -467,11 +467,12 @@ public class Connection extends Connecti
{
synchronized (lock)
{
+ List <Binary> transactedSessions = new ArrayList();
for (Session ssn : sessions.values())
{
if (ssn.isTransacted())
- {
- removeSession(ssn);
+ {
+ transactedSessions.add(ssn.getName());
ssn.setState(Session.State.CLOSED);
}
else
@@ -481,6 +482,11 @@ public class Connection extends Connecti
ssn.resume();
}
}
+
+ for (Binary ssn_name : transactedSessions)
+ {
+ sessions.remove(ssn_name);
+ }
setState(OPEN);
}
}
@@ -646,16 +652,6 @@ public class Connection extends Connecti
return idleTimeout;
}
- public void setAuthorizationID(String authorizationID)
- {
- _authorizationID = authorizationID;
- }
-
- public String getAuthorizationID()
- {
- return _authorizationID;
- }
-
public String getUserID()
{
return userID;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionDelegate.java Mon Aug 15 11:26:46 2011
@@ -95,6 +95,7 @@ public abstract class ConnectionDelegate
Session ssn = conn.getSession(dtc.getChannel());
if (ssn != null)
{
+ ssn.setDetachCode(dtc.getCode());
conn.unmap(ssn);
ssn.closed();
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionSettings.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionSettings.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionSettings.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionSettings.java Mon Aug 15 11:26:46 2011
@@ -30,6 +30,8 @@ import java.util.Map;
*/
public class ConnectionSettings
{
+ public static final String WILDCARD_ADDRESS = "*";
+
String protocol = "tcp";
String host = "localhost";
String vhost;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java Mon Aug 15 11:26:46 2011
@@ -75,10 +75,7 @@ public class ServerDelegate extends Conn
if (mechanism == null || mechanism.length() == 0)
{
- conn.connectionTune
- (getChannelMax(),
- org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
- 0, getHeartbeatMax());
+ tuneAuthorizedConnection(conn);
return;
}
@@ -97,8 +94,7 @@ public class ServerDelegate extends Conn
}
catch (SaslException e)
{
- conn.exception(e);
- conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+ connectionAuthFailed(conn, e);
}
}
@@ -109,33 +105,52 @@ public class ServerDelegate extends Conn
return ss;
}
- private void secure(Connection conn, byte[] response)
+ protected void secure(final SaslServer ss, final Connection conn, final byte[] response)
{
- SaslServer ss = conn.getSaslServer();
try
{
byte[] challenge = ss.evaluateResponse(response);
if (ss.isComplete())
{
ss.dispose();
- conn.connectionTune
- (getChannelMax(),
- org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
- 0, getHeartbeatMax());
- conn.setAuthorizationID(ss.getAuthorizationID());
+ tuneAuthorizedConnection(conn);
}
else
{
- conn.connectionSecure(challenge);
+ connectionAuthContinue(conn, challenge);
}
}
catch (SaslException e)
{
- conn.exception(e);
- conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+ connectionAuthFailed(conn, e);
}
}
+ protected void connectionAuthFailed(final Connection conn, Exception e)
+ {
+ conn.exception(e);
+ conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+ }
+
+ protected void connectionAuthContinue(final Connection conn, byte[] challenge)
+ {
+ conn.connectionSecure(challenge);
+ }
+
+ protected void tuneAuthorizedConnection(final Connection conn)
+ {
+ conn.connectionTune
+ (getChannelMax(),
+ org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
+ 0, getHeartbeatMax());
+ }
+
+ protected void secure(final Connection conn, final byte[] response)
+ {
+ final SaslServer ss = conn.getSaslServer();
+ secure(ss, conn, response);
+ }
+
protected int getHeartbeatMax()
{
return 0xFFFF;
@@ -155,22 +170,7 @@ public class ServerDelegate extends Conn
@Override
public void connectionTuneOk(Connection conn, ConnectionTuneOk ok)
{
- int okChannelMax = ok.getChannelMax();
-
- if (okChannelMax > getChannelMax())
- {
- _logger.error("Connection '" + conn.getConnectionId() + "' being severed, " +
- "client connectionTuneOk returned a channelMax (" + okChannelMax +
- ") above the servers offered limit (" + getChannelMax() +")");
- //Due to the error we must forcefully close the connection without negotiation
- conn.getSender().close();
- return;
- }
-
- //0 means no implied limit, except available server resources
- //(or that forced by protocol limitations [0xFFFF])
- conn.setChannelMax(okChannelMax == 0 ? Connection.MAX_CHANNEL_MAX : okChannelMax);
}
@Override
@@ -200,4 +200,11 @@ public class ServerDelegate extends Conn
ssn.sessionAttached(atc.getName());
ssn.setState(Session.State.OPEN);
}
+
+ protected void setConnectionTuneOkChannelMax(final Connection conn, final int okChannelMax)
+ {
+ //0 means no implied limit, except available server resources
+ //(or that forced by protocol limitations [0xFFFF])
+ conn.setChannelMax(okChannelMax == 0 ? Connection.MAX_CHANNEL_MAX : okChannelMax);
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java Mon Aug 15 11:26:46 2011
@@ -42,7 +42,10 @@ import static org.apache.qpid.util.Seria
import static org.apache.qpid.util.Strings.toUTF8;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -117,7 +120,9 @@ public class Session extends SessionInvo
private Thread resumer = null;
private boolean transacted = false;
-
+ private SessionDetachCode detachCode;
+ private final Object stateLock = new Object();
+
protected Session(Connection connection, Binary name, long expiry)
{
this(connection, new SessionDelegate(), name, expiry);
@@ -262,7 +267,32 @@ public class Session extends SessionInvo
}
else if (m instanceof MessageTransfer)
{
- ((MessageTransfer)m).getHeader().get(DeliveryProperties.class).setRedelivered(true);
+ MessageTransfer xfr = (MessageTransfer)m;
+
+ if (xfr.getHeader() != null)
+ {
+ if (xfr.getHeader().get(DeliveryProperties.class) != null)
+ {
+ xfr.getHeader().get(DeliveryProperties.class).setRedelivered(true);
+ }
+ else
+ {
+ Struct[] structs = xfr.getHeader().getStructs();
+ DeliveryProperties deliveryProps = new DeliveryProperties();
+ deliveryProps.setRedelivered(true);
+
+ List<Struct> list = Arrays.asList(structs);
+ list.add(deliveryProps);
+ xfr.setHeader(new Header(list));
+ }
+
+ }
+ else
+ {
+ DeliveryProperties deliveryProps = new DeliveryProperties();
+ deliveryProps.setRedelivered(true);
+ xfr.setHeader(new Header(deliveryProps));
+ }
}
sessionCommandPoint(m.getId(), 0);
send(m);
@@ -1003,7 +1033,8 @@ public class Session extends SessionInvo
if(state == CLOSED)
{
- connection.removeSession(this);
+ connection.removeSession(this);
+ listener.closed(this);
}
}
@@ -1016,13 +1047,54 @@ public class Session extends SessionInvo
{
return String.format("ssn:%s", name);
}
-
+
public void setTransacted(boolean b) {
this.transacted = b;
}
-
+
public boolean isTransacted(){
return transacted;
}
-
+
+ public void setDetachCode(SessionDetachCode dtc)
+ {
+ this.detachCode = dtc;
+ }
+
+ public SessionDetachCode getDetachCode()
+ {
+ return this.detachCode;
+ }
+
+ public void awaitOpen()
+ {
+ switch (state)
+ {
+ case NEW:
+ synchronized(stateLock)
+ {
+ Waiter w = new Waiter(stateLock, timeout);
+ while (w.hasTime() && state == NEW)
+ {
+ w.await();
+ }
+ }
+
+ if (state != OPEN)
+ {
+ throw new SessionException("Timed out waiting for Session to open");
+ }
+ case DETACHED:
+ case CLOSING:
+ case CLOSED:
+ throw new SessionException("Session closed");
+ default :
+ break;
+ }
+ }
+
+ public Object getStateLock()
+ {
+ return stateLock;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionDelegate.java Mon Aug 15 11:26:46 2011
@@ -76,6 +76,10 @@ public class SessionDelegate
@Override public void sessionAttached(Session ssn, SessionAttached atc)
{
ssn.setState(Session.State.OPEN);
+ synchronized (ssn.getStateLock())
+ {
+ ssn.getStateLock().notifyAll();
+ }
}
@Override public void sessionTimeout(Session ssn, SessionTimeout t)
@@ -202,11 +206,19 @@ public class SessionDelegate
public void closed(Session session)
{
- log.warn("CLOSED: [%s]", session);
+ log.debug("CLOSED: [%s]", session);
+ synchronized (session.getStateLock())
+ {
+ session.getStateLock().notifyAll();
+ }
}
public void detached(Session session)
{
- log.warn("DETACHED: [%s]", session);
+ log.debug("DETACHED: [%s]", session);
+ synchronized (session.getStateLock())
+ {
+ session.getStateLock().notifyAll();
+ }
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkTransport.java Mon Aug 15 11:26:46 2011
@@ -20,19 +20,13 @@
*/
package org.apache.qpid.transport.network;
-import java.nio.ByteBuffer;
-
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.Sender;
-import org.apache.qpid.transport.ConnectionSettings;
-
+/**
+ * A network transport is responsible for the establishment of network connections.
+ * NetworkTransport implementations are pluggable via the {@link Transport} class.
+ */
public interface NetworkTransport
{
- public void init(ConnectionSettings settings);
-
- public Sender<ByteBuffer> sender();
-
- public void receiver(Receiver<ByteBuffer> delegate);
-
public void close();
-}
\ No newline at end of file
+
+ public NetworkConnection getConnection();
+}
Copied: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java (from r1145000, qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java)
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java?p2=qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java&p1=qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java&r1=1145000&r2=1157780&rev=1157780&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/Transport.java Mon Aug 15 11:26:46 2011
@@ -20,7 +20,116 @@
*/
package org.apache.qpid.transport.network;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.transport.TransportException;
+
public class Transport
{
+ public static final String QPID_TRANSPORT_PROPNAME = "qpid.transport";
+ public static final String QPID_TRANSPORT_V0_8_PROPNAME = "qpid.transport.v0_8";
+ public static final String QPID_TRANSPORT_V0_9_PROPNAME = "qpid.transport.v0_9";
+ public static final String QPID_TRANSPORT_V0_9_1_PROPNAME = "qpid.transport.v0_9_1";
+ public static final String QPID_TRANSPORT_V0_10_PROPNAME = "qpid.transport.v0_10";
+ public static final String QPID_BROKER_TRANSPORT_PROPNAME = "qpid.broker.transport";
+
+ // Can't reference the class directly here, as this would preclude the ability to bundle transports separately.
+ private static final String MINA_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.mina.MinaNetworkTransport";
+ private static final String IO_TRANSPORT_CLASSNAME = "org.apache.qpid.transport.network.io.IoNetworkTransport";
+
public static final String TCP = "tcp";
+
+ private final static Map<ProtocolVersion,String> OUTGOING_PROTOCOL_TO_IMPLDEFAULTS_MAP;
+
+ static
+ {
+ final Map<ProtocolVersion,String> map = new HashMap<ProtocolVersion, String>();
+ map.put(ProtocolVersion.v8_0, MINA_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v0_9, MINA_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v0_91, MINA_TRANSPORT_CLASSNAME);
+ map.put(ProtocolVersion.v0_10, IO_TRANSPORT_CLASSNAME);
+
+ OUTGOING_PROTOCOL_TO_IMPLDEFAULTS_MAP = Collections.unmodifiableMap(map);
+ }
+
+ public static IncomingNetworkTransport getIncomingTransportInstance()
+ {
+ return (IncomingNetworkTransport) loadTransportClass(
+ System.getProperty(QPID_BROKER_TRANSPORT_PROPNAME, MINA_TRANSPORT_CLASSNAME));
+ }
+
+ public static OutgoingNetworkTransport getOutgoingTransportInstance(
+ final ProtocolVersion protocolVersion)
+ {
+
+ final String overrride = getOverrideClassNameFromSystemProperty(protocolVersion);
+ final String networkTransportClassName;
+ if (overrride != null)
+ {
+ networkTransportClassName = overrride;
+ }
+ else
+ {
+ networkTransportClassName = OUTGOING_PROTOCOL_TO_IMPLDEFAULTS_MAP.get(protocolVersion);
+ }
+
+ return (OutgoingNetworkTransport) loadTransportClass(networkTransportClassName);
+ }
+
+ private static NetworkTransport loadTransportClass(final String networkTransportClassName)
+ {
+ if (networkTransportClassName == null)
+ {
+ throw new IllegalArgumentException("transport class name must not be null");
+ }
+
+ try
+ {
+ final Class<?> clazz = Class.forName(networkTransportClassName);
+ return (NetworkTransport) clazz.newInstance();
+ }
+ catch (InstantiationException e)
+ {
+ throw new TransportException("Unable to instantiate transport class " + networkTransportClassName, e);
+ }
+ catch (IllegalAccessException e)
+ {
+ throw new TransportException("Access exception " + networkTransportClassName, e);
+ }
+ catch (ClassNotFoundException e)
+ {
+ throw new TransportException("Unable to load transport class " + networkTransportClassName, e);
+ }
+ }
+
+ private static String getOverrideClassNameFromSystemProperty(final ProtocolVersion protocolVersion)
+ {
+ final String protocolSpecificSystemProperty;
+
+ if (ProtocolVersion.v0_10.equals(protocolVersion))
+ {
+ protocolSpecificSystemProperty = QPID_TRANSPORT_V0_10_PROPNAME;
+ }
+ else if (ProtocolVersion.v0_91.equals(protocolVersion))
+ {
+ protocolSpecificSystemProperty = QPID_TRANSPORT_V0_9_1_PROPNAME;
+ }
+ else if (ProtocolVersion.v0_9.equals(protocolVersion))
+ {
+ protocolSpecificSystemProperty = QPID_TRANSPORT_V0_9_PROPNAME;
+ }
+ else if (ProtocolVersion.v8_0.equals(protocolVersion))
+ {
+ protocolSpecificSystemProperty = QPID_TRANSPORT_V0_8_PROPNAME;
+ }
+ else
+ {
+ throw new IllegalArgumentException("Unknown ProtocolVersion " + protocolVersion);
+ }
+
+ return System.getProperty(protocolSpecificSystemProperty, System.getProperty(QPID_TRANSPORT_PROPNAME));
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java Mon Aug 15 11:26:46 2011
@@ -27,14 +27,15 @@ import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
+import org.apache.qpid.ssl.SSLContextFactory;
import org.apache.qpid.transport.ConnectionSettings;
import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.NetworkTransport;
+import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.OutgoingNetworkTransport;
import org.apache.qpid.transport.util.Logger;
-public class IoNetworkTransport implements NetworkTransport, IoContext
+public class IoNetworkTransport implements OutgoingNetworkTransport
{
static
{
@@ -44,34 +45,31 @@ public class IoNetworkTransport implemen
(Boolean.getBoolean("amqj.enableDirectBuffers"));
}
- private static final Logger log = Logger.get(IoNetworkTransport.class);
+ private static final Logger LOGGER = Logger.get(IoNetworkTransport.class);
- private Socket socket;
- private Sender<ByteBuffer> sender;
- private IoReceiver receiver;
- private long timeout = 60000;
- private ConnectionSettings settings;
+ private Socket _socket;
+ private IoNetworkConnection _connection;
+ private long _timeout = 60000;
- public void init(ConnectionSettings settings)
+ public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContextFactory sslFactory)
{
+ int sendBufferSize = settings.getWriteBufferSize();
+ int receiveBufferSize = settings.getReadBufferSize();
+
try
{
- this.settings = settings;
- InetAddress address = InetAddress.getByName(settings.getHost());
- socket = new Socket();
- socket.setReuseAddress(true);
- socket.setTcpNoDelay(settings.isTcpNodelay());
-
- log.debug("default-SO_RCVBUF : %s", socket.getReceiveBufferSize());
- log.debug("default-SO_SNDBUF : %s", socket.getSendBufferSize());
+ _socket = new Socket();
+ _socket.setReuseAddress(true);
+ _socket.setTcpNoDelay(settings.isTcpNodelay());
+ _socket.setSendBufferSize(sendBufferSize);
+ _socket.setReceiveBufferSize(receiveBufferSize);
- socket.setSendBufferSize(settings.getWriteBufferSize());
- socket.setReceiveBufferSize(settings.getReadBufferSize());
+ LOGGER.debug("SO_RCVBUF : %s", _socket.getReceiveBufferSize());
+ LOGGER.debug("SO_SNDBUF : %s", _socket.getSendBufferSize());
- log.debug("new-SO_RCVBUF : %s", socket.getReceiveBufferSize());
- log.debug("new-SO_SNDBUF : %s", socket.getSendBufferSize());
+ InetAddress address = InetAddress.getByName(settings.getHost());
- socket.connect(new InetSocketAddress(address, settings.getPort()));
+ _socket.connect(new InetSocketAddress(address, settings.getPort()));
}
catch (SocketException e)
{
@@ -81,36 +79,35 @@ public class IoNetworkTransport implemen
{
throw new TransportException("Error connecting to broker", e);
}
- }
- public void receiver(Receiver<ByteBuffer> delegate)
- {
- receiver = new IoReceiver(this, delegate,
- 2*settings.getReadBufferSize() , timeout);
- }
-
- public Sender<ByteBuffer> sender()
- {
- return new IoSender(this, 2*settings.getWriteBufferSize(), timeout);
- }
+ try
+ {
+ _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, _timeout);
+ }
+ catch(Exception e)
+ {
+ try
+ {
+ _socket.close();
+ }
+ catch(IOException ioe)
+ {
+ //ignored, throw based on original exception
+ }
- public void close()
- {
-
- }
+ throw new TransportException("Error creating network connection", e);
+ }
- public Sender<ByteBuffer> getSender()
- {
- return sender;
+ return _connection;
}
- public IoReceiver getReceiver()
+ public void close()
{
- return receiver;
+ _connection.close();
}
- public Socket getSocket()
+ public NetworkConnection getConnection()
{
- return socket;
+ return _connection;
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java Mon Aug 15 11:26:46 2011
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.transport.network.io;
+import org.apache.qpid.common.Closeable;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
@@ -37,43 +38,54 @@ import java.util.concurrent.atomic.Atomi
*
*/
-final class IoReceiver implements Runnable
+final class IoReceiver implements Runnable, Closeable
{
private static final Logger log = Logger.get(IoReceiver.class);
- private final IoContext ioCtx;
private final Receiver<ByteBuffer> receiver;
private final int bufferSize;
private final Socket socket;
private final long timeout;
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Thread receiverThread;
- private final boolean shutdownBroken =
- ((String) System.getProperties().get("os.name")).matches("(?i).*windows.*");
+ private static final boolean shutdownBroken;
+ static
+ {
+ String osName = System.getProperty("os.name");
+ shutdownBroken = osName == null ? false : osName.matches("(?i).*windows.*");
+ }
- public IoReceiver(IoContext ioCtx, Receiver<ByteBuffer> receiver,
- int bufferSize, long timeout)
+ public IoReceiver(Socket socket, Receiver<ByteBuffer> receiver, int bufferSize, long timeout)
{
- this.ioCtx = ioCtx;
this.receiver = receiver;
this.bufferSize = bufferSize;
- this.socket = ioCtx.getSocket();
+ this.socket = socket;
this.timeout = timeout;
try
{
+ //Create but deliberately don't start the thread.
receiverThread = Threading.getThreadFactory().createThread(this);
}
catch(Exception e)
{
- throw new Error("Error creating IOReceiver thread",e);
+ throw new RuntimeException("Error creating IOReceiver thread",e);
}
receiverThread.setDaemon(true);
receiverThread.setName(String.format("IoReceiver - %s", socket.getRemoteSocketAddress()));
+ }
+
+ public void initiate()
+ {
receiverThread.start();
}
+ public void close()
+ {
+ close(false);
+ }
+
void close(boolean block)
{
if (!closed.getAndSet(true))
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java Mon Aug 15 11:26:46 2011
@@ -24,8 +24,11 @@ import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.qpid.common.Closeable;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.SenderException;
@@ -43,7 +46,6 @@ public final class IoSender implements R
// we can test other cases as well
private final static int START = Integer.MAX_VALUE - 10;
- private final IoContext ioCtx;
private final long timeout;
private final Socket socket;
private final OutputStream out;
@@ -56,14 +58,13 @@ public final class IoSender implements R
private final Object notEmpty = new Object();
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Thread senderThread;
+ private final List<Closeable> _listeners = new ArrayList<Closeable>();
private volatile Throwable exception = null;
-
- public IoSender(IoContext ioCtx, int bufferSize, long timeout)
+ public IoSender(Socket socket, int bufferSize, long timeout)
{
- this.ioCtx = ioCtx;
- this.socket = ioCtx.getSocket();
+ this.socket = socket;
this.buffer = new byte[pof2(bufferSize)]; // buffer size must be a power of 2
this.timeout = timeout;
@@ -78,6 +79,7 @@ public final class IoSender implements R
try
{
+ //Create but deliberately don't start the thread.
senderThread = Threading.getThreadFactory().createThread(this);
}
catch(Exception e)
@@ -87,6 +89,10 @@ public final class IoSender implements R
senderThread.setDaemon(true);
senderThread.setName(String.format("IoSender - %s", socket.getRemoteSocketAddress()));
+ }
+
+ public void initiate()
+ {
senderThread.start();
}
@@ -204,16 +210,20 @@ public final class IoSender implements R
senderThread.join(timeout);
if (senderThread.isAlive())
{
+ log.error("join timed out");
throw new SenderException("join timed out");
}
}
- ioCtx.getReceiver().close(false);
}
catch (InterruptedException e)
{
+ log.error("interrupted whilst waiting for sender thread to stop");
throw new SenderException(e);
}
-
+ finally
+ {
+ closeListeners();
+ }
if (reportException && exception != null)
{
throw new SenderException(exception);
@@ -221,6 +231,28 @@ public final class IoSender implements R
}
}
+ private void closeListeners()
+ {
+ Exception ex = null;
+ for(Closeable listener : _listeners)
+ {
+ try
+ {
+ listener.close();
+ }
+ catch(Exception e)
+ {
+ log.error("Exception closing listener: " + e.getMessage());
+ ex = e;
+ }
+ }
+
+ if (ex != null)
+ {
+ throw new SenderException(ex.getMessage(), ex);
+ }
+ }
+
public void run()
{
final int size = buffer.length;
@@ -304,4 +336,9 @@ public final class IoSender implements R
throw new SenderException(e);
}
}
+
+ public void registerCloseListener(Closeable listener)
+ {
+ _listeners.add(listener);
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayer.java Mon Aug 15 11:26:46 2011
@@ -43,8 +43,8 @@ public class SecurityLayer
Connection con;
SSLSecurityLayer sslLayer;
SASLSecurityLayer saslLayer;
-
- public void init(Connection con) throws TransportException
+
+ public SecurityLayer(Connection con)
{
this.con = con;
this.settings = con.getConnectionSettings();
@@ -55,10 +55,9 @@ public class SecurityLayer
if (settings.isUseSASLEncryption())
{
saslLayer = new SASLSecurityLayer();
- }
-
+ }
}
-
+
public Sender<ByteBuffer> sender(Sender<ByteBuffer> delegate)
{
Sender<ByteBuffer> sender = delegate;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java Mon Aug 15 11:26:46 2011
@@ -43,8 +43,7 @@ public class SASLSender extends SASLEncr
this.delegate = delegate;
log.debug("SASL Sender enabled");
}
-
- @Override
+
public void close()
{
@@ -65,13 +64,11 @@ public class SASLSender extends SASLEncr
}
}
- @Override
public void flush()
{
delegate.flush();
}
- @Override
public void send(ByteBuffer buf)
{
if (closed.get())
@@ -108,7 +105,6 @@ public class SASLSender extends SASLEncr
}
}
- @Override
public void setIdleTimeout(int i)
{
delegate.setIdleTimeout(i);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/QpidClientX509KeyManager.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/QpidClientX509KeyManager.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/QpidClientX509KeyManager.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/QpidClientX509KeyManager.java Mon Aug 15 11:26:46 2011
@@ -48,51 +48,45 @@ public class QpidClientX509KeyManager ex
kmf.init(ks, keyStorePassword.toCharArray());
this.delegate = (X509ExtendedKeyManager)kmf.getKeyManagers()[0];
}
-
- @Override
+
public String chooseClientAlias(String[] keyType, Principal[] issuers, Socket socket)
{
log.debug("chooseClientAlias:Returning alias " + alias);
return alias;
}
- @Override
public String chooseServerAlias(String keyType, Principal[] issuers, Socket socket)
{
return delegate.chooseServerAlias(keyType, issuers, socket);
}
- @Override
public X509Certificate[] getCertificateChain(String alias)
{
return delegate.getCertificateChain(alias);
}
- @Override
public String[] getClientAliases(String keyType, Principal[] issuers)
{
log.debug("getClientAliases:Returning alias " + alias);
return new String[]{alias};
}
- @Override
public PrivateKey getPrivateKey(String alias)
{
return delegate.getPrivateKey(alias);
}
- @Override
public String[] getServerAliases(String keyType, Principal[] issuers)
{
return delegate.getServerAliases(keyType, issuers);
}
-
+
public String chooseEngineClientAlias(String[] keyType, Principal[] issuers, SSLEngine engine)
{
log.debug("chooseEngineClientAlias:Returning alias " + alias);
return alias;
}
-
+
public String chooseEngineServerAlias(String keyType, Principal[] issuers, SSLEngine engine)
{
return delegate.chooseEngineServerAlias(keyType, issuers, engine);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/url/URLHelper.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/url/URLHelper.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/url/URLHelper.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/url/URLHelper.java Mon Aug 15 11:26:46 2011
@@ -31,9 +31,6 @@ public class URLHelper
public static void parseOptions(Map<String, String> optionMap, String options) throws URLSyntaxException
{
- // options looks like this
- // brokerlist='tcp://host:port?option='value',option='value';vm://:3/virtualpath?option='value'',failover='method?option='value',option='value''
-
if ((options == null) || (options.indexOf('=') == -1))
{
return;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java Mon Aug 15 11:26:46 2011
@@ -143,8 +143,9 @@ public class FileUtils
}
/**
- * Either opens the specified filename as an input stream, or uses the default resource loaded using the
- * specified class loader, if opening the file fails or no file name is specified.
+ * Either opens the specified filename as an input stream or either the filesystem or classpath,
+ * or uses the default resource loaded using the specified class loader, if opening the file fails
+ * or no file name is specified.
*
* @param filename The name of the file to open.
* @param defaultResource The name of the default resource on the classpath if the file cannot be opened.
@@ -156,28 +157,28 @@ public class FileUtils
{
InputStream is = null;
- // Flag to indicate whether the default resource should be used. By default this is true, so that the default
- // is used when opening the file fails.
- boolean useDefault = true;
-
// Try to open the file if one was specified.
if (filename != null)
{
+ // try on filesystem
try
{
is = new BufferedInputStream(new FileInputStream(new File(filename)));
-
- // Clear the default flag because the file was succesfully opened.
- useDefault = false;
}
catch (FileNotFoundException e)
{
- // Ignore this exception, the default will be used instead.
+ is = null;
+ }
+
+ if (is == null)
+ {
+ // failed on filesystem, so try on classpath
+ is = cl.getResourceAsStream(filename);
}
}
// Load the default resource if a file was not specified, or if opening the file failed.
- if (useDefault)
+ if (is == null)
{
is = cl.getResourceAsStream(defaultResource);
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java?rev=1157780&r1=1157779&r2=1157780&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java Mon Aug 15 11:26:46 2011
@@ -23,6 +23,7 @@ package org.apache.qpid;
import junit.framework.TestCase;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.qpid.framing.AMQShortString;
/**
* This test is to ensure that when an AMQException is rethrown that the specified exception is correctly wrapped up.
@@ -91,6 +92,18 @@ public class AMQExceptionTest extends Te
return amqe;
}
+ public void testGetMessageAsString()
+ {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < 25; i++)
+ {
+ sb.append("message [" + i + "]");
+ }
+ AMQException e = new AMQException(AMQConstant.INTERNAL_ERROR, sb.toString(), null);
+ AMQShortString message = e.getMessageAsShortString();
+ assertEquals(sb.substring(0, AMQShortString.MAX_LENGTH - 3) + "...", message.toString());
+ }
+
/**
* Private class that extends AMQException but does not have a default exception.
*/
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org