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 11:43:21 UTC
svn commit: r1157750 [3/5] - in
/qpid/branches/rg-amqp-1-0-sandbox/qpid/java: ./
broker-plugins/experimental/shutdown/src/main/java/ broker/ broker/bin/
broker/src/main/java/org/apache/qpid/qmf/
broker/src/main/java/org/apache/qpid/server/ broker/src/m...
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java Mon Aug 15 09:43:16 2011
@@ -227,10 +227,10 @@ public class SimpleAMQQueueTest extends
}
/**
- * Tests that a re-queued message is resent to the subscriber. Verifies also that the
+ * Tests that a released queue entry is resent to the subscriber. Verifies also that the
* QueueContext._releasedEntry is reset to null after the entry has been reset.
*/
- public void testRequeuedMessageIsResentToSubscriber() throws Exception
+ public void testReleasedMessageIsResentToSubscriber() throws Exception
{
_queue.registerSubscription(_subscription, false);
@@ -253,19 +253,18 @@ public class SimpleAMQQueueTest extends
_queue.enqueue(messageB, postEnqueueAction);
_queue.enqueue(messageC, postEnqueueAction);
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
assertEquals("Unexpected total number of messages sent to subscription", 3, _subscription.getMessages().size());
assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered());
assertFalse("Redelivery flag should not be set", queueEntries.get(1).isRedelivered());
assertFalse("Redelivery flag should not be set", queueEntries.get(2).isRedelivered());
- /* Now requeue the first message only */
+ /* Now release the first message only, causing it to be requeued */
queueEntries.get(0).release();
- _queue.requeue(queueEntries.get(0));
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
assertEquals("Unexpected total number of messages sent to subscription", 4, _subscription.getMessages().size());
assertTrue("Redelivery flag should now be set", queueEntries.get(0).isRedelivered());
@@ -275,11 +274,11 @@ public class SimpleAMQQueueTest extends
}
/**
- * Tests that a re-queued message that becomes expired is not resent to the subscriber.
+ * Tests that a released message that becomes expired is not resent to the subscriber.
* This tests ensures that SimpleAMQQueueEntry.getNextAvailableEntry avoids expired entries.
* Verifies also that the QueueContext._releasedEntry is reset to null after the entry has been reset.
*/
- public void testRequeuedMessageThatBecomesExpiredIsNotRedelivered() throws Exception
+ public void testReleaseMessageThatBecomesExpiredIsNotRedelivered() throws Exception
{
_queue.registerSubscription(_subscription, false);
@@ -301,17 +300,16 @@ public class SimpleAMQQueueTest extends
_queue.enqueue(messageA, postEnqueueAction);
int subFlushWaitTime = 150;
- Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner Thread
+ Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner/QueueRunner Threads
assertEquals("Unexpected total number of messages sent to subscription", 1, _subscription.getMessages().size());
assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered());
- /* Wait a little more to be sure that message will have expired, then requeue it */
+ /* Wait a little more to be sure that message will have expired, then release the first message only, causing it to be requeued */
Thread.sleep(messageExpirationOffset - subFlushWaitTime + 10);
queueEntries.get(0).release();
- _queue.requeue(queueEntries.get(0));
- Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner Thread
+ Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner/QueueRunner Threads
assertTrue("Expecting the queue entry to be now expired", queueEntries.get(0).expired());
assertEquals("Total number of messages sent should not have changed", 1, _subscription.getMessages().size());
@@ -321,12 +319,12 @@ public class SimpleAMQQueueTest extends
}
/**
- * Tests that if a client requeues messages 'out of order' (the order
+ * Tests that if a client releases entries 'out of order' (the order
* used by QueueEntryImpl.compareTo) that messages are still resent
* successfully. Specifically this test ensures the {@see SimpleAMQQueue#requeue()}
* can correctly move the _releasedEntry to an earlier position in the QueueEntry list.
*/
- public void testMessagesRequeuedOutOfComparableOrderAreDelivered() throws Exception
+ public void testReleasedOutOfComparableOrderAreRedelivered() throws Exception
{
_queue.registerSubscription(_subscription, false);
@@ -349,21 +347,19 @@ public class SimpleAMQQueueTest extends
_queue.enqueue(messageB, postEnqueueAction);
_queue.enqueue(messageC, postEnqueueAction);
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
assertEquals("Unexpected total number of messages sent to subscription", 3, _subscription.getMessages().size());
assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered());
assertFalse("Redelivery flag should not be set", queueEntries.get(1).isRedelivered());
assertFalse("Redelivery flag should not be set", queueEntries.get(2).isRedelivered());
- /* Now requeue the third and first message only */
+ /* Now release the third and first message only, causing it to be requeued */
queueEntries.get(2).release();
queueEntries.get(0).release();
- _queue.requeue(queueEntries.get(2));
- _queue.requeue(queueEntries.get(0));
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
assertEquals("Unexpected total number of messages sent to subscription", 5, _subscription.getMessages().size());
assertTrue("Redelivery flag should now be set", queueEntries.get(0).isRedelivered());
@@ -374,10 +370,10 @@ public class SimpleAMQQueueTest extends
/**
- * Tests a requeue for a queue with multiple subscriptions. Verifies that a
+ * Tests that a release requeues an entry for a queue with multiple subscriptions. Verifies that a
* requeue resends a message to a <i>single</i> subscriber.
*/
- public void testRequeueForQueueWithMultipleSubscriptions() throws Exception
+ public void testReleaseForQueueWithMultipleSubscriptions() throws Exception
{
MockSubscription subscription1 = new MockSubscription();
MockSubscription subscription2 = new MockSubscription();
@@ -402,66 +398,16 @@ public class SimpleAMQQueueTest extends
_queue.enqueue(messageA, postEnqueueAction);
_queue.enqueue(messageB, postEnqueueAction);
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
- assertEquals("Unexpected total number of messages sent to subscription1 after enqueue", 1, subscription1.getMessages().size());
- assertEquals("Unexpected total number of messages sent to subscription2 after enqueue", 1, subscription2.getMessages().size());
+ assertEquals("Unexpected total number of messages sent to both after enqueue", 2, subscription1.getMessages().size() + subscription2.getMessages().size());
- /* Now requeue a message (for any subscription) */
+ /* Now release the first message only, causing it to be requeued */
+ queueEntries.get(0).release();
- queueEntries.get(0).release();
- _queue.requeue((QueueEntryImpl)queueEntries.get(0));
-
- Thread.sleep(150); // Work done by SubFlushRunner Thread
+ Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads
- assertEquals("Unexpected total number of messages sent to all subscriptions after requeue", 3, subscription1.getMessages().size() + subscription2.getMessages().size());
- assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription1.getQueueContext())._releasedEntry);
- assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription2.getQueueContext())._releasedEntry);
- }
-
- /**
- * Tests a requeue for a queue with multiple subscriptions. Verifies that a
- * subscriber specific requeue resends the message to <i>that</i> subscriber.
- */
- public void testSubscriptionSpecificRequeueForQueueWithMultipleSubscriptions() throws Exception
- {
- MockSubscription subscription1 = new MockSubscription();
- MockSubscription subscription2 = new MockSubscription();
-
- _queue.registerSubscription(subscription1, false);
- _queue.registerSubscription(subscription2, false);
-
- final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>();
- PostEnqueueAction postEnqueueAction = new PostEnqueueAction()
- {
- public void onEnqueue(QueueEntry entry)
- {
- queueEntries.add(entry);
- }
- };
-
- AMQMessage messageA = createMessage(new Long(24));
- AMQMessage messageB = createMessage(new Long(25));
-
- /* Enqueue two messages */
-
- _queue.enqueue(messageA, postEnqueueAction);
- _queue.enqueue(messageB, postEnqueueAction);
-
- Thread.sleep(150); // Work done by SubFlushRunner Thread
-
- assertEquals("Unexpected total number of messages sent to subscription1 after enqueue", 1, subscription1.getMessages().size());
- assertEquals("Unexpected total number of messages sent to subscription2 after enqueue", 1, subscription2.getMessages().size());
-
- /* Now requeue a message (for first subscription) */
-
- queueEntries.get(0).release();
- _queue.requeue((QueueEntryImpl)queueEntries.get(0), subscription1);
-
- Thread.sleep(150); // Work done by SubFlushRunner Thread
-
- assertEquals("Unexpected total number of messages sent to subscription1 after requeue", 2, subscription1.getMessages().size());
- assertEquals("Unexpected total number of messages sent to subscription2 after requeue", 1, subscription2.getMessages().size());
+ assertEquals("Unexpected total number of messages sent to both subscriptions after release", 3, subscription1.getMessages().size() + subscription2.getMessages().size());
assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription1.getQueueContext())._releasedEntry);
assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription2.getQueueContext())._releasedEntry);
}
@@ -660,8 +606,8 @@ public class SimpleAMQQueueTest extends
// Create IncomingMessage and nondurable queue
final IncomingMessage msg = new IncomingMessage(info);
ContentHeaderBody contentHeaderBody = new ContentHeaderBody();
- contentHeaderBody.properties = new BasicContentHeaderProperties();
- ((BasicContentHeaderProperties) contentHeaderBody.properties).setDeliveryMode((byte) 2);
+ contentHeaderBody.setProperties(new BasicContentHeaderProperties());
+ ((BasicContentHeaderProperties) contentHeaderBody.getProperties()).setDeliveryMode((byte) 2);
msg.setContentHeaderBody(contentHeaderBody);
final ArrayList<BaseQueue> qs = new ArrayList<BaseQueue>();
@@ -707,6 +653,111 @@ public class SimpleAMQQueueTest extends
}
+ /**
+ * processQueue() is used when asynchronously delivering messages to
+ * subscriptions which could not be delivered immediately during the
+ * enqueue() operation.
+ *
+ * A defect within the method would mean that delivery of these messages may
+ * not occur should the Runner stop before all messages have been processed.
+ * Such a defect was discovered when Selectors were used such that one and
+ * only one subscription can/will accept any given messages, but multiple
+ * subscriptions are present, and one of the earlier subscriptions receives
+ * more messages than the others.
+ *
+ * This test is to validate that the processQueue() method is able to
+ * correctly deliver all of the messages present for asynchronous delivery
+ * to subscriptions in such a scenario.
+ */
+ public void testProcessQueueWithUniqueSelectors() throws Exception
+ {
+ TestSimpleQueueEntryListFactory factory = new TestSimpleQueueEntryListFactory();
+ SimpleAMQQueue testQueue = new SimpleAMQQueue("testQueue", false, "testOwner",false,
+ false, _virtualHost, factory, null)
+ {
+ @Override
+ public void deliverAsync(Subscription sub)
+ {
+ // do nothing, i.e prevent deliveries by the SubFlushRunner
+ // when registering the new subscriptions
+ }
+ };
+
+ // retrieve the QueueEntryList the queue creates and insert the test
+ // messages, thus avoiding straight-through delivery attempts during
+ //enqueue() process.
+ QueueEntryList list = factory.getQueueEntryList();
+ assertNotNull("QueueEntryList should have been created", list);
+
+ QueueEntry msg1 = list.add(createMessage(1L));
+ QueueEntry msg2 = list.add(createMessage(2L));
+ QueueEntry msg3 = list.add(createMessage(3L));
+ QueueEntry msg4 = list.add(createMessage(4L));
+ QueueEntry msg5 = list.add(createMessage(5L));
+
+ // Create lists of the entries each subscription should be interested
+ // in.Bias over 50% of the messages to the first subscription so that
+ // the later subscriptions reject them and report being done before
+ // the first subscription as the processQueue method proceeds.
+ List<QueueEntry> msgListSub1 = createEntriesList(msg1, msg2, msg3);
+ List<QueueEntry> msgListSub2 = createEntriesList(msg4);
+ List<QueueEntry> msgListSub3 = createEntriesList(msg5);
+
+ MockSubscription sub1 = new MockSubscription(msgListSub1);
+ MockSubscription sub2 = new MockSubscription(msgListSub2);
+ MockSubscription sub3 = new MockSubscription(msgListSub3);
+
+ // register the subscriptions
+ testQueue.registerSubscription(sub1, false);
+ testQueue.registerSubscription(sub2, false);
+ testQueue.registerSubscription(sub3, false);
+
+ //check that no messages have been delivered to the
+ //subscriptions during registration
+ assertEquals("No messages should have been delivered yet", 0, sub1.getMessages().size());
+ assertEquals("No messages should have been delivered yet", 0, sub2.getMessages().size());
+ assertEquals("No messages should have been delivered yet", 0, sub3.getMessages().size());
+
+ // call processQueue to deliver the messages
+ testQueue.processQueue(new QueueRunner(testQueue, 1)
+ {
+ @Override
+ public void run()
+ {
+ // we dont actually want/need this runner to do any work
+ // because we we are already doing it!
+ }
+ });
+
+ // check expected messages delivered to correct consumers
+ verifyRecievedMessages(msgListSub1, sub1.getMessages());
+ verifyRecievedMessages(msgListSub2, sub2.getMessages());
+ verifyRecievedMessages(msgListSub3, sub3.getMessages());
+ }
+
+ private List<QueueEntry> createEntriesList(QueueEntry... entries)
+ {
+ ArrayList<QueueEntry> entriesList = new ArrayList<QueueEntry>();
+ for (QueueEntry entry : entries)
+ {
+ entriesList.add(entry);
+ }
+ return entriesList;
+ }
+
+ private void verifyRecievedMessages(List<QueueEntry> expected,
+ List<QueueEntry> delivered)
+ {
+ assertEquals("Consumer did not receive the expected number of messages",
+ expected.size(), delivered.size());
+
+ for (QueueEntry msg : expected)
+ {
+ assertTrue("Consumer did not recieve msg: "
+ + msg.getMessage().getMessageNumber(), delivered.contains(msg));
+ }
+ }
+
public class TestMessage extends AMQMessage
{
private final long _tag;
@@ -747,4 +798,20 @@ public class SimpleAMQQueueTest extends
AMQMessage messageA = new TestMessage(id, id, info);
return messageA;
}
+
+ class TestSimpleQueueEntryListFactory implements QueueEntryListFactory
+ {
+ QueueEntryList _list;
+
+ public QueueEntryList createQueueEntryList(AMQQueue queue)
+ {
+ _list = new SimpleQueueEntryList(queue);
+ return _list;
+ }
+
+ public QueueEntryList getQueueEntryList()
+ {
+ return _list;
+ }
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java Mon Aug 15 09:43:16 2011
@@ -589,7 +589,7 @@ public class MessageStoreTest extends In
headerBody.classId = BasicConsumeBodyImpl.CLASS_ID;
headerBody.bodySize = 0;
- headerBody.properties = properties;
+ headerBody.setProperties(properties);
try
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java Mon Aug 15 09:43:16 2011
@@ -102,7 +102,7 @@ public class ReferenceCountingTest exten
ContentHeaderBody chb = new ContentHeaderBody();
BasicContentHeaderProperties bchp = new BasicContentHeaderProperties();
bchp.setDeliveryMode((byte)2);
- chb.properties = bchp;
+ chb.setProperties(bchp);
return chb;
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java Mon Aug 15 09:43:16 2011
@@ -22,6 +22,7 @@ package org.apache.qpid.server.subscript
*/
import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -45,6 +46,7 @@ public class MockSubscription implements
private State _state = State.ACTIVE;
private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>();
private final Lock _stateChangeLock = new ReentrantLock();
+ private List<QueueEntry> _acceptEntries = null;
private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this);
private final QueueEntry.SubscriptionAssignedState _assignedState = new QueueEntry.SubscriptionAssignedState(this);
@@ -54,6 +56,15 @@ public class MockSubscription implements
// Create a simple ID that increments for ever new Subscription
private final long _subscriptionID = idGenerator.getAndIncrement();
+ public MockSubscription()
+ {
+ }
+
+ public MockSubscription(List<QueueEntry> acceptEntries)
+ {
+ _acceptEntries = acceptEntries;
+ }
+
public void close()
{
_closed = true;
@@ -119,8 +130,15 @@ public class MockSubscription implements
_stateChangeLock.lock();
}
- public boolean hasInterest(QueueEntry msg)
+ public boolean hasInterest(QueueEntry entry)
{
+ if(_acceptEntries != null)
+ {
+ //simulate selector behaviour, only signal
+ //interest in the dictated queue entries
+ return _acceptEntries.contains(entry);
+ }
+
return true;
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java Mon Aug 15 09:43:16 2011
@@ -243,7 +243,7 @@ public class InternalBrokerBaseCase exte
//Make Message Persistent
properties.setDeliveryMode((byte) 2);
- _headerBody.properties = properties;
+ _headerBody.setProperties(properties);
channel.publishContentHeader(_headerBody);
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java Mon Aug 15 09:43:16 2011
@@ -481,7 +481,7 @@ public class GenerateLogMessages
// Only check the text inside the braces '{}'
int typeIndexEnd = parametersString[index].indexOf("}", typeIndex);
String typeString = parametersString[index].substring(typeIndex, typeIndexEnd);
- if (typeString.contains("number"))
+ if (typeString.contains("number") || typeString.contains("choice"))
{
type = "Number";
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/build.xml
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/build.xml?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/build.xml (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/build.xml Mon Aug 15 09:43:16 2011
@@ -93,6 +93,10 @@
<fail if="failed" message="TEST SUITE FAILED"/>
</target>
+
+ <target name="report-module" description="generate junitreport for modules">
+ <iterate target="report-module"/>
+ </target>
<target name="jar" description="create module jars">
<iterate target="jar"/>
@@ -213,7 +217,18 @@
<include name="**/*.ser"/>
</fileset>
</cobertura-merge>
- <cobertura-report format="html"
+ <cobertura-report format="xml"
+ destdir="${build.coveragereport}"
+ datafile="${build.coveragereport}/cobertura.ser"
+ >
+ <fileset dir="${project.root}/common/src/main/java" includes="**/*.java" />
+ <fileset dir="${project.root}/build/scratch/common/src" includes="**/*.java" />
+ <fileset dir="${project.root}/broker/src/main/java" includes="**/*.java" />
+ <fileset dir="${project.root}/build/scratch/broker/src" includes="**/*.java" />
+ <fileset dir="${project.root}/client/src/main/java" includes="**/*.java" />
+ <fileset dir="${project.root}/build/scratch/client/src" includes="**/*.java" />
+ </cobertura-report>
+ <cobertura-report format="html"
destdir="${build.coveragereport}"
datafile="${build.coveragereport}/cobertura.ser"
>
@@ -277,4 +292,7 @@
</findbugs>
</target>
+ <target name="eclipse" description="build eclipse project and classpath files">
+ <iterate target="eclipse"/>
+ </target>
</project>
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/ConnectionSetup.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/ConnectionSetup.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/ConnectionSetup.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/ConnectionSetup.java Mon Aug 15 09:43:16 2011
@@ -50,7 +50,7 @@ public class ConnectionSetup
final static String QUEUE_NAME = "example.MyQueue";
public static final String TOPIC_JNDI_NAME = "topic";
- final static String TOPIC_NAME = "example.hierarical.topic";
+ final static String TOPIC_NAME = "usa.news";
private Context _ctx;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/Publisher.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/Publisher.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/Publisher.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/example/src/main/java/org/apache/qpid/example/pubsub/Publisher.java Mon Aug 15 09:43:16 2011
@@ -71,7 +71,7 @@ public class Publisher extends Client
public static void main(String[] args)
{
- String destination = args.length > 2 ? args[1] : null;
+ String destination = args.length > 2 ? args[1] : "usa.news";
int msgCount = args.length > 2 ? Integer.parseInt(args[2]) : 100;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/client.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/client.bnd?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/client.bnd (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/client.bnd Mon Aug 15 09:43:16 2011
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.9.0
+ver: 0.11.0
Bundle-SymbolicName: qpid-client
Bundle-Version: ${ver}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java Mon Aug 15 09:43:16 2011
@@ -105,6 +105,21 @@ public class AMQBrokerDetails implements
if (host == null)
{
host = "";
+
+ String auth = connection.getAuthority();
+ if (auth != null)
+ {
+ // contains both host & port myhost:5672
+ if (auth.contains(":"))
+ {
+ host = auth.substring(0,auth.indexOf(":"));
+ }
+ else
+ {
+ host = auth;
+ }
+ }
+
}
setHost(host);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Mon Aug 15 09:43:16 2011
@@ -69,6 +69,12 @@ public class AMQConnectionDelegate_0_10
org.apache.qpid.transport.Connection _qpidConnection;
private ConnectionException exception = null;
+ static
+ {
+ // Register any configured SASL client factories.
+ org.apache.qpid.client.security.DynamicSaslRegistrar.registerSaslProviders();
+ }
+
//--- constructor
public AMQConnectionDelegate_0_10(AMQConnection conn)
{
@@ -211,6 +217,8 @@ public class AMQConnectionDelegate_0_10
public void resubscribeSessions() throws JMSException, AMQException, FailoverException
{
+ _logger.info("Resuming connection");
+ getQpidConnection().resume();
List<AMQSession> sessions = new ArrayList<AMQSession>(_conn.getSessions().values());
_logger.info(String.format("Resubscribing sessions = %s sessions.size=%d", sessions, sessions.size()));
for (AMQSession s : sessions)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Mon Aug 15 09:43:16 2011
@@ -1043,7 +1043,7 @@ public abstract class AMQSession<C exten
throws JMSException
{
checkNotClosed();
- AMQTopic origTopic = checkValidTopic(topic, true);
+ Topic origTopic = checkValidTopic(topic, true);
AMQTopic dest = AMQTopic.createDurableTopic(origTopic, name, _connection);
String messageSelector = ((selector == null) || (selector.trim().length() == 0)) ? null : selector;
@@ -1225,7 +1225,6 @@ public abstract class AMQSession<C exten
else
{
AMQQueue queue = new AMQQueue(queueName);
- queue.setCreate(AddressOption.ALWAYS);
return queue;
}
@@ -1307,8 +1306,8 @@ public abstract class AMQSession<C exten
public QueueReceiver createQueueReceiver(Destination destination) throws JMSException
{
checkValidDestination(destination);
- AMQQueue dest = (AMQQueue) destination;
- C consumer = (C) createConsumer(destination);
+ Queue dest = validateQueue(destination);
+ C consumer = (C) createConsumer(dest);
return new QueueReceiverAdaptor(dest, consumer);
}
@@ -1326,8 +1325,8 @@ public abstract class AMQSession<C exten
public QueueReceiver createQueueReceiver(Destination destination, String messageSelector) throws JMSException
{
checkValidDestination(destination);
- AMQQueue dest = (AMQQueue) destination;
- C consumer = (C) createConsumer(destination, messageSelector);
+ Queue dest = validateQueue(destination);
+ C consumer = (C) createConsumer(dest, messageSelector);
return new QueueReceiverAdaptor(dest, consumer);
}
@@ -1344,7 +1343,7 @@ public abstract class AMQSession<C exten
public QueueReceiver createReceiver(Queue queue) throws JMSException
{
checkNotClosed();
- AMQQueue dest = (AMQQueue) queue;
+ Queue dest = validateQueue(queue);
C consumer = (C) createConsumer(dest);
return new QueueReceiverAdaptor(dest, consumer);
@@ -1363,11 +1362,23 @@ public abstract class AMQSession<C exten
public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
{
checkNotClosed();
- AMQQueue dest = (AMQQueue) queue;
+ Queue dest = validateQueue(queue);
C consumer = (C) createConsumer(dest, messageSelector);
return new QueueReceiverAdaptor(dest, consumer);
}
+
+ private Queue validateQueue(Destination dest) throws InvalidDestinationException
+ {
+ if (dest instanceof AMQDestination && dest instanceof javax.jms.Queue)
+ {
+ return (Queue)dest;
+ }
+ else
+ {
+ throw new InvalidDestinationException("The destination object used is not from this provider or of type javax.jms.Queue");
+ }
+ }
public QueueSender createSender(Queue queue) throws JMSException
{
@@ -1408,7 +1419,7 @@ public abstract class AMQSession<C exten
public TopicSubscriber createSubscriber(Topic topic) throws JMSException
{
checkNotClosed();
- AMQTopic dest = checkValidTopic(topic);
+ Topic dest = checkValidTopic(topic);
// AMQTopic dest = new AMQTopic(topic.getTopicName());
return new TopicSubscriberAdaptor(dest, (C) createExclusiveConsumer(dest));
@@ -1428,7 +1439,7 @@ public abstract class AMQSession<C exten
public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
{
checkNotClosed();
- AMQTopic dest = checkValidTopic(topic);
+ Topic dest = checkValidTopic(topic);
// AMQTopic dest = new AMQTopic(topic.getTopicName());
return new TopicSubscriberAdaptor(dest, (C) createExclusiveConsumer(dest, messageSelector, noLocal));
@@ -2395,7 +2406,7 @@ public abstract class AMQSession<C exten
/*
* I could have combined the last 3 methods, but this way it improves readability
*/
- protected AMQTopic checkValidTopic(Topic topic, boolean durable) throws JMSException
+ protected Topic checkValidTopic(Topic topic, boolean durable) throws JMSException
{
if (topic == null)
{
@@ -2414,17 +2425,17 @@ public abstract class AMQSession<C exten
("Cannot create a durable subscription with a temporary topic: " + topic);
}
- if (!(topic instanceof AMQTopic))
+ if (!(topic instanceof AMQDestination && topic instanceof javax.jms.Topic))
{
throw new javax.jms.InvalidDestinationException(
"Cannot create a subscription on topic created for another JMS Provider, class of topic provided is: "
+ topic.getClass().getName());
}
- return (AMQTopic) topic;
+ return topic;
}
- protected AMQTopic checkValidTopic(Topic topic) throws JMSException
+ protected Topic checkValidTopic(Topic topic) throws JMSException
{
return checkValidTopic(topic, false);
}
@@ -2819,6 +2830,7 @@ public abstract class AMQSession<C exten
{
declareQueue(amqd, protocolHandler, consumer.isNoLocal(), nowait);
}
+ bindQueue(amqd.getAMQQueueName(), amqd.getRoutingKey(), consumer.getArguments(), amqd.getExchangeName(), amqd, nowait);
}
AMQShortString queueName = amqd.getAMQQueueName();
@@ -2826,8 +2838,6 @@ public abstract class AMQSession<C exten
// store the consumer queue name
consumer.setQueuename(queueName);
- bindQueue(queueName, amqd.getRoutingKey(), consumer.getArguments(), amqd.getExchangeName(), amqd, nowait);
-
// If IMMEDIATE_PREFETCH is not required then suspsend the channel to delay prefetch
if (!_immediatePrefetch)
{
@@ -3450,4 +3460,9 @@ public abstract class AMQSession<C exten
{
return _closing.get()|| _connection.isClosing();
}
+
+ public boolean isDeclareExchanges()
+ {
+ return DECLARE_EXCHANGES;
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java Mon Aug 15 09:43:16 2011
@@ -47,6 +47,8 @@ import org.apache.qpid.client.message.AM
import org.apache.qpid.client.message.FieldTableSupport;
import org.apache.qpid.client.message.MessageFactoryRegistry;
import org.apache.qpid.client.message.UnprocessedMessage_0_10;
+import org.apache.qpid.client.messaging.address.Link;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
import org.apache.qpid.client.messaging.address.Node.ExchangeNode;
import org.apache.qpid.client.messaging.address.Node.QueueNode;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
@@ -56,6 +58,7 @@ import org.apache.qpid.framing.FieldTabl
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.transport.ExchangeBoundResult;
import org.apache.qpid.transport.ExchangeQueryResult;
+import org.apache.qpid.transport.ExecutionErrorCode;
import org.apache.qpid.transport.ExecutionException;
import org.apache.qpid.transport.MessageAcceptMode;
import org.apache.qpid.transport.MessageAcquireMode;
@@ -314,7 +317,7 @@ public class AMQSession_0_10 extends AMQ
public void sendQueueBind(final AMQShortString queueName, final AMQShortString routingKey,
final FieldTable arguments, final AMQShortString exchangeName,
final AMQDestination destination, final boolean nowait)
- throws AMQException, FailoverException
+ throws AMQException
{
if (destination.getDestSyntax() == DestSyntax.BURL)
{
@@ -600,10 +603,16 @@ public class AMQSession_0_10 extends AMQ
(Map<? extends String, ? extends Object>) consumer.getDestination().getLink().getSubscription().getArgs());
}
+ boolean acceptModeNone = getAcknowledgeMode() == NO_ACKNOWLEDGE;
+
+ if (consumer.getDestination().getLink() != null)
+ {
+ acceptModeNone = consumer.getDestination().getLink().getReliability() == Link.Reliability.UNRELIABLE;
+ }
getQpidSession().messageSubscribe
(queueName.toString(), String.valueOf(tag),
- getAcknowledgeMode() == NO_ACKNOWLEDGE ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT,
+ acceptModeNone ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT,
preAcquire ? MessageAcquireMode.PRE_ACQUIRED : MessageAcquireMode.NOT_ACQUIRED, null, 0, arguments,
consumer.isExclusive() ? Option.EXCLUSIVE : Option.NONE);
}
@@ -1068,22 +1077,37 @@ public class AMQSession_0_10 extends AMQ
return match;
}
- public boolean isQueueExist(AMQDestination dest,QueueNode node,boolean assertNode)
+ public boolean isQueueExist(AMQDestination dest,QueueNode node,boolean assertNode) throws AMQException
{
boolean match = true;
- QueueQueryResult result = getQpidSession().queueQuery(dest.getAddressName(), Option.NONE).get();
- match = dest.getAddressName().equals(result.getQueue());
-
- if (match && assertNode)
+ try
{
- match = (result.getDurable() == node.isDurable()) &&
- (result.getAutoDelete() == node.isAutoDelete()) &&
- (result.getExclusive() == node.isExclusive()) &&
- (matchProps(result.getArguments(),node.getDeclareArgs()));
+ QueueQueryResult result = getQpidSession().queueQuery(dest.getAddressName(), Option.NONE).get();
+ match = dest.getAddressName().equals(result.getQueue());
+
+ if (match && assertNode)
+ {
+ match = (result.getDurable() == node.isDurable()) &&
+ (result.getAutoDelete() == node.isAutoDelete()) &&
+ (result.getExclusive() == node.isExclusive()) &&
+ (matchProps(result.getArguments(),node.getDeclareArgs()));
+ }
+ else if (match)
+ {
+ // should I use the queried details to update the local data structure.
+ }
}
- else if (match)
+ catch(SessionException e)
{
- // should I use the queried details to update the local data structure.
+ if (e.getException().getErrorCode() == ExecutionErrorCode.RESOURCE_DELETED)
+ {
+ match = false;
+ }
+ else
+ {
+ throw new AMQException(AMQConstant.getConstant(e.getException().getErrorCode().getValue()),
+ "Error querying queue",e);
+ }
}
return match;
@@ -1149,6 +1173,22 @@ public class AMQSession_0_10 extends AMQ
int type = resolveAddressType(dest);
+ if (type == AMQDestination.QUEUE_TYPE &&
+ dest.getLink().getReliability() == Reliability.UNSPECIFIED)
+ {
+ dest.getLink().setReliability(Reliability.AT_LEAST_ONCE);
+ }
+ else if (type == AMQDestination.TOPIC_TYPE &&
+ dest.getLink().getReliability() == Reliability.UNSPECIFIED)
+ {
+ dest.getLink().setReliability(Reliability.UNRELIABLE);
+ }
+ else if (type == AMQDestination.TOPIC_TYPE &&
+ dest.getLink().getReliability() == Reliability.AT_LEAST_ONCE)
+ {
+ throw new AMQException("AT-LEAST-ONCE is not yet supported for Topics");
+ }
+
switch (type)
{
case AMQDestination.QUEUE_TYPE:
@@ -1162,6 +1202,8 @@ public class AMQSession_0_10 extends AMQ
{
setLegacyFiledsForQueueType(dest);
send0_10QueueDeclare(dest,null,false,noWait);
+ sendQueueBind(dest.getAMQQueueName(), dest.getRoutingKey(),
+ null,dest.getExchangeName(),dest, false);
break;
}
}
@@ -1270,6 +1312,8 @@ public class AMQSession_0_10 extends AMQ
dest.getQueueName(),// should have one by now
dest.getSubject(),
Collections.<String,Object>emptyMap()));
+ sendQueueBind(dest.getAMQQueueName(), dest.getRoutingKey(),
+ null,dest.getExchangeName(),dest, false);
}
public void setLegacyFiledsForQueueType(AMQDestination dest)
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java Mon Aug 15 09:43:16 2011
@@ -22,6 +22,7 @@ package org.apache.qpid.client;
import java.net.URISyntaxException;
+import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Topic;
@@ -95,39 +96,47 @@ public class AMQTopic extends AMQDestina
super(exchangeName, exchangeClass, routingKey, isExclusive, isAutoDelete, queueName, isDurable,bindingKeys);
}
- public static AMQTopic createDurableTopic(AMQTopic topic, String subscriptionName, AMQConnection connection)
+ public static AMQTopic createDurableTopic(Topic topic, String subscriptionName, AMQConnection connection)
throws JMSException
{
- if (topic.getDestSyntax() == DestSyntax.ADDR)
+ if (topic instanceof AMQDestination && topic instanceof javax.jms.Topic)
{
- try
+ AMQDestination qpidTopic = (AMQDestination)topic;
+ if (qpidTopic.getDestSyntax() == DestSyntax.ADDR)
{
- AMQTopic t = new AMQTopic(topic.getAddress());
- AMQShortString queueName = getDurableTopicQueueName(subscriptionName, connection);
- // link is never null if dest was created using an address string.
- t.getLink().setName(queueName.asString());
- t.getSourceNode().setAutoDelete(false);
- t.getSourceNode().setDurable(true);
-
- // The legacy fields are also populated just in case.
- t.setQueueName(queueName);
- t.setAutoDelete(false);
- t.setDurable(true);
- return t;
+ try
+ {
+ AMQTopic t = new AMQTopic(qpidTopic.getAddress());
+ AMQShortString queueName = getDurableTopicQueueName(subscriptionName, connection);
+ // link is never null if dest was created using an address string.
+ t.getLink().setName(queueName.asString());
+ t.getSourceNode().setAutoDelete(false);
+ t.getSourceNode().setDurable(true);
+
+ // The legacy fields are also populated just in case.
+ t.setQueueName(queueName);
+ t.setAutoDelete(false);
+ t.setDurable(true);
+ return t;
+ }
+ catch(Exception e)
+ {
+ JMSException ex = new JMSException("Error creating durable topic");
+ ex.initCause(e);
+ ex.setLinkedException(e);
+ throw ex;
+ }
}
- catch(Exception e)
+ else
{
- JMSException ex = new JMSException("Error creating durable topic");
- ex.initCause(e);
- ex.setLinkedException(e);
- throw ex;
+ return new AMQTopic(qpidTopic.getExchangeName(), qpidTopic.getRoutingKey(), false,
+ getDurableTopicQueueName(subscriptionName, connection),
+ true);
}
}
else
{
- return new AMQTopic(topic.getExchangeName(), topic.getRoutingKey(), false,
- getDurableTopicQueueName(subscriptionName, connection),
- true);
+ throw new InvalidDestinationException("The destination object used is not from this provider or of type javax.jms.Topic");
}
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Mon Aug 15 09:43:16 2011
@@ -571,6 +571,7 @@ public abstract class BasicMessageConsum
if (!_session.isClosed() || _session.isClosing())
{
sendCancel();
+ cleanupQueue();
}
}
catch (AMQException e)
@@ -608,6 +609,8 @@ public abstract class BasicMessageConsum
}
abstract void sendCancel() throws AMQException, FailoverException;
+
+ abstract void cleanupQueue() throws AMQException, FailoverException;
/**
* Called when you need to invalidate a consumer. Used for example when failover has occurred and the client has
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java Mon Aug 15 09:43:16 2011
@@ -19,8 +19,11 @@ package org.apache.qpid.client;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.qpid.client.AMQDestination.AddressOption;
import org.apache.qpid.client.AMQDestination.DestSyntax;
+import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.message.*;
+import org.apache.qpid.client.messaging.address.Node.QueueNode;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
@@ -509,4 +512,18 @@ public class BasicMessageConsumer_0_10 e
return _exclusive;
}
}
+
+ void cleanupQueue() throws AMQException, FailoverException
+ {
+ AMQDestination dest = this.getDestination();
+ if (dest != null && dest.getDestSyntax() == AMQDestination.DestSyntax.ADDR)
+ {
+ if (dest.getDelete() == AddressOption.ALWAYS ||
+ dest.getDelete() == AddressOption.RECEIVER )
+ {
+ ((AMQSession_0_10) getSession()).getQpidSession().queueDelete(
+ this.getDestination().getQueueName());
+ }
+ }
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java Mon Aug 15 09:43:16 2011
@@ -88,4 +88,8 @@ public class BasicMessageConsumer_0_8 ex
return receive();
}
+ void cleanupQueue() throws AMQException, FailoverException
+ {
+
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java Mon Aug 15 09:43:16 2011
@@ -19,6 +19,7 @@ package org.apache.qpid.client;
import static org.apache.qpid.transport.Option.NONE;
import static org.apache.qpid.transport.Option.SYNC;
+import static org.apache.qpid.transport.Option.UNRELIABLE;
import java.nio.ByteBuffer;
import java.util.HashMap;
@@ -30,9 +31,12 @@ import javax.jms.JMSException;
import javax.jms.Message;
import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQDestination.AddressOption;
import org.apache.qpid.client.AMQDestination.DestSyntax;
import org.apache.qpid.client.message.AMQMessageDelegate_0_10;
import org.apache.qpid.client.message.AbstractJMSMessage;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
+import org.apache.qpid.client.messaging.address.Node.QueueNode;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.transport.DeliveryProperties;
import org.apache.qpid.transport.Header;
@@ -68,12 +72,15 @@ public class BasicMessageProducer_0_10 e
{
if (destination.getDestSyntax() == DestSyntax.BURL)
{
- String name = destination.getExchangeName().toString();
- ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
- (name,
- destination.getExchangeClass().toString(),
- null, null,
- name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+ if (getSession().isDeclareExchanges())
+ {
+ String name = destination.getExchangeName().toString();
+ ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
+ (name,
+ destination.getExchangeClass().toString(),
+ null, null,
+ name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+ }
}
else
{
@@ -210,6 +217,9 @@ public class BasicMessageProducer_0_10 e
deliveryMode == DeliveryMode.PERSISTENT)
);
+ boolean unreliable = (destination.getDestSyntax() == DestSyntax.ADDR) &&
+ (destination.getLink().getReliability() == Reliability.UNRELIABLE);
+
org.apache.mina.common.ByteBuffer data = message.getData();
ByteBuffer buffer = data == null ? ByteBuffer.allocate(0) : data.buf().slice();
@@ -217,7 +227,7 @@ public class BasicMessageProducer_0_10 e
MessageAcceptMode.NONE,
MessageAcquireMode.PRE_ACQUIRED,
new Header(deliveryProp, messageProps),
- buffer, sync ? SYNC : NONE);
+ buffer, sync ? SYNC : NONE, unreliable ? UNRELIABLE : NONE);
if (sync)
{
ssn.sync();
@@ -239,5 +249,21 @@ public class BasicMessageProducer_0_10 e
{
return _session.isQueueBound(destination);
}
+
+ @Override
+ public void close()
+ {
+ super.close();
+ AMQDestination dest = _destination;
+ if (dest != null && dest.getDestSyntax() == AMQDestination.DestSyntax.ADDR)
+ {
+ if (dest.getDelete() == AddressOption.ALWAYS ||
+ dest.getDelete() == AddressOption.SENDER )
+ {
+ ((AMQSession_0_10) getSession()).getQpidSession().queueDelete(
+ _destination.getQueueName());
+ }
+ }
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java Mon Aug 15 09:43:16 2011
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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.client;
import java.util.ArrayList;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java Mon Aug 15 09:43:16 2011
@@ -50,25 +50,25 @@ public class QueueSenderAdapter implemen
public void send(Message msg) throws JMSException
{
- checkPreConditions();
+ checkQueuePreConditions(_queue);
_delegate.send(msg);
}
public void send(Queue queue, Message msg) throws JMSException
{
- checkPreConditions(queue);
+ checkQueuePreConditions(queue);
_delegate.send(queue, msg);
}
public void publish(Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException
{
- checkPreConditions();
+ checkQueuePreConditions(_queue);
_delegate.send(msg, deliveryMode, priority, timeToLive);
}
public void send(Queue queue, Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException
{
- checkPreConditions(queue);
+ checkQueuePreConditions(queue);
_delegate.send(queue, msg, deliveryMode, priority, timeToLive);
}
@@ -122,19 +122,19 @@ public class QueueSenderAdapter implemen
public void send(Destination dest, Message msg) throws JMSException
{
- checkPreConditions((Queue) dest);
+ checkQueuePreConditions((Queue) dest);
_delegate.send(dest, msg);
}
public void send(Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException
{
- checkPreConditions();
+ checkQueuePreConditions(_queue);
_delegate.send(msg, deliveryMode, priority, timeToLive);
}
public void send(Destination dest, Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException
{
- checkPreConditions((Queue) dest);
+ checkQueuePreConditions((Queue) dest);
_delegate.send(dest, msg, deliveryMode, priority, timeToLive);
}
@@ -170,11 +170,6 @@ public class QueueSenderAdapter implemen
private void checkPreConditions() throws JMSException
{
- checkPreConditions(_queue);
- }
-
- private void checkPreConditions(Queue queue) throws JMSException
- {
if (closed)
{
throw new javax.jms.IllegalStateException("Publisher is closed");
@@ -186,39 +181,43 @@ public class QueueSenderAdapter implemen
{
throw new javax.jms.IllegalStateException("Invalid Session");
}
+ }
- if (queue == null)
- {
- throw new UnsupportedOperationException("Queue is null.");
- }
-
- if (!(queue instanceof AMQDestination))
- {
- throw new InvalidDestinationException("Queue: " + queue + " is not a valid Qpid queue");
- }
-
- AMQDestination destination = (AMQDestination) queue;
- if (!destination.isCheckedForQueueBinding() && checkQueueBeforePublish())
- {
-
- if (_delegate.getSession().isStrictAMQP())
- {
- _delegate._logger.warn("AMQP does not support destination validation before publish, ");
- destination.setCheckedForQueueBinding(true);
- }
- else
- {
- if (_delegate.isBound(destination))
- {
- destination.setCheckedForQueueBinding(true);
- }
- else
- {
- throw new InvalidDestinationException("Queue: " + queue
- + " is not a valid destination (no bindings on server");
- }
- }
- }
+ private void checkQueuePreConditions(Queue queue) throws JMSException
+ {
+ checkPreConditions() ;
+
+ if (queue == null)
+ {
+ throw new UnsupportedOperationException("Queue is null.");
+ }
+
+ if (!(queue instanceof AMQDestination))
+ {
+ throw new InvalidDestinationException("Queue: " + queue + " is not a valid Qpid queue");
+ }
+
+ AMQDestination destination = (AMQDestination) queue;
+ if (!destination.isCheckedForQueueBinding() && checkQueueBeforePublish())
+ {
+ if (_delegate.getSession().isStrictAMQP())
+ {
+ _delegate._logger.warn("AMQP does not support destination validation before publish, ");
+ destination.setCheckedForQueueBinding(true);
+ }
+ else
+ {
+ if (_delegate.isBound(destination))
+ {
+ destination.setCheckedForQueueBinding(true);
+ }
+ else
+ {
+ throw new InvalidDestinationException("Queue: " + queue
+ + " is not a valid destination (no bindings on server");
+ }
+ }
+ }
}
private boolean checkQueueBeforePublish()
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java Mon Aug 15 09:43:16 2011
@@ -22,10 +22,12 @@
package org.apache.qpid.client.message;
import java.lang.ref.SoftReference;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -670,7 +672,19 @@ public class AMQMessageDelegate_0_10 ext
public Enumeration getPropertyNames() throws JMSException
{
- return java.util.Collections.enumeration(getApplicationHeaders().keySet());
+ List<String> props = new ArrayList<String>();
+ Map<String, Object> propertyMap = getApplicationHeaders();
+ for (String prop: getApplicationHeaders().keySet())
+ {
+ Object value = propertyMap.get(prop);
+ if (value instanceof Boolean || value instanceof Number
+ || value instanceof String)
+ {
+ props.add(prop);
+ }
+ }
+
+ return java.util.Collections.enumeration(props);
}
public void setBooleanProperty(String propertyName, boolean b) throws JMSException
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQPEncodedMapMessage.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQPEncodedMapMessage.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQPEncodedMapMessage.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQPEncodedMapMessage.java Mon Aug 15 09:43:16 2011
@@ -23,6 +23,7 @@ package org.apache.qpid.client.message;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
@@ -65,7 +66,7 @@ public class AMQPEncodedMapMessage exten
if ((value instanceof Boolean) || (value instanceof Byte) || (value instanceof Short) || (value instanceof Integer)
|| (value instanceof Long) || (value instanceof Character) || (value instanceof Float)
|| (value instanceof Double) || (value instanceof String) || (value instanceof byte[])
- || (value instanceof List) || (value instanceof Map) || (value == null))
+ || (value instanceof List) || (value instanceof Map) || (value instanceof UUID) || (value == null))
{
_map.put(propName, value);
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java Mon Aug 15 09:43:16 2011
@@ -99,7 +99,7 @@ public abstract class AbstractJMSMessage
}
AMQMessageDelegate delegate = new AMQMessageDelegate_0_8(messageNbr,
- (BasicContentHeaderProperties) contentHeader.properties,
+ (BasicContentHeaderProperties) contentHeader.getProperties(),
exchange, routingKey);
return createMessage(delegate, data);
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java Mon Aug 15 09:43:16 2011
@@ -104,7 +104,7 @@ public class MessageFactoryRegistry
AMQShortString routingKey, ContentHeaderBody contentHeader, List bodies)
throws AMQException, JMSException
{
- BasicContentHeaderProperties properties = (BasicContentHeaderProperties) contentHeader.properties;
+ BasicContentHeaderProperties properties = (BasicContentHeaderProperties) contentHeader.getProperties();
// Get the message content type. This may be null for pure AMQP messages, but will always be set for JMS over
// AMQP. When the type is null, it can only be assumed that the message is a byte message.
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java Mon Aug 15 09:43:16 2011
@@ -27,6 +27,7 @@ import java.util.Map;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQDestination.Binding;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
import org.apache.qpid.client.messaging.address.Link.Subscription;
import org.apache.qpid.client.messaging.address.Node.ExchangeNode;
import org.apache.qpid.client.messaging.address.Node.QueueNode;
@@ -231,14 +232,9 @@ public class AddressHelper
private boolean getDurability(Map map)
{
- if (map != null && map.get(DURABLE) != null)
- {
- return Boolean.parseBoolean((String)map.get(DURABLE));
- }
- else
- {
- return false;
- }
+ Accessor access = new MapAccessor(map);
+ Boolean result = access.getBoolean(DURABLE);
+ return (result == null) ? false : result.booleanValue();
}
/**
@@ -262,7 +258,7 @@ public class AddressHelper
}
}
- public Link getLink()
+ public Link getLink() throws Exception
{
Link link = new Link();
link.setSubscription(new Subscription());
@@ -272,6 +268,25 @@ public class AddressHelper
: linkProps.getBoolean(DURABLE));
link.setName(linkProps.getString(NAME));
+ String reliability = linkProps.getString(RELIABILITY);
+ if ( reliability != null)
+ {
+ if (reliability.equalsIgnoreCase("unreliable"))
+ {
+ link.setReliability(Reliability.UNRELIABLE);
+ }
+ else if (reliability.equalsIgnoreCase("at-least-once"))
+ {
+ link.setReliability(Reliability.AT_LEAST_ONCE);
+ }
+ else
+ {
+ throw new Exception("The reliability mode '" +
+ reliability + "' is not yet supported");
+ }
+
+ }
+
if (((Map) address.getOptions().get(LINK)).get(CAPACITY) instanceof Map)
{
MapAccessor capacityProps = new MapAccessor(
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java Mon Aug 15 09:43:16 2011
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.client.messaging.address;
+import static org.apache.qpid.client.messaging.address.Link.Reliability.UNSPECIFIED;
+
import java.util.HashMap;
import java.util.Map;
@@ -29,6 +31,8 @@ public class Link
{
public enum FilterType { SQL92, XQUERY, SUBJECT }
+ public enum Reliability { UNRELIABLE, AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE, UNSPECIFIED }
+
protected String name;
protected String _filter;
protected FilterType _filterType = FilterType.SUBJECT;
@@ -38,7 +42,18 @@ public class Link
protected int _producerCapacity = 0;
protected Node node;
protected Subscription subscription;
+ protected Reliability reliability = UNSPECIFIED;
+ public Reliability getReliability()
+ {
+ return reliability;
+ }
+
+ public void setReliability(Reliability reliability)
+ {
+ this.reliability = reliability;
+ }
+
public Node getNode()
{
return node;
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties Mon Aug 15 09:43:16 2011
@@ -18,3 +18,4 @@
#
AMQPLAIN=org.apache.qpid.client.security.amqplain.AmqPlainSaslClientFactory
CRAM-MD5-HASHED=org.apache.qpid.client.security.crammd5hashed.CRAMMD5HashedSaslClientFactory
+ANONYMOUS=org.apache.qpid.client.security.anonymous.AnonymousSaslClientFactory
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -549,6 +549,37 @@ public class ConnectionURLTest extends T
assertTrue("String representation should contain options and values", url.toString().contains("maxprefetch='12345'"));
}
+ public void testHostNamesWithUnderScore() throws URLSyntaxException
+ {
+ String url = "amqp://guest:guest@clientid/test?brokerlist='tcp://under_score:6672'";
+
+ ConnectionURL connectionurl = new AMQConnectionURL(url);
+
+ 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("tcp"));
+ assertTrue(service.getHost().equals("under_score"));
+ assertTrue(service.getPort() == 6672);
+
+ url = "amqp://guest:guest@clientid/test?brokerlist='tcp://under_score'";
+
+ connectionurl = new AMQConnectionURL(url);
+
+ assertTrue(connectionurl.getUsername().equals("guest"));
+ assertTrue(connectionurl.getPassword().equals("guest"));
+ assertTrue(connectionurl.getVirtualHost().equals("/test"));
+
+ assertTrue(connectionurl.getBrokerCount() == 1);
+ service = connectionurl.getBrokerDetails(0);
+ assertTrue(service.getTransport().equals("tcp"));
+ assertTrue(service.getHost().equals("under_score"));
+ assertTrue(service.getPort() == 5672);
+ }
+
public static junit.framework.Test suite()
{
return new junit.framework.TestSuite(ConnectionURLTest.class);
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -23,7 +23,7 @@
<dirname property="project.root" file="${ant.file.common}"/>
<property name="project.name" value="qpid"/>
- <property name="project.version" value="0.9"/>
+ <property name="project.version" value="0.11"/>
<property name="project.namever" value="${project.name}-${project.version}"/>
<property name="resources" location="${project.root}/resources"/>
@@ -63,6 +63,11 @@
<property name="mllib.dir" value="${project.root}/../python" />
<property name="findbugs.dir" value="${project.root}/lib/findbugs" />
+ <!-- properties used to control Ant Eclipse for Eclipse classpath/project files etc -->
+ <property name="eclipse.updatealways" value="false"/>
+ <property name="eclipse.compilercompliance" value="5.0"/>
+ <property name="eclipse.container" value="JVM 1.5"/>
+
<path id="cobertura.classpath">
<fileset dir="${cobertura.dir}">
<include name="cobertura.jar" />
@@ -322,6 +327,20 @@
results directory:
${build.results}
+
+ ant eclipse
+
+ Generates project and classpath files for the Eclispe IDE. Requires that
+ the Ant Eclipse task (http://ant-eclipse.sourceforge.net/) has been installed
+ in $ANT_HOME/lib.
+
+ The following system properties will be passed to the task. These can be usefully
+ overridden from the command line.
+
+ eclipse.updatealways - forces Eclipse files to be regenerated even if they are newer then the build.xml (default ${eclipse.updatealways}).
+ eclipse.container - controls the Eclipse container (default ${eclipse.container}).
+ eclipse.compilercompliance" - controls the Eclipse compiler compliance (default ${eclipse.compilercompliance}).
+
</echo>
</target>
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.9.0
+ver: 0.11.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/framing/ContentHeaderBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java Mon Aug 15 09:43:16 2011
@@ -36,7 +36,7 @@ public class ContentHeaderBody implement
public long bodySize;
/** must never be null */
- public ContentHeaderProperties properties;
+ private ContentHeaderProperties properties;
public ContentHeaderBody()
{
@@ -128,4 +128,14 @@ public class ContentHeaderBody implement
{
return new AMQFrame(channelId, body);
}
+
+ public ContentHeaderProperties getProperties()
+ {
+ return properties;
+ }
+
+ public void setProperties(ContentHeaderProperties props)
+ {
+ properties = props;
+ }
}
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -28,6 +28,7 @@ import static org.apache.qpid.transport.
import static org.apache.qpid.transport.Connection.State.RESUMING;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -695,4 +696,8 @@ public class Connection extends Connecti
return connectionLost.get();
}
+ protected Collection<Session> getChannels()
+ {
+ return channels.values();
+ }
}
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -422,7 +422,10 @@ public class Session extends SessionInvo
{
return;
}
- sessionCompleted(copy, options);
+ if (copy.size() > 0)
+ {
+ sessionCompleted(copy, options);
+ }
}
}
@@ -661,7 +664,12 @@ public class Session extends SessionInvo
{
sessionCommandPoint(0, 0);
}
- if ((!closing && !transacted && m instanceof MessageTransfer) || m.hasCompletionListener())
+
+ boolean replayTransfer = !closing && !transacted &&
+ m instanceof MessageTransfer &&
+ ! m.isUnreliable();
+
+ if ((replayTransfer) || m.hasCompletionListener())
{
commands[mod(next, commands.length)] = m;
commandBytes += m.getBodySize();
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/codec/AbstractEncoder.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/codec/AbstractEncoder.java?rev=1157750&r1=1157749&r2=1157750&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/codec/AbstractEncoder.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/common/src/main/java/org/apache/qpid/transport/codec/AbstractEncoder.java Mon Aug 15 09:43:16 2011
@@ -63,6 +63,7 @@ abstract class AbstractEncoder implement
ENCODINGS.put(Double.class, Type.DOUBLE);
ENCODINGS.put(Character.class, Type.CHAR);
ENCODINGS.put(byte[].class, Type.VBIN32);
+ ENCODINGS.put(UUID.class, Type.UUID);
}
private final Map<String,byte[]> str8cache = new LinkedHashMap<String,byte[]>()
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=1157750&r1=1157749&r2=1157750&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 09:43:16 2011
@@ -339,7 +339,7 @@ public class FileUtils
}
//else we have a source directory
- if (!dst.isDirectory() && !dst.mkdir())
+ if (!dst.isDirectory() && !dst.mkdirs())
{
throw new UnableToCopyException("Unable to create destination directory");
}
Propchange: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/management/client/src/main/java/org/apache/qpid/management/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Mon Aug 15 09:43:16 2011
@@ -2,3 +2,4 @@
/qpid/branches/java-broker-0-10/qpid/java/management/client/src/main/java/org/apache/qpid/management:795950-829653
/qpid/branches/java-network-refactor/qpid/java/management/client/src/main/java/org/apache/qpid/management:805429-821809
/qpid/branches/qpid-2935/qpid/java/management/client/src/main/java/org/apache/qpid/management:1061302-1072333
+/qpid/trunk/qpid/java/management/client/src/main/java/org/apache/qpid/management:1073294-1090000
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org