You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2016/05/10 16:41:09 UTC
svn commit: r1743228 - in /qpid/java/trunk:
client/src/main/java/org/apache/qpid/client/
systests/src/test/java/org/apache/qpid/test/client/destination/
Author: kwall
Date: Tue May 10 16:41:09 2016
New Revision: 1743228
URL: http://svn.apache.org/viewvc?rev=1743228&view=rev
Log:
QPID-5816: [Java Client] Maintain a per-session (weak) cache of resolved Destinations
Modified:
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQDestination.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
Modified: qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQDestination.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQDestination.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQDestination.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQDestination.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession.java Tue May 10 16:41:09 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. */
@@ -588,7 +591,7 @@ public abstract class AMQSession<C exten
boolean isConsumer,
boolean noLocal) throws QpidException
{
- if (dest.isAddressResolved() && dest.isResolvedAfter(getAMQConnection().getLastFailoverTime()))
+ if (isResolved(dest))
{
return;
}
@@ -640,10 +643,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;
@@ -1057,8 +1080,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
{
@@ -2372,6 +2394,7 @@ public abstract class AMQSession<C exten
_prefetchedMessageTags.clear();
_rollbackMark.set(-1);
+ clearResolvedDestinations();
resubscribeProducers();
resubscribeConsumers();
}
Modified: qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java Tue May 10 16:41:09 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/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java Tue May 10 16:41:09 2016
@@ -83,17 +83,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/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java (original)
+++ qpid/java/trunk/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java Tue May 10 16:41:09 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/trunk/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java?rev=1743228&r1=1743227&r2=1743228&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java Tue May 10 16:41:09 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,56 @@ 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();
+
+ stopDefaultBroker();
+
+ final BrokerHolder alternativeBroker = createSpawnedBroker();
+ alternativeBroker.start();
+ final int alternativePort = alternativeBroker.getAmqpPort();
+
+ setTestSystemProperty("test.port.alt", String.valueOf(alternativePort));
+
+ 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();
+ }
+ }
+ }
+
private void createExchangeImpl(final boolean withExchangeArgs,
final boolean useNonsenseArguments,
final boolean useNonsenseExchangeType) throws Exception
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org