Author: orudyy
Date: Wed May 11 16:31:40 2016
New Revision: 1743394
URL: http://svn.apache.org/viewvc?rev=1743394&view=rev
Log:
QPID-5816: [Java Client] Maintain a per-session (weak) cache of resolved
Destinations
merged from trunk using
svn merge -c 1743228 ^/qpid/java/trunk
Modified:
qpid/java/branches/6.0.x/ (props changed)
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQDestination.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
qpid/java/branches/6.0.x/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
Propchange: qpid/java/branches/6.0.x/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Wed May 11 16:31:40 2016
@@ -9,5 +9,5 @@
/qpid/branches/java-broker-vhost-refactor/java:1493674-1494547
/qpid/branches/java-network-refactor/qpid/java:805429-821809
/qpid/branches/qpid-2935/qpid/java:1061302-1072333
-/qpid/java/trunk:1715445-1715447,1715586,1715940,1716086-1716087,1716127-1716128,1716141,1716153,1716155,1716194,1716204,1716209,1716227,1716277,1716357,1716368,1716370,1716374,1716432,1716444-1716445,1716455,1716461,1716474,1716489,1716497,1716515,1716555,1716602,1716606-1716610,1716619,1716636,1717269,1717299,1717401,1717446,1717449,1717626,1717691,1717735,1717780,1718744,1718889,1718893,1718918,1718922,1719026,1719028,1719033,1719037,1719047,1719051,1720340,1720664,1721151,1721198,1722019-1722020,1722246,1722339,1722416,1722674,1722678,1722683,1722711,1723064,1723194,1723563,1724216,1724251,1724257,1724292,1724375,1724397,1724432,1724582,1724603,1724780,1724843-1724844,1725295,1725569,1725760,1726176,1726244-1726246,1726249,1726358,1726436,1726449,1726456,1726646,1726653,1726755,1726778,1727532,1727555,1727608,1727951,1727954,1728089,1728167,1728302,1728497,1728501,1728524,1728639,1728651,1728772,1729215,1729297,1729347,1729356,1729406,1729408,1729412,1729515,1729638,1729656-1729
657,1729783,1729828,1729832,1729841,1729851,1729886,1729904,1729973,1730019,1730025,1730052,1730072,1730088,1730494,1730499,1730547,1730559,1730567,1730578,1730585,1730651,1730697,1730712-1730713,1730805,1731029,1731110,1731210,1731225,1731444,1731551,1731612,1732184,1732452,1732461,1732525,1732812,1733467,1734452,1736478,1736751,1736838,1737804,1737835,1737853,1737984,1737992,1738119,1738135,1738231,1738271,1738607,1738610,1738731,1738914,1741702,1742257,1742284,1742544,1742900,1742926,1743161,1743383
+/qpid/java/trunk:1715445-1715447,1715586,1715940,1716086-1716087,1716127-1716128,1716141,1716153,1716155,1716194,1716204,1716209,1716227,1716277,1716357,1716368,1716370,1716374,1716432,1716444-1716445,1716455,1716461,1716474,1716489,1716497,1716515,1716555,1716602,1716606-1716610,1716619,1716636,1717269,1717299,1717401,1717446,1717449,1717626,1717691,1717735,1717780,1718744,1718889,1718893,1718918,1718922,1719026,1719028,1719033,1719037,1719047,1719051,1720340,1720664,1721151,1721198,1722019-1722020,1722246,1722339,1722416,1722674,1722678,1722683,1722711,1723064,1723194,1723563,1724216,1724251,1724257,1724292,1724375,1724397,1724432,1724582,1724603,1724780,1724843-1724844,1725295,1725569,1725760,1726176,1726244-1726246,1726249,1726358,1726436,1726449,1726456,1726646,1726653,1726755,1726778,1727532,1727555,1727608,1727951,1727954,1728089,1728167,1728302,1728497,1728501,1728524,1728639,1728651,1728772,1729215,1729297,1729347,1729356,1729406,1729408,1729412,1729515,1729638,1729656-1729
657,1729783,1729828,1729832,1729841,1729851,1729886,1729904,1729973,1730019,1730025,1730052,1730072,1730088,1730494,1730499,1730547,1730559,1730567,1730578,1730585,1730651,1730697,1730712-1730713,1730805,1731029,1731110,1731210,1731225,1731444,1731551,1731612,1732184,1732452,1732461,1732525,1732812,1733467,1734452,1736478,1736751,1736838,1737804,1737835,1737853,1737984,1737992,1738119,1738135,1738231,1738271,1738607,1738610,1738731,1738914,1741702,1742257,1742284,1742544,1742900,1742926,1743161,1743228,1743383
/qpid/trunk/qpid:796646-796653
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQDestination.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQDestination.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQDestination.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQDestination.java
Wed May 11 16:31:40 2016
@@ -26,7 +26,6 @@ import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.net.URISyntaxException;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
import javax.jms.Destination;
import javax.naming.NamingException;
@@ -80,8 +79,6 @@ public abstract class AMQDestination imp
private boolean _browseOnly;
- private AtomicLong _addressResolved = new AtomicLong(0);
-
private String _queueName;
private String _routingKey;
@@ -93,8 +90,6 @@ public abstract class AMQDestination imp
private boolean _checkedForQueueBinding;
- private boolean _exchangeExistsChecked;
-
private RejectBehaviour _rejectBehaviour;
private Map<String,Object> _consumerArguments;
@@ -548,17 +543,6 @@ public abstract class AMQDestination imp
_checkedForQueueBinding = checkedForQueueBinding;
}
-
- public boolean isExchangeExistsChecked()
- {
- return _exchangeExistsChecked;
- }
-
- public void setExchangeExistsChecked(final boolean exchangeExistsChecked)
- {
- _exchangeExistsChecked = exchangeExistsChecked;
- }
-
public String toURL()
{
String url = _url;
@@ -947,16 +931,6 @@ public abstract class AMQDestination imp
this._routingKey = rk;
}
- public boolean isAddressResolved()
- {
- return _addressResolved.get() > 0;
- }
-
- public void setAddressResolved(long addressResolved)
- {
- _addressResolved.set(addressResolved);
- }
-
private static Address createAddressFromString(String str)
{
return Address.parse(str);
@@ -1024,7 +998,6 @@ public abstract class AMQDestination imp
dest.setAddressType(_addressType);
dest.setNode(_node);
dest.setLink(_link);
- dest.setAddressResolved(_addressResolved.get());
return dest;
}
@@ -1038,11 +1011,6 @@ public abstract class AMQDestination imp
_isDurable = b;
}
- public boolean isResolvedAfter(long time)
- {
- return _addressResolved.get() > time;
- }
-
/**
* This option is only applicable for 0-8/0-9/0-9-1 protocols connection
* <p>
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession.java
Wed May 11 16:31:40 2016
@@ -141,6 +141,9 @@ public abstract class AMQSession<C exten
*/
protected final boolean DAEMON_DISPATCHER_THREAD =
Boolean.getBoolean(ClientProperties.DAEMON_DISPATCHER);
+ private final Set<AMQDestination>
+ _resolvedDestinations =
Collections.synchronizedSet(Collections.newSetFromMap(new
WeakHashMap<AMQDestination, Boolean>()));
+
private final long _dispatcherShutdownTimeoutMs;
/** The connection to which this session belongs. */
@@ -591,7 +594,7 @@ public abstract class AMQSession<C exten
boolean isConsumer,
boolean noLocal) throws
QpidException
{
- if (dest.isAddressResolved() &&
dest.isResolvedAfter(getAMQConnection().getLastFailoverTime()))
+ if (isResolved(dest))
{
return;
}
@@ -643,10 +646,30 @@ public abstract class AMQSession<C exten
"The name '" + dest.getAddressName() +
"' supplied in the address doesn't resolve to an
exchange or a queue");
}
- dest.setAddressResolved(System.currentTimeMillis());
+ setResolved(dest);
}
}
+ void setResolved(final AMQDestination dest)
+ {
+ _resolvedDestinations.add(dest);
+ }
+
+ void setUnresolved(final AMQDestination dest)
+ {
+ _resolvedDestinations.remove(dest);
+ }
+
+ private void clearResolvedDestinations()
+ {
+ _resolvedDestinations.clear();
+ }
+
+ boolean isResolved(final AMQDestination dest)
+ {
+ return _resolvedDestinations.contains(dest);
+ }
+
public abstract int resolveAddressType(AMQDestination dest) throws
QpidException;
protected abstract void acknowledgeImpl() throws JMSException;
@@ -1060,8 +1083,7 @@ public abstract class AMQSession<C exten
Topic origTopic = checkValidTopic(topic, true);
AMQTopic dest = AMQTopic.createDurableTopic(origTopic, name,
_connection);
- if (dest.getDestSyntax() == DestSyntax.ADDR &&
- !dest.isAddressResolved())
+ if (dest.getDestSyntax() == DestSyntax.ADDR && !isResolved(dest))
{
try
{
@@ -2363,6 +2385,7 @@ public abstract class AMQSession<C exten
_prefetchedMessageTags.clear();
_rollbackMark.set(-1);
+ clearResolvedDestinations();
resubscribeProducers();
resubscribeConsumers();
}
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
Wed May 11 16:31:40 2016
@@ -1449,7 +1449,7 @@ public class AMQSession_0_10 extends AMQ
if (isExchangeExist(dest,false))
{
getQpidSession().exchangeDelete(dest.getAddressName());
- dest.setAddressResolved(0);
+ setUnresolved(dest);
}
}
else
@@ -1457,7 +1457,7 @@ public class AMQSession_0_10 extends AMQ
if (isQueueExist(dest,false))
{
getQpidSession().queueDelete(dest.getAddressName());
- dest.setAddressResolved(0);
+ setUnresolved(dest);
}
}
}
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
Wed May 11 16:31:40 2016
@@ -1291,7 +1291,7 @@ public class AMQSession_0_8 extends AMQS
return null;
}
}, getAMQConnection()).execute();
- dest.setAddressResolved(0);
+ setUnresolved(dest);
}
}
else
@@ -1306,7 +1306,7 @@ public class AMQSession_0_8 extends AMQS
return null;
}
}, getAMQConnection()).execute();
- dest.setAddressResolved(0);
+ setUnresolved(dest);
}
}
}
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
Wed May 11 16:31:40 2016
@@ -98,7 +98,7 @@ public class BasicMessageConsumer_0_10 e
_capacity = evaluateCapacity(destination);
// This is due to the Destination carrying the temporary subscription
name which is incorrect.
- if (destination.isAddressResolved() && AMQDestination.TOPIC_TYPE ==
destination.getAddressType())
+ if (_0_10session.isResolved(destination) && AMQDestination.TOPIC_TYPE
== destination.getAddressType())
{
boolean namedQueue = destination.getLink() != null &&
destination.getLink().getName() != null ;
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
Wed May 11 16:31:40 2016
@@ -77,7 +77,7 @@ public class BasicMessageConsumer_0_8 ex
// This is due to the Destination carrying the temporary subscription
name which is incorrect.
- if (destination.isAddressResolved() && AMQDestination.TOPIC_TYPE ==
destination.getAddressType())
+ if (session.isResolved(destination) && AMQDestination.TOPIC_TYPE ==
destination.getAddressType())
{
boolean namedQueue = destination.getLink() != null &&
destination.getLink().getName() != null ;
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
Wed May 11 16:31:40 2016
@@ -473,7 +473,7 @@ public abstract class BasicMessageProduc
}
AMQDestination amqDestination = (AMQDestination) destination;
- if(!amqDestination.isExchangeExistsChecked())
+ if (!_session.isResolved(amqDestination))
{
try
{
@@ -484,7 +484,6 @@ public abstract class BasicMessageProduc
throw JMSExceptionHelper.chainJMSException(new
InvalidDestinationException(
"Error validating destination"), e);
}
- amqDestination.setExchangeExistsChecked(true);
}
}
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
Wed May 11 16:31:40 2016
@@ -84,17 +84,19 @@ public class BasicMessageProducer_0_10 e
{
if (destination.getDestSyntax() == DestSyntax.BURL)
{
- if (getSession().isDeclareExchanges())
- {
- String name = destination.getExchangeName();
- ((AMQSession_0_10)
getSession()).getQpidSession().exchangeDeclare
- (name,
- destination.getExchangeClass(),
- null, null,
- name.startsWith("amq.") ? Option.PASSIVE : Option.NONE,
- destination.isExchangeDurable() ? Option.DURABLE :
Option.NONE,
- destination.isExchangeAutoDelete() ?
Option.AUTO_DELETE : Option.NONE);
- }
+ if (getSession().isDeclareExchanges() &&
!getSession().isResolved(destination))
+ {
+ String name = destination.getExchangeName();
+ ((AMQSession_0_10)
getSession()).getQpidSession().exchangeDeclare
+ (name,
+ destination.getExchangeClass(),
+ null, null,
+ name.startsWith("amq.") ? Option.PASSIVE : Option.NONE,
+ destination.isExchangeDurable() ? Option.DURABLE :
Option.NONE,
+ destination.isExchangeAutoDelete() ? Option.AUTO_DELETE :
Option.NONE);
+
+ getSession().setResolved(destination);
+ }
}
else
{
Modified:
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
(original)
+++
qpid/java/branches/6.0.x/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
Wed May 11 16:31:40 2016
@@ -86,7 +86,7 @@ public class BasicMessageProducer_0_8 ex
}
else
{
- if (getSession().isDeclareExchanges())
+ if (getSession().isDeclareExchanges() &&
!getSession().isResolved(destination))
{
final MethodRegistry methodRegistry =
getSession().getMethodRegistry();
ExchangeDeclareBody body =
@@ -103,6 +103,7 @@ public class BasicMessageProducer_0_8 ex
AMQFrame declare = body.generateFrame(getChannelId());
getConnection().getProtocolHandler().writeFrame(declare);
+ getSession().setResolved(destination);
}
}
}
Modified:
qpid/java/branches/6.0.x/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
URL:
http://svn.apache.org/viewvc/qpid/java/branches/6.0.x/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java?rev=1743394&r1=1743393&r2=1743394&view=diff
==============================================================================
---
qpid/java/branches/6.0.x/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
(original)
+++
qpid/java/branches/6.0.x/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
Wed May 11 16:31:40 2016
@@ -57,6 +57,7 @@ import org.apache.qpid.client.message.Qp
import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
import org.apache.qpid.messaging.Address;
import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.test.utils.BrokerHolder;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
import org.apache.qpid.transport.ExecutionErrorCode;
@@ -301,6 +302,52 @@ public class AddressBasedDestinationTest
createExchangeImpl(true, true, false);
}
+
+ /**
+ * QPID-5816 ensure that a Destination used on a second connection is
resolved again
+ * (creating the queue/exchange if necessary).
+ */
+ public void testResolvedDestinationReresolvedBySecondConnection() throws
Exception
+ {
+ Session session = _connection.createSession(true,
Session.SESSION_TRANSACTED);
+
+ String addr = String.format("ADDR:%s; {create: always, node: {durable:
false}}", getTestQueueName());
+
+ Destination dest = session.createQueue(addr);
+
+ MessageConsumer consumer = session.createConsumer(dest);
+ sendMessage(session, dest, 1);
+ Message m = consumer.receive(1000);
+ assertNotNull("Should receive message sent to queue",m);
+
+ _connection.close();
+
+ stopBroker();
+ startBroker(getFailingPort());
+
+ AMQConnection connection = null;
+ try
+ {
+ connection =
getConnectionFactory("failover").createConnection(GUEST_USERNAME,
GUEST_PASSWORD);
+ connection.start();
+ session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+
+ // Queue should be recreated by re-resolution of the address
+ consumer = session.createConsumer(dest);
+ sendMessage(session, dest, 1);
+ m = consumer.receive(1000);
+ assertNotNull("Should receive message sent to queue", m);
+ }
+ finally
+ {
+ if (connection != null)
+ {
+ connection.close();
+ }
+ stopBroker(getFailingPort());
+ }
+ }
+
private void createExchangeImpl(final boolean withExchangeArgs,
final boolean useNonsenseArguments,
final boolean useNonsenseExchangeType)
throws Exception
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
