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 2016/11/27 10:55:36 UTC
svn commit: r1771561 - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/model/
broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/
broker-plugins/management-amqp/src/main/java/org/apache/qpid/server...
Author: rgodfrey
Date: Sun Nov 27 10:55:36 2016
New Revision: 1771561
URL: http://svn.apache.org/viewvc?rev=1771561&view=rev
Log:
QPID-7546 : Make some more tests run under AMQP 1.0, fix getExpiration for 1.0 messages, add the ability to have object names including / found by their path using escaping
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ConfiguredObjectFinder.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java
qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
qpid/java/trunk/test-profiles/Java10UninvestigatedTestsExcludes
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ConfiguredObjectFinder.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ConfiguredObjectFinder.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ConfiguredObjectFinder.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ConfiguredObjectFinder.java Sun Nov 27 10:55:36 2016
@@ -32,7 +32,9 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.ListIterator;
import java.util.Map;
+import java.util.regex.Pattern;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.util.Strings;
@@ -79,10 +81,19 @@ public class ConfiguredObjectFinder
}
}
}
+ private String[] getPathElements(final String path)
+ {
+ String[] pathElements = path.split("(?<!\\\\)" + Pattern.quote("/"));
+ for(int i = 0; i<pathElements.length; i++)
+ {
+ pathElements[i] = pathElements[i].replaceAll("\\\\(.)","$1");
+ }
+ return pathElements;
+ }
public ConfiguredObject<?> findObjectFromPath(String path, Class<? extends ConfiguredObject> category)
{
- return findObjectFromPath(Arrays.asList(path.split("/")), category);
+ return findObjectFromPath(Arrays.asList(getPathElements(path)), category);
}
public ConfiguredObject<?> findObjectFromPath(List<String> path, Class<? extends ConfiguredObject> category)
@@ -348,7 +359,15 @@ public class ConfiguredObjectFinder
public String getPath(final ConfiguredObject<?> object)
{
- return Strings.join("/", getPathAsList(object));
+ final List<String> pathAsList = getPathAsList(object);
+ ListIterator<String> iter = pathAsList.listIterator();
+ while(iter.hasNext())
+ {
+ String element = iter.next();
+ iter.set(element.replaceAll("([\\\\/])", "\\\\$1"));
+
+ }
+ return Strings.join("/", pathAsList);
}
public List<String> getPathAsList(final ConfiguredObject<?> object)
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageMetaData_1_0.java Sun Nov 27 10:55:36 2016
@@ -452,7 +452,14 @@ public class MessageMetaData_1_0 impleme
public long getExpiration()
{
- return 0; //TODO
+ if(_properties.getAbsoluteExpiryTime() != null && _properties.getCreationTime() != null)
+ {
+ return _properties.getAbsoluteExpiryTime().getTime() - _properties.getCreationTime().getTime();
+ }
+ else
+ {
+ return _header.getTtl() == null ? 0L : _header.getTtl().longValue();
+ }
}
public String getMessageId()
Modified: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java (original)
+++ qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java Sun Nov 27 10:55:36 2016
@@ -44,6 +44,7 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.regex.Pattern;
import javax.security.auth.Subject;
@@ -583,7 +584,7 @@ class ManagementNode implements MessageS
}
if (path != null && !attributes.containsKey(ConfiguredObject.NAME))
{
- String[] pathElements = path.split("/");
+ String[] pathElements = getPathElements(path);
attributes.put(ConfiguredObject.NAME, pathElements[pathElements.length - 1]);
}
object.setAttributes(attributes);
@@ -609,6 +610,17 @@ class ManagementNode implements MessageS
}
}
+ private String[] getPathElements(final String path)
+ {
+ String[] pathElements = path.split("(?<!\\\\)" + Pattern.quote("/"));
+ for(int i = 0; i<pathElements.length; i++)
+ {
+ pathElements[i] = pathElements[i].replaceAll("\\\\(.)","$1");
+ }
+ return pathElements;
+ }
+
+
private InternalMessage performReadOperation(final Class<? extends ConfiguredObject> clazz,
final InternalMessage message)
{
@@ -697,7 +709,7 @@ class ManagementNode implements MessageS
{
List<ConfiguredObject> parents =
- _configuredObjectFinder.findObjectParentsFromPath(Arrays.asList(path.split("/")), hierarchy, _model.getTypeRegistry().getCategory(clazz));
+ _configuredObjectFinder.findObjectParentsFromPath(Arrays.asList(getPathElements(path)), hierarchy, _model.getTypeRegistry().getCategory(clazz));
if(parents.isEmpty())
{
return createFailureResponse(message, STATUS_CODE_NOT_FOUND, "The '"+OBJECT_PATH+"' "+path+" does not identify a valid parent");
Modified: qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java (original)
+++ qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java Sun Nov 27 10:55:36 2016
@@ -581,7 +581,8 @@ public class QpidBrokerTestCase extends
MessageConsumer consumer = session.createConsumer(responseQ);
MapMessage message = session.createMapMessage();
message.setStringProperty("index", "object-path");
- message.setStringProperty("key", destination.getQueueName());
+ final String escapedName = destination.getQueueName().replaceAll("([/\\\\])", "\\\\$1");
+ message.setStringProperty("key", escapedName);
message.setStringProperty("type", "org.apache.qpid.Queue");
message.setStringProperty("operation", "getStatistics");
message.setStringProperty("statistics", "[\"queueDepthMessages\"]");
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java Sun Nov 27 10:55:36 2016
@@ -40,9 +40,7 @@ import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
public class TimeToLiveTest extends QpidBrokerTestCase
@@ -83,14 +81,8 @@ public class TimeToLiveTest extends Qpid
Connection clientConnection = getConnection();
Session clientSession = clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = clientSession.createQueue(QUEUE);
+ Queue queue = createTestQueue(clientSession, QUEUE);
- // Create then close the consumer so the queue is actually created
- // Closing it then reopening it ensures that the consumer shouldn't get messages
- // which should have expired and allows a shorter sleep period. See QPID-1418
-
- MessageConsumer consumer = clientSession.createConsumer(queue);
- consumer.close();
//Create Producer
Connection producerConnection = getConnection();
@@ -103,7 +95,7 @@ public class TimeToLiveTest extends Qpid
MessageProducer producer = producerSession.createProducer(queue);
- consumer = clientSession.createConsumer(queue);
+ MessageConsumer consumer = clientSession.createConsumer(queue);
if(prefetchMessages)
{
clientConnection.start();
@@ -207,9 +199,9 @@ public class TimeToLiveTest extends Qpid
public void testActiveTTL() throws Exception
{
Connection producerConnection = getConnection();
- AMQSession producerSession = (AMQSession) producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = producerSession.createTemporaryQueue();
- producerSession.declareAndBind((AMQDestination) queue);
+ Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = createTestQueue(producerSession);
+
MessageProducer producer = producerSession.createProducer(queue);
producer.setTimeToLive(1000L);
@@ -228,7 +220,8 @@ public class TimeToLiveTest extends Qpid
{
lastPass = messageCount;
Thread.sleep(100);
- messageCount = producerSession.getQueueDepth((AMQDestination) queue);
+ producerConnection.start();
+ messageCount = getQueueDepth(producerConnection, queue);
// If we have received messages in the last loop then extend the timeout time.
// if we get messages stuck that are not expiring then the failureTime will occur
@@ -339,6 +332,7 @@ public class TimeToLiveTest extends Qpid
Assert.assertFalse("Final message has first set.", receivedSecond.getBooleanProperty("first"));
Assert.assertEquals("Final message has incorrect TTL.", 0L, receivedSecond.getLongProperty("TTL"));
+ durableSubscriber.close();
clientSession.unsubscribe(getTestQueueName());
clientConnection.close();
@@ -358,7 +352,8 @@ public class TimeToLiveTest extends Qpid
//Create Producer
Connection producerConnection = getConnection();
- AMQSession producerSession = (AMQSession) producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ producerConnection.start();
+ Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = producerSession.createProducer(topic);
producer.setTimeToLive(1000L);
@@ -372,12 +367,12 @@ public class TimeToLiveTest extends Qpid
// check Queue depth for up to TIMEOUT seconds after the Queue Depth hasn't changed for 100ms.
long messageCount = MSG_COUNT;
long lastPass;
- AMQQueue subcriptionQueue = new AMQQueue("amq.topic","clientid" + ":" + "MyDurableTTLSubscription");
+ Queue subcriptionQueue = isBroker10() ? producerSession.createQueue("qpid_/clientid_/MyDurableTTLSubscription") : new AMQQueue("amq.topic","clientid" + ":" + "MyDurableTTLSubscription");
do
{
lastPass = messageCount;
Thread.sleep(100);
- messageCount = producerSession.getQueueDepth((AMQDestination) subcriptionQueue);
+ messageCount = getQueueDepth(producerConnection, subcriptionQueue);
// If we have received messages in the last loop then extend the timeout time.
// if we get messages stuck that are not expiring then the failureTime will occur
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java Sun Nov 27 10:55:36 2016
@@ -26,20 +26,18 @@ import java.util.concurrent.CountDownLat
import java.util.concurrent.TimeUnit;
import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
+import javax.jms.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.message.JMSBytesMessage;
import org.apache.qpid.framing.AMQFrameDecodingException;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.FieldTableFactory;
@@ -49,10 +47,10 @@ public class FieldTableMessageTest exten
{
private static final Logger _logger = LoggerFactory.getLogger(FieldTableMessageTest.class);
- private AMQConnection _connection;
- private AMQDestination _destination;
- private AMQSession _session;
- private final ArrayList<JMSBytesMessage> received = new ArrayList<JMSBytesMessage>();
+ private Connection _connection;
+ private Destination _destination;
+ private Session _session;
+ private final ArrayList<BytesMessage> received = new ArrayList<>();
private FieldTable _expected;
private int _count = 10;
private CountDownLatch _waitForCompletion;
@@ -60,7 +58,15 @@ public class FieldTableMessageTest exten
protected void setUp() throws Exception
{
super.setUp();
- init( (AMQConnection) getConnection("guest", "guest"));
+ _connection = getConnection();
+ _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ _destination = createTestQueue(_session);
+ // set up a slow consumer
+ _session.createConsumer(_destination).setMessageListener(this);
+ _connection.start();
+
+ // _expected = new FieldTableTest().load("FieldTableTest2.properties");
+ _expected = load();
}
protected void tearDown() throws Exception
@@ -68,25 +74,6 @@ public class FieldTableMessageTest exten
super.tearDown();
}
- private void init(AMQConnection connection) throws Exception
- {
- init(connection, new AMQQueue(connection, randomize("FieldTableMessageTest"), true));
- }
-
- private void init(AMQConnection connection, AMQDestination destination) throws Exception
- {
- _connection = connection;
- _destination = destination;
- _session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
-
- // set up a slow consumer
- _session.createConsumer(destination).setMessageListener(this);
- connection.start();
-
- // _expected = new FieldTableTest().load("FieldTableTest2.properties");
- _expected = load();
- }
-
private FieldTable load() throws IOException
{
FieldTable result = FieldTableFactory.newFieldTable();
@@ -125,9 +112,8 @@ public class FieldTableMessageTest exten
void check() throws JMSException, AMQFrameDecodingException, IOException
{
- for (Object m : received)
+ for (BytesMessage bytesMessage : received)
{
- final BytesMessage bytesMessage = (BytesMessage) m;
final long bodyLength = bytesMessage.getBodyLength();
byte[] data = new byte[(int) bodyLength];
bytesMessage.readBytes(data);
@@ -143,14 +129,10 @@ public class FieldTableMessageTest exten
{
synchronized (received)
{
- received.add((JMSBytesMessage) message);
+ received.add((BytesMessage) message);
_waitForCompletion.countDown();
}
}
- private static String randomize(String in)
- {
- return in + System.currentTimeMillis();
- }
}
Modified: qpid/java/trunk/test-profiles/Java10UninvestigatedTestsExcludes
URL: http://svn.apache.org/viewvc/qpid/java/trunk/test-profiles/Java10UninvestigatedTestsExcludes?rev=1771561&r1=1771560&r2=1771561&view=diff
==============================================================================
--- qpid/java/trunk/test-profiles/Java10UninvestigatedTestsExcludes (original)
+++ qpid/java/trunk/test-profiles/Java10UninvestigatedTestsExcludes Sun Nov 27 10:55:36 2016
@@ -43,7 +43,6 @@ org.apache.qpid.server.SupportedProtocol
org.apache.qpid.server.stats.StatisticsReportingTest#*
org.apache.qpid.server.security.acl.ExternalACLTest#*
org.apache.qpid.server.security.acl.ExhaustiveACLTest#*
-org.apache.qpid.server.queue.TimeToLiveTest#*
org.apache.qpid.server.queue.QueueBindTest#*
org.apache.qpid.server.queue.ProducerFlowControlTest#*
org.apache.qpid.server.queue.PriorityQueueTest#*
@@ -69,7 +68,6 @@ org.apache.qpid.test.unit.client.channel
org.apache.qpid.test.unit.client.AMQSessionTest#*
org.apache.qpid.test.unit.basic.SessionStartTest#*
org.apache.qpid.test.unit.basic.InvalidDestinationTest#*
-org.apache.qpid.test.unit.basic.FieldTableMessageTest#*
org.apache.qpid.test.unit.basic.close.CloseTest#*
org.apache.qpid.test.client.queue.LVQTest#*
org.apache.qpid.test.client.message.MessageToStringTest#*
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org