diff options
Diffstat (limited to 'qpid/java')
4 files changed, 128 insertions, 55 deletions
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java index 0973815231..cd69289492 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java @@ -46,12 +46,12 @@ import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageFormatException; import javax.jms.MessageNotWriteableException; -import java.lang.ref.SoftReference; import java.util.ArrayList; import java.util.Collections; import java.util.Enumeration; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -65,7 +65,22 @@ import java.util.UUID; public class AMQMessageDelegate_0_10 extends AbstractAMQMessageDelegate { private static final Logger _logger = LoggerFactory.getLogger(AMQMessageDelegate_0_10.class); - private static final Map<ReplyTo, SoftReference<Destination>> _destinationCache = Collections.synchronizedMap(new HashMap<ReplyTo, SoftReference<Destination>>()); + + private static final float DESTINATION_CACHE_LOAD_FACTOR = 0.75f; + private static final int DESTINATION_CACHE_SIZE = 500; + private static final int DESTINATION_CACHE_CAPACITY = (int) (DESTINATION_CACHE_SIZE / DESTINATION_CACHE_LOAD_FACTOR); + + private static final Map<ReplyTo, Destination> _destinationCache = + Collections.synchronizedMap(new LinkedHashMap<ReplyTo,Destination>(DESTINATION_CACHE_CAPACITY, + DESTINATION_CACHE_LOAD_FACTOR, + true) + { + @Override + protected boolean removeEldestEntry(Map.Entry<ReplyTo, Destination> eldest) + { + return size() >= DESTINATION_CACHE_SIZE; + } + }); public static final String JMS_TYPE = "x-jms-type"; @@ -241,12 +256,8 @@ public class AMQMessageDelegate_0_10 extends AbstractAMQMessageDelegate } else { - Destination dest = null; - SoftReference<Destination> ref = _destinationCache.get(replyTo); - if (ref != null) - { - dest = ref.get(); - } + Destination dest = _destinationCache.get(replyTo); + if (dest == null) { String exchange = replyTo.getExchange(); @@ -254,14 +265,13 @@ public class AMQMessageDelegate_0_10 extends AbstractAMQMessageDelegate if (AMQDestination.getDefaultDestSyntax() == AMQDestination.DestSyntax.BURL) { - dest = generateDestination(new AMQShortString(exchange), new AMQShortString(routingKey)); } else { dest = convertToAddressBasedDestination(exchange,routingKey,null); } - _destinationCache.put(replyTo, new SoftReference<Destination>(dest)); + _destinationCache.put(replyTo, dest); } return dest; @@ -355,13 +365,11 @@ public class AMQMessageDelegate_0_10 extends AbstractAMQMessageDelegate jmse.setLinkedException(e); throw jmse; } - } - + final ReplyTo replyTo = new ReplyTo(amqd.getExchangeName().toString(), amqd.getRoutingKey().toString()); - _destinationCache.put(replyTo, new SoftReference<Destination>(destination)); + _destinationCache.put(replyTo, destination); _messageProps.setReplyTo(replyTo); - } public Destination getJMSDestination() throws JMSException diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java b/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java new file mode 100644 index 0000000000..4636851c79 --- /dev/null +++ b/qpid/java/client/src/test/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10Test.java @@ -0,0 +1,63 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.qpid.client.message; + +import javax.jms.Destination; + +import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.transport.DeliveryProperties; +import org.apache.qpid.transport.MessageProperties; +import org.apache.qpid.transport.ReplyTo; + +public class AMQMessageDelegate_0_10Test extends QpidTestCase +{ + /** + * Tests that when two messages arrive with the same ReplyTo exchange and routingKey values, + * the cache returns the same Destination object from getJMSReplyTo instead of a new one. + */ + public void testDestinationCache() throws Exception + { + //create a message delegate and retrieve the replyTo Destination + AMQMessageDelegate_0_10 delegate1 = generateMessageDelegateWithReplyTo(); + Destination dest1 = delegate1.getJMSReplyTo(); + + //create a new message delegate with the same details, and retrieve the replyTo Destination + AMQMessageDelegate_0_10 delegate2 = generateMessageDelegateWithReplyTo(); + Destination dest2 = delegate2.getJMSReplyTo(); + + //verify that the destination cache means these are the same Destination object + assertSame("Should have received the same Destination objects", dest1, dest2); + } + + private AMQMessageDelegate_0_10 generateMessageDelegateWithReplyTo() + { + MessageProperties mesProps = new MessageProperties(); + ReplyTo reply = new ReplyTo("amq.direct", "myReplyQueue"); + mesProps.setReplyTo(reply); + + DeliveryProperties delProps = new DeliveryProperties(); + delProps.setExchange("amq.direct"); + delProps.setRoutingKey("myRequestQueue"); + + AMQMessageDelegate_0_10 delegate = new AMQMessageDelegate_0_10(mesProps,delProps,1L); + return delegate; + } +} diff --git a/qpid/java/common/Composite.tpl b/qpid/java/common/Composite.tpl index 18b5374965..2b2f8fda09 100644 --- a/qpid/java/common/Composite.tpl +++ b/qpid/java/common/Composite.tpl @@ -362,4 +362,22 @@ for f in fields: return result; } +${ +if name == "ReplyTo": + out(" public boolean equals(final Object obj){\n") + out(" if (this == obj){\n") + out(" return true;\n") + out(" }\n\n") + out(" if(!(obj instanceof ReplyTo)){\n") + out(" return false;\n") + out(" }\n\n") + out(" final ReplyTo reply = (ReplyTo) obj;\n") + out(" return (routingKey == null ? reply.getRoutingKey() == null : routingKey.equals(reply.getRoutingKey()))\n") + out(" && (exchange == null ? reply.getExchange() == null : exchange.equals(reply.getExchange()));\n") + out(" }\n\n") + out(" public int hashCode(){\n") + out(" int result = routingKey == null ? 1 : routingKey.hashCode();\n") + out(" return 31 * result + (exchange == null ? 5 : exchange.hashCode());\n") + out(" }") +} } diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java index 4936934e38..52213d15c4 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java @@ -20,17 +20,14 @@ */ package org.apache.qpid.test.unit.basic; -import junit.framework.Assert; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -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.JMSTextMessage; -import org.apache.qpid.test.utils.QpidBrokerTestCase; -import org.apache.qpid.url.AMQBindingURL; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; import javax.jms.Connection; import javax.jms.Destination; @@ -41,29 +38,34 @@ import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.net.URISyntaxException; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; + +import junit.framework.Assert; + +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.message.JMSTextMessage; +import org.apache.qpid.test.utils.QpidBrokerTestCase; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class PropertyValueTest extends QpidBrokerTestCase implements MessageListener { private static final Logger _logger = LoggerFactory.getLogger(PropertyValueTest.class); - private int count = 0; private AMQConnection _connection; private Destination _destination; private AMQSession _session; private final List<JMSTextMessage> received = new ArrayList<JMSTextMessage>(); private final List<String> messages = new ArrayList<String>(); + private Map<String, Destination> _replyToDestinations; private int _count = 1; public String _connectionString = "vm://:1"; private static final String USERNAME = "guest"; protected void setUp() throws Exception { + _replyToDestinations = new HashMap<String, Destination>(); super.setUp(); } @@ -234,12 +236,11 @@ public class PropertyValueTest extends QpidBrokerTestCase implements MessageList } m.setJMSReplyTo(q); - m.setStringProperty("TempQueue", q.toString()); - _logger.debug("Message:" + m); + m.setStringProperty("ReplyToIndex", String.valueOf(i)); + _replyToDestinations.put(String.valueOf(i), q); - Assert.assertEquals("Check temp queue has been set correctly", m.getJMSReplyTo().toString(), - m.getStringProperty("TempQueue")); + _logger.debug("Message:" + m); m.setJMSType("Test"); m.setLongProperty("UnsignedInt", (long) 4294967295L); @@ -287,8 +288,8 @@ public class PropertyValueTest extends QpidBrokerTestCase implements MessageList Assert.assertEquals("Check Priority properties are correctly transported", 8, m.getJMSPriority()); // Queue - Assert.assertEquals("Check ReplyTo properties are correctly transported", AMQDestination.createDestination(new AMQBindingURL(m.getStringProperty("TempQueue"))), - m.getJMSReplyTo()); + String replyToIndex = m.getStringProperty("ReplyToIndex"); + Assert.assertEquals("Check ReplyTo properties are correctly transported", _replyToDestinations.get(replyToIndex), m.getJMSReplyTo()); Assert.assertEquals("Check Type properties are correctly transported", "Test", m.getJMSType()); @@ -299,24 +300,7 @@ public class PropertyValueTest extends QpidBrokerTestCase implements MessageList Assert.assertEquals("Check Long properties are correctly transported", (long) Long.MAX_VALUE, m.getLongProperty("Long")); Assert.assertEquals("Check String properties are correctly transported", "Test", m.getStringProperty("String")); -/* - // AMQP Tests Specific values - - Assert.assertEquals("Check Timestamp properties are correctly transported", m.getStringProperty("time-str"), - ((AMQMessage) m).getTimestampProperty(new AMQShortString("time")).toString()); - - // Decimal - BigDecimal bd = new BigDecimal(Integer.MAX_VALUE); - - Assert.assertEquals("Check decimal properties are correctly transported", bd.setScale(Byte.MAX_VALUE), - ((AMQMessage) m).getDecimalProperty(new AMQShortString("decimal"))); - - // Void - ((AMQMessage) m).setVoidProperty(new AMQShortString("void")); - Assert.assertTrue("Check void properties are correctly transported", - ((AMQMessage) m).getPropertyHeaders().containsKey("void")); -*/ //JMSXUserID if (m.getStringProperty("JMSXUserID") != null) { |