You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2013/09/20 20:59:50 UTC
svn commit: r1525101 [19/21] - in /qpid/branches/linearstore/qpid: ./ bin/
cpp/ cpp/bindings/ cpp/bindings/qmf/ cpp/bindings/qmf/python/
cpp/bindings/qmf/ruby/ cpp/bindings/qmf/tests/ cpp/bindings/qmf2/
cpp/bindings/qmf2/examples/cpp/ cpp/bindings/qmf2...
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java Fri Sep 20 18:59:30 2013
@@ -19,12 +19,18 @@
*/
package org.apache.qpid.server.queue;
+import java.util.Collections;
import org.apache.qpid.AMQException;
-import org.apache.qpid.server.message.AMQMessage;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import java.util.Arrays;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
public class SortedQueueEntryListTest extends QueueEntryListTestBase
{
private static SelfValidatingSortedQueueEntryList _sqel;
@@ -113,7 +119,18 @@ public class SortedQueueEntryListTest ex
private ServerMessage generateTestMessage(final long id, final String keyValue) throws AMQException
{
- return new AMQMessage(new MockStoredMessage(id, "KEY", keyValue));
+ final ServerMessage message = mock(ServerMessage.class);
+ AMQMessageHeader hdr = mock(AMQMessageHeader.class);
+ when(message.getMessageHeader()).thenReturn(hdr);
+ when(hdr.getHeader(eq("KEY"))).thenReturn(keyValue);
+ when(hdr.containsHeader(eq("KEY"))).thenReturn(true);
+ when(hdr.getHeaderNames()).thenReturn(Collections.singleton("KEY"));
+ MessageReference ref = mock(MessageReference.class);
+ when(ref.getMessage()).thenReturn(message);
+ when(message.newReference()).thenReturn(ref);
+ when(message.getMessageNumber()).thenReturn(id);
+
+ return message;
}
public void testIterator()
@@ -132,12 +149,12 @@ public class SortedQueueEntryListTest ex
private Object getSortedKeyValue(QueueEntryIterator<?> iter)
{
- return ((SortedQueueEntryImpl) iter.getNode()).getMessage().getMessageHeader().getHeader("KEY");
+ return (iter.getNode()).getMessage().getMessageHeader().getHeader("KEY");
}
private Long getMessageId(QueueEntryIterator<?> iter)
{
- return ((SortedQueueEntryImpl) iter.getNode()).getMessage().getMessageNumber();
+ return (iter.getNode()).getMessage().getMessageNumber();
}
public void testNonUniqueSortKeys() throws Exception
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java Fri Sep 20 18:59:30 2013
@@ -114,7 +114,7 @@ public class ExternalAuthenticationManag
result.getStatus());
assertOnlyContainsWrapped(expectedPrincipal, result.getPrincipals());
- // Null princial
+ // Null principal
saslServer = _manager.createSaslServer("EXTERNAL", "example.example.com", null);
result = _manager.authenticate(saslServer, new byte[0]);
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java Fri Sep 20 18:59:30 2013
@@ -38,8 +38,6 @@ import java.util.UUID;
import org.apache.commons.configuration.Configuration;
import org.apache.qpid.AMQStoreException;
import org.apache.qpid.common.AMQPFilterTypes;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.binding.Binding;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.message.EnqueableMessage;
@@ -48,15 +46,22 @@ import org.apache.qpid.server.model.Queu
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.MockStoredMessage;
import org.apache.qpid.server.store.MessageStoreRecoveryHandler.StoredMessageRecoveryHandler;
import org.apache.qpid.server.store.Transaction.Record;
import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.util.FileUtils;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
public abstract class AbstractDurableConfigurationStoreTestCase extends QpidTestCase
{
private static final String EXCHANGE_NAME = "exchangeName";
+
+ private static final String EXCHANGE = org.apache.qpid.server.model.Exchange.class.getSimpleName();
+ private static final String BINDING = org.apache.qpid.server.model.Binding.class.getSimpleName();
+ private static final String QUEUE = Queue.class.getSimpleName();
+
private String _storePath;
private String _storeName;
private MessageStore _messageStore;
@@ -73,7 +78,7 @@ public abstract class AbstractDurableCon
private Exchange _exchange = mock(Exchange.class);
private static final String ROUTING_KEY = "routingKey";
private static final String QUEUE_NAME = "queueName";
- private FieldTable _bindingArgs;
+ private Map<String,Object> _bindingArgs;
private UUID _queueId;
private UUID _exchangeId;
private DurableConfigurationStore _configStore;
@@ -101,14 +106,15 @@ public abstract class AbstractDurableCon
when(_messageStoreRecoveryHandler.begin()).thenReturn(_storedMessageRecoveryHandler);
when(_logRecoveryHandler.begin(any(MessageStore.class))).thenReturn(_queueEntryRecoveryHandler);
when(_queueEntryRecoveryHandler.completeQueueEntryRecovery()).thenReturn(_dtxRecordRecoveryHandler);
- when(_exchange.getNameShortString()).thenReturn(AMQShortString.valueOf(EXCHANGE_NAME));
+ when(_exchange.getName()).thenReturn(EXCHANGE_NAME);
+
when(_exchange.getId()).thenReturn(_exchangeId);
when(_configuration.getString(eq(MessageStoreConstants.ENVIRONMENT_PATH_PROPERTY), anyString())).thenReturn(
_storePath);
when(_virtualHost.getAttribute(eq(VirtualHost.STORE_PATH))).thenReturn(_storePath);
- _bindingArgs = new FieldTable();
- AMQShortString argKey = AMQPFilterTypes.JMS_SELECTOR.getValue();
+ _bindingArgs = new HashMap<String, Object>();
+ String argKey = AMQPFilterTypes.JMS_SELECTOR.toString();
String argValue = "some selector expression";
_bindingArgs.put(argKey, argValue);
@@ -117,8 +123,16 @@ public abstract class AbstractDurableCon
public void tearDown() throws Exception
{
- FileUtils.delete(new File(_storePath), true);
- super.tearDown();
+ try
+ {
+ closeMessageStore();
+ closeConfigStore();
+ FileUtils.delete(new File(_storePath), true);
+ }
+ finally
+ {
+ super.tearDown();
+ }
}
public void testCreateExchange() throws Exception
@@ -127,7 +141,7 @@ public abstract class AbstractDurableCon
DurableConfigurationStoreHelper.createExchange(_configStore, exchange);
reopenStore();
- verify(_recoveryHandler).configuredObject(eq(_exchangeId), eq(org.apache.qpid.server.model.Exchange.class.getName()),
+ verify(_recoveryHandler).configuredObject(eq(_exchangeId), eq(EXCHANGE),
eq(map( org.apache.qpid.server.model.Exchange.NAME, getName(),
org.apache.qpid.server.model.Exchange.TYPE, getName()+"Type",
org.apache.qpid.server.model.Exchange.LIFETIME_POLICY, LifetimePolicy.AUTO_DELETE.toString())));
@@ -166,9 +180,9 @@ public abstract class AbstractDurableCon
public void testBindQueue() throws Exception
{
- AMQQueue queue = createTestQueue(QUEUE_NAME, "queueOwner", false);
+ AMQQueue queue = createTestQueue(QUEUE_NAME, "queueOwner", false, null);
Binding binding = new Binding(UUIDGenerator.generateRandomUUID(), ROUTING_KEY, queue,
- _exchange, FieldTable.convertToMap(_bindingArgs));
+ _exchange, _bindingArgs);
DurableConfigurationStoreHelper.createBinding(_configStore, binding);
reopenStore();
@@ -177,49 +191,48 @@ public abstract class AbstractDurableCon
map.put(org.apache.qpid.server.model.Binding.EXCHANGE, _exchange.getId().toString());
map.put(org.apache.qpid.server.model.Binding.QUEUE, queue.getId().toString());
map.put(org.apache.qpid.server.model.Binding.NAME, ROUTING_KEY);
- map.put(org.apache.qpid.server.model.Binding.ARGUMENTS,FieldTable.convertToMap(_bindingArgs));
+ map.put(org.apache.qpid.server.model.Binding.ARGUMENTS,_bindingArgs);
- verify(_recoveryHandler).configuredObject(eq(binding.getId()), eq(org.apache.qpid.server.model.Binding.class.getName()),
+ verify(_recoveryHandler).configuredObject(eq(binding.getId()), eq(BINDING),
eq(map));
}
public void testUnbindQueue() throws Exception
{
- AMQQueue queue = createTestQueue(QUEUE_NAME, "queueOwner", false);
+ AMQQueue queue = createTestQueue(QUEUE_NAME, "queueOwner", false, null);
Binding binding = new Binding(UUIDGenerator.generateRandomUUID(), ROUTING_KEY, queue,
- _exchange, FieldTable.convertToMap(_bindingArgs));
+ _exchange, _bindingArgs);
DurableConfigurationStoreHelper.createBinding(_configStore, binding);
DurableConfigurationStoreHelper.removeBinding(_configStore, binding);
reopenStore();
verify(_recoveryHandler, never()).configuredObject(any(UUID.class),
- eq(org.apache.qpid.server.model.Binding.class.getName()),
+ eq(BINDING),
anyMap());
}
public void testCreateQueueAMQQueue() throws Exception
{
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, null);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, null);
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
reopenStore();
Map<String, Object> queueAttributes = new HashMap<String, Object>();
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
- verify(_recoveryHandler).configuredObject(eq(_queueId), eq(Queue.class.getName()), eq(queueAttributes));
+ verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
public void testCreateQueueAMQQueueFieldTable() throws Exception
{
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true);
Map<String, Object> attributes = new HashMap<String, Object>();
- attributes.put("x-qpid-dlq-enabled", Boolean.TRUE);
- attributes.put("x-qpid-maximum-delivery-count", new Integer(10));
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, Boolean.TRUE);
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 10);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, attributes);
- FieldTable arguments = FieldTable.convertToFieldTable(attributes);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, arguments);
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
reopenStore();
@@ -229,17 +242,17 @@ public abstract class AbstractDurableCon
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
- queueAttributes.put(Queue.ARGUMENTS, attributes);
+ queueAttributes.putAll(attributes);
- verify(_recoveryHandler).configuredObject(eq(_queueId), eq(Queue.class.getName()), eq(queueAttributes));
+ verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
public void testCreateQueueAMQQueueWithAlternateExchange() throws Exception
{
Exchange alternateExchange = createTestAlternateExchange();
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, alternateExchange);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, null);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, alternateExchange, null);
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
reopenStore();
@@ -249,7 +262,7 @@ public abstract class AbstractDurableCon
queueAttributes.put(Queue.EXCLUSIVE, Boolean.TRUE);
queueAttributes.put(Queue.ALTERNATE_EXCHANGE, alternateExchange.getId().toString());
- verify(_recoveryHandler).configuredObject(eq(_queueId), eq(Queue.class.getName()), eq(queueAttributes));
+ verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
private Exchange createTestAlternateExchange()
@@ -263,16 +276,15 @@ public abstract class AbstractDurableCon
public void testUpdateQueueExclusivity() throws Exception
{
// create queue
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true);
Map<String, Object> attributes = new HashMap<String, Object>();
- attributes.put("x-qpid-dlq-enabled", Boolean.TRUE);
- attributes.put("x-qpid-maximum-delivery-count", new Integer(10));
- FieldTable arguments = FieldTable.convertToFieldTable(attributes);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, arguments);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, Boolean.TRUE);
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 10);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, attributes);
+
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
// update the queue to have exclusive=false
- queue = createTestQueue(getName(), getName() + "Owner", false);
- when(queue.getArguments()).thenReturn(attributes);
+ queue = createTestQueue(getName(), getName() + "Owner", false, attributes);
DurableConfigurationStoreHelper.updateQueue(_configStore, queue);
@@ -283,26 +295,24 @@ public abstract class AbstractDurableCon
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
queueAttributes.put(Queue.EXCLUSIVE, Boolean.FALSE);
- queueAttributes.put(Queue.ARGUMENTS, attributes);
+ queueAttributes.putAll(attributes);
- verify(_recoveryHandler).configuredObject(eq(_queueId), eq(Queue.class.getName()), eq(queueAttributes));
+ verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
public void testUpdateQueueAlternateExchange() throws Exception
{
// create queue
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true);
Map<String, Object> attributes = new HashMap<String, Object>();
- attributes.put("x-qpid-dlq-enabled", Boolean.TRUE);
- attributes.put("x-qpid-maximum-delivery-count", new Integer(10));
- FieldTable arguments = FieldTable.convertToFieldTable(attributes);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, arguments);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, Boolean.TRUE);
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 10);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, attributes);
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
// update the queue to have exclusive=false
Exchange alternateExchange = createTestAlternateExchange();
- queue = createTestQueue(getName(), getName() + "Owner", false, alternateExchange);
- when(queue.getArguments()).thenReturn(attributes);
+ queue = createTestQueue(getName(), getName() + "Owner", false, alternateExchange, attributes);
DurableConfigurationStoreHelper.updateQueue(_configStore, queue);
@@ -313,21 +323,20 @@ public abstract class AbstractDurableCon
queueAttributes.put(Queue.NAME, getName());
queueAttributes.put(Queue.OWNER, getName()+"Owner");
queueAttributes.put(Queue.EXCLUSIVE, Boolean.FALSE);
- queueAttributes.put(Queue.ARGUMENTS, attributes);
+ queueAttributes.putAll(attributes);
queueAttributes.put(Queue.ALTERNATE_EXCHANGE, alternateExchange.getId().toString());
- verify(_recoveryHandler).configuredObject(eq(_queueId), eq(Queue.class.getName()), eq(queueAttributes));
+ verify(_recoveryHandler).configuredObject(eq(_queueId), eq(QUEUE), eq(queueAttributes));
}
public void testRemoveQueue() throws Exception
{
// create queue
- AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true);
Map<String, Object> attributes = new HashMap<String, Object>();
- attributes.put("x-qpid-dlq-enabled", Boolean.TRUE);
- attributes.put("x-qpid-maximum-delivery-count", new Integer(10));
- FieldTable arguments = FieldTable.convertToFieldTable(attributes);
- DurableConfigurationStoreHelper.createQueue(_configStore, queue, arguments);
+ attributes.put(Queue.CREATE_DLQ_ON_CREATION, Boolean.TRUE);
+ attributes.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, 10);
+ AMQQueue queue = createTestQueue(getName(), getName() + "Owner", true, attributes);
+ DurableConfigurationStoreHelper.createQueue(_configStore, queue);
// remove queue
DurableConfigurationStoreHelper.removeQueue(_configStore,queue);
@@ -337,29 +346,51 @@ public abstract class AbstractDurableCon
anyMap());
}
- private AMQQueue createTestQueue(String queueName, String queueOwner, boolean exclusive) throws AMQStoreException
- {
- return createTestQueue(queueName, queueOwner, exclusive, null);
+ private AMQQueue createTestQueue(String queueName,
+ String queueOwner,
+ boolean exclusive,
+ final Map<String, Object> arguments) throws AMQStoreException
+ {
+ return createTestQueue(queueName, queueOwner, exclusive, null, arguments);
}
- private AMQQueue createTestQueue(String queueName, String queueOwner, boolean exclusive, Exchange alternateExchange) throws AMQStoreException
+ private AMQQueue createTestQueue(String queueName,
+ String queueOwner,
+ boolean exclusive,
+ Exchange alternateExchange,
+ final Map<String, Object> arguments) throws AMQStoreException
{
AMQQueue queue = mock(AMQQueue.class);
when(queue.getName()).thenReturn(queueName);
- when(queue.getNameShortString()).thenReturn(AMQShortString.valueOf(queueName));
- when(queue.getOwner()).thenReturn(AMQShortString.valueOf(queueOwner));
+ when(queue.getOwner()).thenReturn(queueOwner);
when(queue.isExclusive()).thenReturn(exclusive);
when(queue.getId()).thenReturn(_queueId);
when(queue.getAlternateExchange()).thenReturn(alternateExchange);
+ if(arguments != null && !arguments.isEmpty())
+ {
+ when(queue.getAvailableAttributes()).thenReturn(arguments.keySet());
+ final ArgumentCaptor<String> requestedAttribute = ArgumentCaptor.forClass(String.class);
+ when(queue.getAttribute(requestedAttribute.capture())).then(
+ new Answer()
+ {
+
+ @Override
+ public Object answer(final InvocationOnMock invocation) throws Throwable
+ {
+ String attrName = requestedAttribute.getValue();
+ return arguments.get(attrName);
+ }
+ });
+ }
+
return queue;
}
private Exchange createTestExchange()
{
Exchange exchange = mock(Exchange.class);
- when(exchange.getNameShortString()).thenReturn(AMQShortString.valueOf(getName()));
when(exchange.getName()).thenReturn(getName());
- when(exchange.getTypeShortString()).thenReturn(AMQShortString.valueOf(getName() + "Type"));
+ when(exchange.getTypeName()).thenReturn(getName() + "Type");
when(exchange.isAutoDelete()).thenReturn(true);
when(exchange.getId()).thenReturn(_exchangeId);
return exchange;
@@ -367,53 +398,21 @@ public abstract class AbstractDurableCon
private void reopenStore() throws Exception
{
- onReopenStore();
- if (_messageStore != null)
- {
- _messageStore.close();
- }
+ closeMessageStore();
+ closeConfigStore();
_messageStore = createMessageStore();
_configStore = createConfigStore();
- _configStore.configureConfigStore(_storeName, _recoveryHandler, _virtualHost);
- _messageStore.configureMessageStore(_storeName, _messageStoreRecoveryHandler, _logRecoveryHandler);
+ _configStore.configureConfigStore(_virtualHost, _recoveryHandler);
+ _messageStore.configureMessageStore(_virtualHost, _messageStoreRecoveryHandler, _logRecoveryHandler);
_messageStore.activate();
}
- protected abstract void onReopenStore();
+ protected abstract MessageStore createMessageStore() throws Exception;
+ protected abstract DurableConfigurationStore createConfigStore() throws Exception;
+ protected abstract void closeMessageStore() throws Exception;
+ protected abstract void closeConfigStore() throws Exception;
- abstract protected MessageStore createMessageStore() throws Exception;
- /*{
- String storeClass = System.getProperty(MESSAGE_STORE_CLASS_NAME_KEY);
- if (storeClass == null)
- {
- storeClass = DerbyMessageStore.class.getName();
- }
- CurrentActor.set(new TestLogActor(new SystemOutMessageLogger()));
- MessageStore messageStore = (MessageStore) Class.forName(storeClass).newInstance();
- return messageStore;
- }
-*/
- abstract protected DurableConfigurationStore createConfigStore() throws Exception;
- /*{
- String storeClass = System.getProperty(CONFIGURATION_STORE_CLASS_NAME_KEY);
- if (storeClass == null)
- {
- storeClass = DerbyMessageStore.class.getName();
- }
- Class<DurableConfigurationStore> clazz = (Class<DurableConfigurationStore>) Class.forName(storeClass);
- DurableConfigurationStore configurationStore ;
- if(clazz.isInstance(_messageStore))
- {
- configurationStore = (DurableConfigurationStore) _messageStore;
- }
- else
- {
- configurationStore = (DurableConfigurationStore) Class.forName(storeClass).newInstance();
- }
- return configurationStore;
- }
-*/
public void testRecordXid() throws Exception
{
Record enqueueRecord = getTestRecord(1);
@@ -445,7 +444,9 @@ public abstract class AbstractDurableCon
EnqueableMessage message1 = mock(EnqueableMessage.class);
when(message1.isPersistent()).thenReturn(true);
when(message1.getMessageNumber()).thenReturn(messageNumber);
- when(message1.getStoredMessage()).thenReturn(new MockStoredMessage(messageNumber));
+ final StoredMessage storedMessage = mock(StoredMessage.class);
+ when(storedMessage.getMessageNumber()).thenReturn(messageNumber);
+ when(message1.getStoredMessage()).thenReturn(storedMessage);
Record enqueueRecord = new TestRecord(queue1, message1);
return enqueueRecord;
}
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java Fri Sep 20 18:59:30 2013
@@ -27,15 +27,8 @@ import java.util.List;
import java.util.UUID;
import org.apache.log4j.Logger;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.framing.abstraction.MessagePublishInfoImpl;
import org.apache.qpid.server.message.EnqueableMessage;
-import org.apache.qpid.server.message.MessageMetaData;
+import org.apache.qpid.server.plugin.MessageMetaDataType;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.util.FileUtils;
@@ -73,10 +66,13 @@ public abstract class MessageStoreQuotaE
VirtualHost vhost = mock(VirtualHost.class);
when(vhost.getAttribute(eq(VirtualHost.STORE_PATH))).thenReturn(_storeLocation.getAbsolutePath());
+ when(vhost.getName()).thenReturn("test");
+
applyStoreSpecificConfiguration(vhost);
_store = createStore();
- ((DurableConfigurationStore)_store).configureConfigStore("test", null, vhost);
+ ((DurableConfigurationStore)_store).configureConfigStore(vhost, null);
+ _store.configureMessageStore(vhost, null, null);
_transactionResource = UUID.randomUUID();
_events = new ArrayList<Event>();
@@ -86,8 +82,18 @@ public abstract class MessageStoreQuotaE
@Override
public void tearDown() throws Exception
{
- super.tearDown();
- FileUtils.delete(_storeLocation, true);
+ try
+ {
+ super.tearDown();
+ }
+ finally
+ {
+ if (_store != null)
+ {
+ _store.close();
+ }
+ FileUtils.delete(_storeLocation, true);
+ }
}
public void testOverflow() throws Exception
@@ -115,23 +121,25 @@ public abstract class MessageStoreQuotaE
protected EnqueableMessage addMessage(long id)
{
- MessagePublishInfo pubInfoBody = new MessagePublishInfoImpl(new AMQShortString(getName()), false, false,
- new AMQShortString(getName()));
- BasicContentHeaderProperties props = new BasicContentHeaderProperties();
- props.setDeliveryMode(Integer.valueOf(BasicContentHeaderProperties.PERSISTENT).byteValue());
- props.setContentType(getTestName());
-
- MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
- int classForBasic = methodRegistry.createBasicQosOkBody().getClazz();
- ContentHeaderBody contentHeaderBody = new ContentHeaderBody(classForBasic, 1, props, MESSAGE_DATA.length);
-
- MessageMetaData metaData = new MessageMetaData(pubInfoBody, contentHeaderBody, 1);
- StoredMessage<MessageMetaData> handle = _store.addMessage(metaData);
+ StorableMessageMetaData metaData = createMetaData(id, MESSAGE_DATA.length);
+ StoredMessage handle = _store.addMessage(metaData);
handle.addContent(0, ByteBuffer.wrap(MESSAGE_DATA));
TestMessage message = new TestMessage(id, handle);
return message;
}
+ private StorableMessageMetaData createMetaData(long id, int length)
+ {
+ StorableMessageMetaData metaData = mock(StorableMessageMetaData.class);
+ when(metaData.isPersistent()).thenReturn(true);
+ when(metaData.getContentSize()).thenReturn(length);
+ when(metaData.getStorableSize()).thenReturn(0);
+ MessageMetaDataType type = mock(MessageMetaDataType.class);
+ when(type.ordinal()).thenReturn(-1);
+ when(metaData.getType()).thenReturn(type);
+ return metaData;
+ }
+
@Override
public void event(Event event)
{
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java Fri Sep 20 18:59:30 2013
@@ -21,7 +21,6 @@
package org.apache.qpid.server.store;
import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -53,16 +52,18 @@ public abstract class MessageStoreTestCa
_dtxRecordRecoveryHandler = mock(TransactionLogRecoveryHandler.DtxRecordRecoveryHandler.class);
_virtualHost = mock(VirtualHost.class);
+
when(_messageStoreRecoveryHandler.begin()).thenReturn(_storedMessageRecoveryHandler);
when(_logRecoveryHandler.begin(any(MessageStore.class))).thenReturn(_queueEntryRecoveryHandler);
when(_queueEntryRecoveryHandler.completeQueueEntryRecovery()).thenReturn(_dtxRecordRecoveryHandler);
setUpStoreConfiguration(_virtualHost);
+ when(_virtualHost.getName()).thenReturn(getTestName());
_store = createMessageStore();
- ((DurableConfigurationStore)_store).configureConfigStore(getTestName(), _recoveryHandler, _virtualHost);
+ ((DurableConfigurationStore)_store).configureConfigStore(_virtualHost, _recoveryHandler);
- _store.configureMessageStore(getTestName(), _messageStoreRecoveryHandler, _logRecoveryHandler);
+ _store.configureMessageStore(_virtualHost, _messageStoreRecoveryHandler, _logRecoveryHandler);
}
protected abstract void setUpStoreConfiguration(VirtualHost virtualHost) throws Exception;
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java Fri Sep 20 18:59:30 2013
@@ -32,7 +32,7 @@ import java.util.concurrent.atomic.Atomi
/**
* Adds some extra methods to the memory message store for testing purposes.
*/
-public class TestableMemoryMessageStore extends MemoryMessageStore
+public class TestableMemoryMessageStore extends TestMemoryMessageStore
{
private final Map<Long, AMQQueue> _messages = new HashMap<Long, AMQQueue>();
private final AtomicInteger _messageCount = new AtomicInteger(0);
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java Fri Sep 20 18:59:30 2013
@@ -47,7 +47,7 @@ public class MockSubscription implements
{
private boolean _closed = false;
- private AMQShortString tag = new AMQShortString("mocktag");
+ private String tag = "mocktag";
private AMQQueue queue = null;
private StateListener _listener = null;
private volatile AMQQueue.Context _queueContext = null;
@@ -84,7 +84,7 @@ public class MockSubscription implements
public String getConsumerName()
{
- return tag == null ? null : tag.asString();
+ return tag;
}
public long getSubscriptionID()
@@ -569,5 +569,11 @@ public class MockSubscription implements
{
return false;
}
+
+ @Override
+ public String getVirtualHostName()
+ {
+ return null;
+ }
}
}
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java Fri Sep 20 18:59:30 2013
@@ -31,11 +31,8 @@ import java.util.UUID;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.configuration.store.JsonConfigurationEntryStore;
import org.apache.qpid.server.exchange.DefaultExchangeFactory;
@@ -47,8 +44,6 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.actors.TestLogActor;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.UUIDGenerator;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.protocol.InternalTestProtocolSession;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.queue.SimpleAMQQueue;
import org.apache.qpid.server.security.SecurityManager;
@@ -143,33 +138,35 @@ public class BrokerTestHelper
return vhostConfig;
}
- public static AMQChannel createChannel(int channelId, AMQProtocolSession session) throws AMQException
+ public static AMQSessionModel createSession(int channelId, AMQConnectionModel connection) throws AMQException
{
- AMQChannel channel = new AMQChannel(session, channelId, session.getVirtualHost().getMessageStore());
- session.addChannel(channel);
- return channel;
+ AMQSessionModel session = mock(AMQSessionModel.class);
+ when(session.getConnectionModel()).thenReturn(connection);
+ when(session.getChannelId()).thenReturn(channelId);
+ return session;
}
- public static AMQChannel createChannel(int channelId) throws Exception
+ public static AMQSessionModel createSession(int channelId) throws Exception
{
- InternalTestProtocolSession session = createSession();
- return createChannel(channelId, session);
+ AMQConnectionModel session = createConnection();
+ return createSession(channelId, session);
}
- public static AMQChannel createChannel() throws Exception
+ public static AMQSessionModel createSession() throws Exception
{
- return createChannel(1);
+ return createSession(1);
}
- public static InternalTestProtocolSession createSession() throws Exception
+ public static AMQConnectionModel createConnection() throws Exception
{
- return createSession("test");
+ return createConnection("test");
}
- public static InternalTestProtocolSession createSession(String hostName) throws Exception
+ public static AMQConnectionModel createConnection(String hostName) throws Exception
{
VirtualHost virtualHost = createVirtualHost(hostName);
- return new InternalTestProtocolSession(virtualHost, createBrokerMock());
+ AMQConnectionModel connection = mock(AMQConnectionModel.class);
+ return connection;
}
public static Exchange createExchange(String hostName) throws Exception
@@ -182,44 +179,10 @@ public class BrokerTestHelper
return factory.createExchange("amp.direct", "direct", false, false);
}
- public static void publishMessages(AMQChannel channel, int numberOfMessages, String queueName, String exchangeName) throws AMQException
- {
- AMQShortString rouningKey = new AMQShortString(queueName);
- AMQShortString exchangeNameAsShortString = new AMQShortString(exchangeName);
- MessagePublishInfo info = mock(MessagePublishInfo.class);
- when(info.getExchange()).thenReturn(exchangeNameAsShortString);
- when(info.getRoutingKey()).thenReturn(rouningKey);
-
- Exchange exchange = channel.getVirtualHost().getExchange(exchangeName);
- for (int count = 0; count < numberOfMessages; count++)
- {
- channel.setPublishFrame(info, exchange);
-
- // Set the body size
- ContentHeaderBody _headerBody = new ContentHeaderBody();
- _headerBody.setBodySize(0);
-
- // Set Minimum properties
- BasicContentHeaderProperties properties = new BasicContentHeaderProperties();
-
- properties.setExpiration(0L);
- properties.setTimestamp(System.currentTimeMillis());
-
- // Make Message Persistent
- properties.setDeliveryMode((byte) 2);
-
- _headerBody.setProperties(properties);
-
- channel.publishContentHeader(_headerBody);
- }
- channel.sync();
- }
-
public static SimpleAMQQueue createQueue(String queueName, VirtualHost virtualHost) throws AMQException
{
- SimpleAMQQueue queue = (SimpleAMQQueue) AMQQueueFactory.createAMQQueueImpl(UUIDGenerator.generateRandomUUID(), queueName, false, null,
- false, false, virtualHost, Collections.<String, Object>emptyMap());
- virtualHost.getQueueRegistry().registerQueue(queue);
+ SimpleAMQQueue queue = (SimpleAMQQueue) virtualHost.createQueue(UUIDGenerator.generateRandomUUID(), queueName, false, null,
+ false, false, false, Collections.<String, Object>emptyMap());
return queue;
}
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java Fri Sep 20 18:59:30 2013
@@ -21,16 +21,15 @@
package org.apache.qpid.server.virtualhost;
import java.util.Collection;
+import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.connection.IConnectionRegistry;
-import org.apache.qpid.server.exchange.AbstractExchange;
import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.exchange.ExchangeFactory;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.plugin.ExchangeType;
-import org.apache.qpid.server.protocol.v1_0.LinkRegistry;
+import org.apache.qpid.server.protocol.LinkRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
@@ -122,6 +121,43 @@ public class MockVirtualHost implements
}
@Override
+ public AMQQueue getQueue(String name)
+ {
+ return null;
+ }
+
+ @Override
+ public AMQQueue getQueue(UUID id)
+ {
+ return null;
+ }
+
+ @Override
+ public Collection<AMQQueue> getQueues()
+ {
+ return null;
+ }
+
+ @Override
+ public int removeQueue(AMQQueue queue) throws AMQException
+ {
+ return 0;
+ }
+
+ @Override
+ public AMQQueue createQueue(UUID id,
+ String queueName,
+ boolean durable,
+ String owner,
+ boolean autoDelete,
+ boolean exclusive,
+ boolean deleteOnNoConsumer,
+ Map<String, Object> arguments) throws AMQException
+ {
+ return null;
+ }
+
+ @Override
public Exchange createExchange(UUID id,
String exchange,
String type,
@@ -144,6 +180,12 @@ public class MockVirtualHost implements
}
@Override
+ public Exchange getExchange(UUID id)
+ {
+ return null;
+ }
+
+ @Override
public Exchange getDefaultExchange()
{
return null;
Modified: qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/StandardVirtualHostTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/StandardVirtualHostTest.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/StandardVirtualHostTest.java (original)
+++ qpid/branches/linearstore/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/StandardVirtualHostTest.java Fri Sep 20 18:59:30 2013
@@ -20,7 +20,9 @@
*/
package org.apache.qpid.server.virtualhost;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
@@ -34,7 +36,7 @@ import org.apache.qpid.server.model.Brok
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.stats.StatisticsGatherer;
-import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.store.TestMemoryMessageStore;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -242,7 +244,7 @@ public class StandardVirtualHostTest ext
VirtualHost vhost = createVirtualHost(vhostName, config);
assertNotNull("virtualhost should exist", vhost);
- AMQQueue queue = vhost.getQueueRegistry().getQueue(queueName);
+ AMQQueue queue = vhost.getQueue(queueName);
assertNotNull("queue should exist", queue);
Exchange defaultExch = vhost.getDefaultExchange();
@@ -306,7 +308,7 @@ public class StandardVirtualHostTest ext
writer.write(" <" + vhostName + ">");
writer.write(" <type>" + StandardVirtualHostFactory.TYPE + "</type>");
writer.write(" <store>");
- writer.write(" <class>" + MemoryMessageStore.class.getName() + "</class>");
+ writer.write(" <class>" + TestMemoryMessageStore.class.getName() + "</class>");
writer.write(" </store>");
if(exchangeName != null && !dontDeclare)
{
@@ -363,10 +365,11 @@ public class StandardVirtualHostTest ext
_virtualHostRegistry = broker.getVirtualHostRegistry();
Configuration config = new PropertiesConfiguration();
- config.setProperty("store.type", MemoryMessageStore.TYPE);
VirtualHostConfiguration configuration = new VirtualHostConfiguration(virtualHostName, config, broker);
+ final org.apache.qpid.server.model.VirtualHost virtualHost = mock(org.apache.qpid.server.model.VirtualHost.class);
+ when(virtualHost.getAttribute(eq(org.apache.qpid.server.model.VirtualHost.STORE_TYPE))).thenReturn(TestMemoryMessageStore.TYPE);
VirtualHost host = new StandardVirtualHostFactory().createVirtualHost(_virtualHostRegistry, mock(StatisticsGatherer.class), new SecurityManager(mock(Broker.class), false), configuration,
- mock(org.apache.qpid.server.model.VirtualHost.class));
+ virtualHost);
_virtualHostRegistry.registerVirtualHost(host);
return host;
}
Modified: qpid/branches/linearstore/qpid/java/build.deps
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/build.deps?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/build.deps (original)
+++ qpid/branches/linearstore/qpid/java/build.deps Fri Sep 20 18:59:30 2013
@@ -98,6 +98,10 @@ broker-plugins-management-http.test.libs
broker-plugins-management-jmx.test.libs=${test.libs}
broker-plugins-jdbc-store.test.libs=${test.libs}
broker-plugins-derby-store.test.libs=${test.libs}
+broker-plugins-memory-store.test.libs=${test.libs}
+broker-plugins-amqp-0-8-protocol.test.libs=${test.libs}
+broker-plugins-amqp-0-10-protocol.test.libs=${test.libs}
+broker-plugins-amqp-1-0-protocol.test.libs=${test.libs}
management-common.test.libs=${test.libs}
@@ -106,7 +110,7 @@ jca.libs=${geronimo-j2ee} ${geronimo-jta
jca.test.libs=${test.libs}
# optional bdbstore module deps
-bdb-je=lib/bdbstore/je-5.0.73.jar
+bdb-je=lib/bdbstore/je-5.0.84.jar
bdbstore.libs=${bdb-je}
bdbstore.test.libs=${test.libs}
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/client.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/client.bnd?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/client.bnd (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/client.bnd Fri Sep 20 18:59:30 2013
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.23.0
+ver: 0.25.0
Bundle-SymbolicName: qpid-client
Bundle-Version: ${ver}
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java Fri Sep 20 18:59:30 2013
@@ -126,7 +126,8 @@ public class AMQConnection extends Close
/** The virtual path to connect to on the AMQ server */
private String _virtualHost;
- private ExceptionListener _exceptionListener;
+ /** The exception listener for this connection object. */
+ private volatile ExceptionListener _exceptionListener;
private ConnectionListener _connectionListener;
@@ -151,10 +152,10 @@ public class AMQConnection extends Close
*/
private QpidConnectionMetaData _connectionMetaData;
- private AMQShortString _defaultTopicExchangeName = ExchangeDefaults.TOPIC_EXCHANGE_NAME;
- private AMQShortString _defaultQueueExchangeName = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
- private AMQShortString _temporaryTopicExchangeName = ExchangeDefaults.TOPIC_EXCHANGE_NAME;
- private AMQShortString _temporaryQueueExchangeName = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
+ private AMQShortString _defaultTopicExchangeName = AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_NAME);
+ private AMQShortString _defaultQueueExchangeName = AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_NAME);
+ private AMQShortString _temporaryTopicExchangeName = AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_NAME);
+ private AMQShortString _temporaryQueueExchangeName = AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_NAME);
/** Thread Pool for executing connection level processes. Such as returning bounced messages. */
private final ExecutorService _taskPool = Executors.newCachedThreadPool();
@@ -784,13 +785,13 @@ public class AMQConnection extends Close
public ExceptionListener getExceptionListener() throws JMSException
{
checkNotClosed();
-
- return _exceptionListener;
+ return getExceptionListenerNoCheck();
}
public void setExceptionListener(ExceptionListener listener) throws JMSException
{
checkNotClosed();
+
_exceptionListener = listener;
}
@@ -1307,45 +1308,56 @@ public class AMQConnection extends Close
_protocolHandler.getProtocolSession().notifyError(je);
}
- // get the failover mutex before trying to close
- synchronized (getFailoverMutex())
+ try
{
- // decide if we are going to close the session
- if (hardError(cause))
+ // get the failover mutex before trying to close
+ synchronized (getFailoverMutex())
{
- closer = (!setClosed()) || closer;
+ // decide if we are going to close the session
+ if (hardError(cause))
{
- _logger.info("Closing AMQConnection due to :" + cause);
+ closer = (!setClosed()) || closer;
+ {
+ _logger.info("Closing AMQConnection due to :" + cause);
+ }
}
- }
- else
- {
- _logger.info("Not a hard-error connection not closing: " + cause);
- }
-
- // deliver the exception if there is a listener
- if (_exceptionListener != null)
- {
- _exceptionListener.onException(je);
- }
- else
- {
- _logger.error("Throwable Received but no listener set: " + cause);
- }
-
- // if we are closing the connection, close sessions first
- if (closer)
- {
- try
+ else
{
- closeAllSessions(cause, -1, -1); // FIXME: when doing this end up with RejectedExecutionException from executor.
+ _logger.info("Not a hard-error connection not closing: " + cause);
}
- catch (JMSException e)
+
+ // if we are closing the connection, close sessions first
+ if (closer)
{
- _logger.error("Error closing all sessions: " + e, e);
+ try
+ {
+ closeAllSessions(cause, -1, -1); // FIXME: when doing this end up with RejectedExecutionException from executor.
+ }
+ catch (JMSException e)
+ {
+ _logger.error("Error closing all sessions: " + e, e);
+ }
}
}
}
+ finally
+ {
+ deliverJMSExceptionToExceptionListenerOrLog(je, cause);
+ }
+ }
+
+ private void deliverJMSExceptionToExceptionListenerOrLog(final JMSException je, final Throwable cause)
+ {
+ // deliver the exception if there is a listener
+ ExceptionListener exceptionListener = getExceptionListenerNoCheck();
+ if (exceptionListener != null)
+ {
+ exceptionListener.onException(je);
+ }
+ else
+ {
+ _logger.error("Throwable Received but no listener set: " + cause);
+ }
}
private boolean hardError(Throwable cause)
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java Fri Sep 20 18:59:30 2013
@@ -47,7 +47,7 @@ import java.util.concurrent.atomic.Atomi
public abstract class AMQDestination implements Destination, Referenceable
{
private static final Logger _logger = LoggerFactory.getLogger(AMQDestination.class);
-
+
private AMQShortString _exchangeName;
private AMQShortString _exchangeClass;
@@ -65,7 +65,7 @@ public abstract class AMQDestination imp
private boolean _isAutoDelete;
private boolean _browseOnly;
-
+
private AtomicLong _addressResolved = new AtomicLong(0);
private AMQShortString _queueName;
@@ -113,10 +113,10 @@ public abstract class AMQDestination imp
}
// ----- Fields required to support new address syntax -------
-
- public enum DestSyntax {
+
+ public enum DestSyntax {
BURL,ADDR;
-
+
public static DestSyntax getSyntaxType(String s)
{
if (("BURL").equals(s))
@@ -133,11 +133,11 @@ public abstract class AMQDestination imp
" should be one of {BURL|ADDR}");
}
}
- }
-
- public enum AddressOption {
- ALWAYS, NEVER, SENDER, RECEIVER;
-
+ }
+
+ public enum AddressOption {
+ ALWAYS, NEVER, SENDER, RECEIVER;
+
public static AddressOption getOption(String str)
{
if ("always".equals(str))
@@ -162,9 +162,9 @@ public abstract class AMQDestination imp
}
}
}
-
+
private final static DestSyntax defaultDestSyntax;
-
+
private DestSyntax _destSyntax = DestSyntax.ADDR;
private AddressHelper _addrHelper;
@@ -179,25 +179,25 @@ public abstract class AMQDestination imp
private Node _node;
private Link _link;
-
+
// ----- / Fields required to support new address syntax -------
-
+
static
{
defaultDestSyntax = DestSyntax.getSyntaxType(
System.getProperty(ClientProperties.DEST_SYNTAX,
DestSyntax.ADDR.toString()));
-
-
+
+
}
-
+
public static DestSyntax getDefaultDestSyntax()
{
return defaultDestSyntax;
}
protected AMQDestination()
- {
+ {
}
protected AMQDestination(Address address) throws Exception
@@ -207,10 +207,10 @@ public abstract class AMQDestination imp
_destSyntax = DestSyntax.ADDR;
_logger.debug("Based on " + address + " the selected destination syntax is " + _destSyntax);
}
-
+
public static DestSyntax getDestType(String str)
{
- if (str.startsWith("BURL:") ||
+ if (str.startsWith("BURL:") ||
(!str.startsWith("ADDR:") && defaultDestSyntax == DestSyntax.BURL))
{
return DestSyntax.BURL;
@@ -220,7 +220,7 @@ public abstract class AMQDestination imp
return DestSyntax.ADDR;
}
}
-
+
public static String stripSyntaxPrefix(String str)
{
if (str.startsWith("BURL:") || str.startsWith("ADDR:"))
@@ -232,7 +232,7 @@ public abstract class AMQDestination imp
return str;
}
}
-
+
protected AMQDestination(String str) throws URISyntaxException
{
parseDestinationString(str);
@@ -243,8 +243,8 @@ public abstract class AMQDestination imp
_destSyntax = getDestType(str);
str = stripSyntaxPrefix(str);
if (_destSyntax == DestSyntax.BURL)
- {
- getInfoFromBindingURL(new AMQBindingURL(str));
+ {
+ getInfoFromBindingURL(new AMQBindingURL(str));
}
else
{
@@ -262,7 +262,7 @@ public abstract class AMQDestination imp
}
_logger.debug("Based on " + str + " the selected destination syntax is " + _destSyntax);
}
-
+
//retained for legacy support
protected AMQDestination(BindingURL binding)
{
@@ -331,8 +331,8 @@ public abstract class AMQDestination imp
protected AMQDestination(AMQShortString exchangeName, AMQShortString exchangeClass, AMQShortString routingKey, boolean isExclusive,
boolean isAutoDelete, AMQShortString queueName, boolean isDurable,AMQShortString[] bindingKeys, boolean browseOnly)
{
- if ( (ExchangeDefaults.DIRECT_EXCHANGE_CLASS.equals(exchangeClass) ||
- ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(exchangeClass))
+ if ( (AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(exchangeClass) ||
+ AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS).equals(exchangeClass))
&& routingKey == null)
{
throw new IllegalArgumentException("routing/binding key must not be null");
@@ -376,16 +376,16 @@ public abstract class AMQDestination imp
return toString();
}
- public DestSyntax getDestSyntax()
+ public DestSyntax getDestSyntax()
{
return _destSyntax;
}
-
+
protected void setDestSyntax(DestSyntax syntax)
{
_destSyntax = syntax;
}
-
+
public AMQShortString getEncodedName()
{
if(_urlAsShortString == null)
@@ -431,12 +431,12 @@ public abstract class AMQDestination imp
public boolean isTopic()
{
- return ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(_exchangeClass);
+ return AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS).equals(_exchangeClass);
}
public boolean isQueue()
{
- return ExchangeDefaults.DIRECT_EXCHANGE_CLASS.equals(_exchangeClass);
+ return AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(_exchangeClass);
}
public String getQueueName()
@@ -481,7 +481,7 @@ public abstract class AMQDestination imp
{
return _isExclusive;
}
-
+
public boolean isAutoDelete()
{
return _isAutoDelete;
@@ -720,15 +720,15 @@ public abstract class AMQDestination imp
{
AMQShortString type = binding.getExchangeClass();
- if (type.equals(ExchangeDefaults.DIRECT_EXCHANGE_CLASS))
+ if (type.equals(AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS)))
{
return new AMQQueue(binding);
}
- else if (type.equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS))
+ else if (type.equals(AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS)))
{
return new AMQTopic(binding);
}
- else if (type.equals(ExchangeDefaults.HEADERS_EXCHANGE_CLASS))
+ else if (type.equals(AMQShortString.valueOf(ExchangeDefaults.HEADERS_EXCHANGE_CLASS)))
{
return new AMQHeadersExchange(binding);
}
@@ -743,8 +743,8 @@ public abstract class AMQDestination imp
DestSyntax syntax = getDestType(str);
str = stripSyntaxPrefix(str);
if (syntax == DestSyntax.BURL)
- {
- return createDestination(new AMQBindingURL(str));
+ {
+ return createDestination(new AMQBindingURL(str));
}
else
{
@@ -752,16 +752,16 @@ public abstract class AMQDestination imp
return new AMQAnyDestination(address);
}
}
-
+
// ----- new address syntax -----------
-
+
public static class Binding
{
private String exchange;
private String bindingKey;
private String queue;
private Map<String,Object> args;
-
+
public Binding(String exchange,
String queue,
String bindingKey,
@@ -772,36 +772,36 @@ public abstract class AMQDestination imp
this.bindingKey = bindingKey;
this.args = args;
}
-
- public String getExchange()
+
+ public String getExchange()
{
return exchange;
}
- public String getQueue()
+ public String getQueue()
{
return queue;
}
-
- public String getBindingKey()
+
+ public String getBindingKey()
{
return bindingKey;
}
-
- public Map<String, Object> getArgs()
+
+ public Map<String, Object> getArgs()
{
return args;
}
}
-
+
public Address getAddress() {
return _address;
}
-
+
protected void setAddress(Address addr) {
_address = addr;
}
-
+
public int getAddressType(){
return _addressType;
}
@@ -809,11 +809,11 @@ public abstract class AMQDestination imp
public void setAddressType(int addressType){
_addressType = addressType;
}
-
+
public String getAddressName() {
return _name;
}
-
+
public void setAddressName(String name){
_name = name;
}
@@ -825,15 +825,15 @@ public abstract class AMQDestination imp
public void setSubject(String subject) {
_subject = subject;
}
-
+
public AddressOption getCreate() {
return _create;
}
-
+
public void setCreate(AddressOption option) {
_create = option;
}
-
+
public AddressOption getAssert() {
return _assert;
}
@@ -841,7 +841,7 @@ public abstract class AMQDestination imp
public void setAssert(AddressOption option) {
_assert = option;
}
-
+
public AddressOption getDelete() {
return _delete;
}
@@ -869,22 +869,22 @@ public abstract class AMQDestination imp
{
_link = link;
}
-
+
public void setExchangeName(AMQShortString name)
{
this._exchangeName = name;
}
-
+
public void setExchangeClass(AMQShortString type)
{
this._exchangeClass = type;
}
-
+
public void setRoutingKey(AMQShortString rk)
{
this._routingKey = rk;
}
-
+
public boolean isAddressResolved()
{
return _addressResolved.get() > 0;
@@ -894,80 +894,80 @@ public abstract class AMQDestination imp
{
_addressResolved.set(addressResolved);
}
-
+
private static Address createAddressFromString(String str)
{
return Address.parse(str);
}
-
+
private void getInfoFromAddress() throws Exception
{
_name = _address.getName();
_subject = _address.getSubject();
-
+
_addrHelper = new AddressHelper(_address);
-
+
_create = _addrHelper.getCreate() != null ?
AddressOption.getOption(_addrHelper.getCreate()):AddressOption.NEVER;
-
+
_assert = _addrHelper.getAssert() != null ?
AddressOption.getOption(_addrHelper.getAssert()):AddressOption.NEVER;
_delete = _addrHelper.getDelete() != null ?
AddressOption.getOption(_addrHelper.getDelete()):AddressOption.NEVER;
-
+
_browseOnly = _addrHelper.isBrowseOnly();
-
+
_addressType = _addrHelper.getNodeType();
_node = _addrHelper.getNode();
- _link = _addrHelper.getLink();
+ _link = _addrHelper.getLink();
}
-
- // ----- / new address syntax -----------
+
+ // ----- / new address syntax -----------
public boolean isBrowseOnly()
{
return _browseOnly;
}
-
+
private void setBrowseOnly(boolean b)
{
_browseOnly = b;
}
-
+
public AMQDestination copyDestination()
{
- AMQDestination dest =
+ AMQDestination dest =
new AMQAnyDestination(_exchangeName,
_exchangeClass,
_routingKey,
- _isExclusive,
+ _isExclusive,
_isAutoDelete,
- _queueName,
+ _queueName,
_isDurable,
_bindingKeys
);
-
+
dest.setDestSyntax(_destSyntax);
dest.setAddress(_address);
dest.setAddressName(_name);
dest.setSubject(_subject);
- dest.setCreate(_create);
- dest.setAssert(_assert);
- dest.setDelete(_delete);
+ dest.setCreate(_create);
+ dest.setAssert(_assert);
+ dest.setDelete(_delete);
dest.setBrowseOnly(_browseOnly);
dest.setAddressType(_addressType);
dest.setNode(_node);
dest.setLink(_link);
dest.setAddressResolved(_addressResolved.get());
- return dest;
+ return dest;
}
-
+
protected void setAutoDelete(boolean b)
{
_isAutoDelete = b;
}
-
+
protected void setDurable(boolean b)
{
_isDurable = b;
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQHeadersExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQHeadersExchange.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQHeadersExchange.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQHeadersExchange.java Fri Sep 20 18:59:30 2013
@@ -41,7 +41,7 @@ public class AMQHeadersExchange extends
public AMQHeadersExchange(AMQShortString queueName)
{
- super(queueName, ExchangeDefaults.HEADERS_EXCHANGE_CLASS, queueName, true, true, null);
+ super(queueName, AMQShortString.valueOf(ExchangeDefaults.HEADERS_EXCHANGE_CLASS), queueName, true, true, null);
}
public boolean isNameRequired()
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java Fri Sep 20 18:59:30 2013
@@ -38,7 +38,7 @@ public class AMQQueue extends AMQDestina
{
super(address);
}
-
+
/**
* Create a reference to a non temporary queue using a BindingURL object.
* Note this does not actually imply the queue exists.
@@ -70,13 +70,13 @@ public class AMQQueue extends AMQDestina
public AMQQueue(AMQShortString exchangeName, AMQShortString routingKey, AMQShortString queueName)
{
- super(exchangeName, ExchangeDefaults.DIRECT_EXCHANGE_CLASS, routingKey, false,
+ super(exchangeName, AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS), routingKey, false,
false, queueName, false);
}
public AMQQueue(AMQShortString exchangeName, AMQShortString routingKey, AMQShortString queueName,AMQShortString[] bindingKeys)
{
- super(exchangeName, ExchangeDefaults.DIRECT_EXCHANGE_CLASS, routingKey, false,
+ super(exchangeName, AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS), routingKey, false,
false, queueName, false,bindingKeys);
}
@@ -149,7 +149,7 @@ public class AMQQueue extends AMQDestina
public AMQQueue(AMQShortString exchangeName, AMQShortString routingKey, AMQShortString queueName, boolean exclusive, boolean autoDelete, boolean durable,AMQShortString[] bindingKeys)
{
- super(exchangeName, ExchangeDefaults.DIRECT_EXCHANGE_CLASS, routingKey, exclusive,
+ super(exchangeName, AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS), routingKey, exclusive,
autoDelete, queueName, durable, bindingKeys);
}
@@ -167,9 +167,8 @@ public class AMQQueue extends AMQDestina
public boolean isNameRequired()
{
- //If the name is null, we require one to be generated by the client so that it will#
- //remain valid if we failover (see BLZ-24)
- return getQueueName() == null;
+ AMQShortString queueName = getAMQQueueName();
+ return queueName == null || AMQShortString.EMPTY_STRING.equals(queueName);
}
@Override
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java Fri Sep 20 18:59:30 2013
@@ -1380,7 +1380,7 @@ public class AMQSession_0_10 extends AMQ
dest.setExchangeName(new AMQShortString(dest.getAddressName()));
Node node = dest.getNode();
dest.setExchangeClass(node.getExchangeType() == null?
- ExchangeDefaults.TOPIC_EXCHANGE_CLASS:
+ AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS):
new AMQShortString(node.getExchangeType()));
dest.setRoutingKey(new AMQShortString(dest.getSubject()));
}
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java Fri Sep 20 18:59:30 2013
@@ -60,12 +60,12 @@ public class AMQTopic extends AMQDestina
public AMQTopic(AMQShortString exchange, AMQShortString routingKey, AMQShortString queueName)
{
- super(exchange, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, routingKey, true, true, queueName, false);
+ super(exchange, AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS), routingKey, true, true, queueName, false);
}
public AMQTopic(AMQShortString exchange, AMQShortString routingKey, AMQShortString queueName,AMQShortString[] bindingKeys)
{
- super(exchange, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, routingKey, true, true, queueName, false,bindingKeys);
+ super(exchange, AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS), routingKey, true, true, queueName, false,bindingKeys);
}
public AMQTopic(AMQConnection conn, String routingKey)
@@ -73,6 +73,10 @@ public class AMQTopic extends AMQDestina
this(conn.getDefaultTopicExchangeName(), new AMQShortString(routingKey));
}
+ public AMQTopic(String exchangeName, String routingKey)
+ {
+ this(AMQShortString.valueOf(exchangeName), new AMQShortString(routingKey));
+ }
public AMQTopic(AMQShortString exchangeName, String routingKey)
{
@@ -86,7 +90,7 @@ public class AMQTopic extends AMQDestina
public AMQTopic(AMQShortString exchangeName, AMQShortString name, boolean isAutoDelete, AMQShortString queueName, boolean isDurable)
{
- super(exchangeName, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, name, true, isAutoDelete, queueName, isDurable);
+ super(exchangeName, AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS), name, true, isAutoDelete, queueName, isDurable);
}
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java Fri Sep 20 18:59:30 2013
@@ -440,7 +440,7 @@ public abstract class BasicMessageProduc
{
if (!(destination instanceof AMQDestination))
{
- throw new JMSException("Unsupported destination class: "
+ throw new InvalidDestinationException("Unsupported destination class: "
+ ((destination != null) ? destination.getClass() : null));
}
@@ -453,7 +453,7 @@ public abstract class BasicMessageProduc
}
catch(Exception e)
{
- JMSException ex = new JMSException("Error validating destination");
+ JMSException ex = new InvalidDestinationException("Error validating destination");
ex.initCause(e);
ex.setLinkedException(e);
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java Fri Sep 20 18:59:30 2013
@@ -125,7 +125,6 @@ public class AMQMessageDelegate_0_10 ext
String subject = null;
if (isStrictJMS && messageProps != null && messageProps.getApplicationHeaders() != null)
{
- System.out.println("%%%% Going to set subject");
subject = (String)messageProps.getApplicationHeaders().get(QpidMessageProperties.QPID_SUBJECT);
if (subject != null)
{
@@ -207,7 +206,7 @@ public class AMQMessageDelegate_0_10 ext
}
else
{
- throw new JMSException("MessageId '"+messageId+"' is not of the correct format, it must be ID: followed by a UUID");
+ throw new JMSException("MessageId '"+messageId+"' is not of the correct format, it must be prefixed with 'ID:'");
}
}
}
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java Fri Sep 20 18:59:30 2013
@@ -46,7 +46,7 @@ import java.util.concurrent.ConcurrentHa
public abstract class AbstractAMQMessageDelegate implements AMQMessageDelegate
{
- private static Map<String, Integer> _exchangeTypeToDestinationType = new ConcurrentHashMap<String, Integer>();
+ private static Map<String, Integer> _exchangeTypeToDestinationType = new ConcurrentHashMap<String, Integer>();
private static Map<String,ExchangeInfo> _exchangeMap = new ConcurrentHashMap<String, ExchangeInfo>();
/**
@@ -55,32 +55,32 @@ public abstract class AbstractAMQMessage
static
{
_exchangeTypeToDestinationType.put("", AMQDestination.QUEUE_TYPE);
- _exchangeTypeToDestinationType.put(ExchangeDefaults.DIRECT_EXCHANGE_CLASS.toString(), AMQDestination.QUEUE_TYPE);
- _exchangeTypeToDestinationType.put(ExchangeDefaults.TOPIC_EXCHANGE_CLASS.toString(), AMQDestination.TOPIC_TYPE);
- _exchangeTypeToDestinationType.put(ExchangeDefaults.FANOUT_EXCHANGE_CLASS.toString(), AMQDestination.TOPIC_TYPE);
- _exchangeTypeToDestinationType.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS.toString(), AMQDestination.QUEUE_TYPE);
-
+ _exchangeTypeToDestinationType.put(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, AMQDestination.QUEUE_TYPE);
+ _exchangeTypeToDestinationType.put(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, AMQDestination.TOPIC_TYPE);
+ _exchangeTypeToDestinationType.put(ExchangeDefaults.FANOUT_EXCHANGE_CLASS, AMQDestination.TOPIC_TYPE);
+ _exchangeTypeToDestinationType.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS, AMQDestination.QUEUE_TYPE);
+
_exchangeMap.put("", new ExchangeInfo("","",AMQDestination.QUEUE_TYPE));
-
- _exchangeMap.put(ExchangeDefaults.DIRECT_EXCHANGE_NAME.toString(),
- new ExchangeInfo(ExchangeDefaults.DIRECT_EXCHANGE_NAME.toString(),
- ExchangeDefaults.DIRECT_EXCHANGE_CLASS.toString(),
+
+ _exchangeMap.put(ExchangeDefaults.DIRECT_EXCHANGE_NAME,
+ new ExchangeInfo(ExchangeDefaults.DIRECT_EXCHANGE_NAME,
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
AMQDestination.QUEUE_TYPE));
-
- _exchangeMap.put(ExchangeDefaults.TOPIC_EXCHANGE_NAME.toString(),
- new ExchangeInfo(ExchangeDefaults.TOPIC_EXCHANGE_NAME.toString(),
- ExchangeDefaults.TOPIC_EXCHANGE_CLASS.toString(),
+
+ _exchangeMap.put(ExchangeDefaults.TOPIC_EXCHANGE_NAME,
+ new ExchangeInfo(ExchangeDefaults.TOPIC_EXCHANGE_NAME,
+ ExchangeDefaults.TOPIC_EXCHANGE_CLASS,
AMQDestination.TOPIC_TYPE));
-
- _exchangeMap.put(ExchangeDefaults.FANOUT_EXCHANGE_NAME.toString(),
- new ExchangeInfo(ExchangeDefaults.FANOUT_EXCHANGE_NAME.toString(),
- ExchangeDefaults.FANOUT_EXCHANGE_CLASS.toString(),
+
+ _exchangeMap.put(ExchangeDefaults.FANOUT_EXCHANGE_NAME,
+ new ExchangeInfo(ExchangeDefaults.FANOUT_EXCHANGE_NAME,
+ ExchangeDefaults.FANOUT_EXCHANGE_CLASS,
AMQDestination.TOPIC_TYPE));
-
- _exchangeMap.put(ExchangeDefaults.HEADERS_EXCHANGE_NAME.toString(),
- new ExchangeInfo(ExchangeDefaults.HEADERS_EXCHANGE_NAME.toString(),
- ExchangeDefaults.HEADERS_EXCHANGE_CLASS.toString(),
- AMQDestination.QUEUE_TYPE));
+
+ _exchangeMap.put(ExchangeDefaults.HEADERS_EXCHANGE_NAME,
+ new ExchangeInfo(ExchangeDefaults.HEADERS_EXCHANGE_NAME,
+ ExchangeDefaults.HEADERS_EXCHANGE_CLASS,
+ AMQDestination.QUEUE_TYPE));
}
/** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
@@ -115,19 +115,19 @@ public abstract class AbstractAMQMessage
{
AMQDestination dest;
ExchangeInfo exchangeInfo = _exchangeMap.get(exchange.asString());
-
+
if (exchangeInfo == null)
{
exchangeInfo = new ExchangeInfo(exchange.asString(),"",AMQDestination.UNKNOWN_TYPE);
}
-
+
if ("topic".equals(exchangeInfo.getExchangeType()))
{
dest = new AMQTopic(exchange, routingKey, null);
}
else if ("direct".equals(exchangeInfo.getExchangeType()))
{
- dest = new AMQQueue(exchange, routingKey, routingKey);
+ dest = new AMQQueue(exchange, routingKey, routingKey);
}
else
{
@@ -161,7 +161,7 @@ public abstract class AbstractAMQMessage
{
type = AMQDestination.UNKNOWN_TYPE;
}
-
+
_exchangeMap.put(exchange, new ExchangeInfo(exchange,newtype,type));
}
@@ -226,7 +226,7 @@ class ExchangeInfo
private String exchangeName;
private String exchangeType;
private int destType = AMQDestination.QUEUE_TYPE;
-
+
public ExchangeInfo(String exchangeName, String exchangeType,
int destType)
{
Modified: qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java Fri Sep 20 18:59:30 2013
@@ -274,11 +274,11 @@ public class PropertiesFileInitialContex
{
if (value instanceof AMQShortString)
{
- return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, (AMQShortString) value);
+ return new AMQQueue(AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_NAME), (AMQShortString) value);
}
else if (value instanceof String)
{
- return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString((String) value));
+ return new AMQQueue(AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_NAME), new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
{
@@ -295,7 +295,7 @@ public class PropertiesFileInitialContex
{
if (value instanceof AMQShortString)
{
- return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, (AMQShortString) value);
+ return new AMQTopic(AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_NAME), (AMQShortString) value);
}
else if (value instanceof String)
{
@@ -309,7 +309,7 @@ public class PropertiesFileInitialContex
}
// The Destination has a dual nature. If this was used for a producer the key is used
// for the routing key. If it was used for the consumer it becomes the bindingKey
- return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME,bindings[0],null,bindings);
+ return new AMQTopic(AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_NAME),bindings[0],null,bindings);
}
else if (value instanceof BindingURL)
{
Modified: qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java Fri Sep 20 18:59:30 2013
@@ -124,7 +124,6 @@ public class AMQMessageDelegate_0_10Test
for (Enumeration props = delegate.getPropertyNames(); props.hasMoreElements();)
{
String key = (String)props.nextElement();
- System.out.println("PropName : " + key);
if (key.equals("JMS_" + QpidMessageProperties.QPID_SUBJECT))
{
propFound = true;
Modified: qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactoryTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactoryTest.java?rev=1525101&r1=1525100&r2=1525101&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactoryTest.java (original)
+++ qpid/branches/linearstore/qpid/java/client/src/test/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactoryTest.java Fri Sep 20 18:59:30 2013
@@ -115,7 +115,7 @@ public class PropertiesFileInitialContex
setTestSystemProperty(ClientProperties.DEST_SYNTAX, "ADDR");
setTestSystemProperty(InitialContext.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
- setTestSystemProperty(InitialContext.PROVIDER_URL, "file://" + f.getCanonicalPath());
+ setTestSystemProperty(InitialContext.PROVIDER_URL, f.toURI().toURL().toString());
InitialContext context = new InitialContext();
Destination dest = (Destination) context.lookup("topicExchange");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org