From 95fc93485ab66966713611a4e1429d917dabde64 Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Fri, 17 Oct 2014 14:29:21 +0000 Subject: QPID-6164 : Add synchronous publish capability to 0-8/9/9-1 git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1632585 13f79535-47bb-0310-9956-ffa450edef68 --- .../qpid/server/protocol/v0_8/AMQChannel.java | 119 ++++++++++++++++- .../server/protocol/v0_8/AMQProtocolEngine.java | 14 +- .../protocol/v0_8/UnacknowledgedMessageMap.java | 3 +- .../qpid/client/AMQConnectionDelegate_8_0.java | 66 +++++++++- .../apache/qpid/client/BasicMessageProducer.java | 12 +- .../qpid/client/BasicMessageProducer_0_8.java | 85 +++++++++++- .../qpid/client/failover/FailoverNoopSupport.java | 2 +- .../qpid/client/failover/FailoverRetrySupport.java | 8 +- .../client/handler/ClientMethodDispatcherImpl.java | 24 +++- .../qpid/client/protocol/AMQProtocolHandler.java | 4 +- .../java/org/apache/qpid/codec/ClientDecoder.java | 15 +++ .../java/org/apache/qpid/codec/ServerDecoder.java | 10 ++ .../java/org/apache/qpid/framing/BasicAckBody.java | 2 +- .../org/apache/qpid/framing/BasicNackBody.java | 143 +++++++++++++++++++++ .../qpid/framing/ChannelMethodProcessor.java | 4 + .../qpid/framing/ClientChannelMethodProcessor.java | 1 + .../qpid/framing/ClientMethodDispatcher.java | 2 + .../org/apache/qpid/framing/ConfirmSelectBody.java | 105 +++++++++++++++ .../apache/qpid/framing/ConfirmSelectOkBody.java | 77 +++++++++++ .../qpid/framing/FrameCreatingMethodProcessor.java | 18 +++ .../org/apache/qpid/framing/MethodDispatcher.java | 2 + .../qpid/framing/ServerChannelMethodProcessor.java | 1 + .../qpid/framing/ServerMethodDispatcher.java | 2 + .../qpid/properties/ConnectionStartProperties.java | 2 + .../org/apache/qpid/client/SyncPublishTest.java | 131 +++++++++++++++++++ qpid/java/test-profiles/Java010Excludes | 2 + 26 files changed, 822 insertions(+), 32 deletions(-) create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/framing/BasicNackBody.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/framing/ConfirmSelectBody.java create mode 100644 qpid/java/common/src/main/java/org/apache/qpid/framing/ConfirmSelectOkBody.java create mode 100644 qpid/java/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java (limited to 'qpid/java') diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java index 5ca94891c1..d3ddaa16dd 100644 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java +++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java @@ -201,6 +201,8 @@ public class AMQChannel private final ConfigurationChangeListener _consumerClosedListener = new ConsumerClosedListener(); private final CopyOnWriteArrayList _consumerListeners = new CopyOnWriteArrayList(); private Session _modelObject; + private boolean _confirmOnPublish; + private long _confirmedMessageCounter; public AMQChannel(AMQProtocolEngine connection, int channelId, final MessageStore messageStore) @@ -394,6 +396,11 @@ public class AMQChannel // check and deliver if header says body length is zero if (_currentMessage.allContentReceived()) { + if(_confirmOnPublish) + { + _confirmedMessageCounter++; + } + try { @@ -421,6 +428,10 @@ public class AMQChannel if(!checkMessageUserId(_currentMessage.getContentHeader())) { + if(_confirmOnPublish) + { + _connection.writeFrame(new AMQFrame(_channelId, new BasicNackBody(_confirmedMessageCounter, false, false))); + } _transaction.addPostTransactionAction(new WriteReturnAction(AMQConstant.ACCESS_REFUSED, "Access Refused", amqMessage)); } else @@ -461,6 +472,12 @@ public class AMQChannel } else { + if(_confirmOnPublish) + { + BasicAckBody responseBody = _connection.getMethodRegistry() + .createBasicAckBody(_confirmedMessageCounter, false); + _connection.writeFrame(responseBody.generateFrame(_channelId)); + } incrementOutstandingTxnsIfNecessary(); } } @@ -503,7 +520,7 @@ public class AMQChannel description, mandatory, isTransactional(), closeOnNoRoute)); } - if (mandatory && isTransactional() && _connection.isCloseWhenNoRoute()) + if (mandatory && isTransactional() && !_confirmOnPublish && _connection.isCloseWhenNoRoute()) { _connection.closeConnection(AMQConstant.NO_ROUTE, "No route for message " + currentMessageDescription(), _channelId); @@ -512,6 +529,10 @@ public class AMQChannel { if (mandatory || message.isImmediate()) { + if(_confirmOnPublish) + { + _connection.writeFrame(new AMQFrame(_channelId, new BasicNackBody(_confirmedMessageCounter, false, false))); + } _transaction.addPostTransactionAction(new WriteReturnAction(AMQConstant.NO_ROUTE, "No Route for message " + currentMessageDescription(), @@ -2236,8 +2257,6 @@ public class AMQChannel if (requeue) { - //this requeue represents a message rejected from the pre-dispatch queue - //therefore we need to amend the delivery counter. message.decrementDeliveryCount(); requeue(deliveryTag); @@ -2358,6 +2377,85 @@ public class AMQChannel return _connection.ignoreAllButCloseOk() || _connection.channelAwaitingClosure(_channelId); } + @Override + public void receiveBasicNack(final long deliveryTag, final boolean multiple, final boolean requeue) + { + if(_logger.isDebugEnabled()) + { + _logger.debug("RECV[" + _channelId + "] BasicNack[" +" deliveryTag: " + deliveryTag + " multiple: " + multiple + " requeue: " + requeue + " ]"); + } + + Map nackedMessageMap = new LinkedHashMap<>(); + _unacknowledgedMessageMap.collect(deliveryTag, multiple, nackedMessageMap); + + for(MessageInstance message : nackedMessageMap.values()) + { + + if (message == null) + { + _logger.warn("Ignoring nack request as message is null for tag:" + deliveryTag); + } + else + { + + if (message.getMessage() == null) + { + _logger.warn("Message has already been purged, unable to nack."); + } + else + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Nack-ing: DT:" + deliveryTag + + "-" + message.getMessage() + + ": Requeue:" + requeue + + + " on channel:" + debugIdentity()); + } + + if (requeue) + { + message.decrementDeliveryCount(); + + requeue(deliveryTag); + } + else + { + message.reject(); + + final boolean maxDeliveryCountEnabled = isMaxDeliveryCountEnabled(deliveryTag); + _logger.debug("maxDeliveryCountEnabled: " + + maxDeliveryCountEnabled + + " deliveryTag " + + deliveryTag); + if (maxDeliveryCountEnabled) + { + final boolean deliveredTooManyTimes = isDeliveredTooManyTimes(deliveryTag); + _logger.debug("deliveredTooManyTimes: " + + deliveredTooManyTimes + + " deliveryTag " + + deliveryTag); + if (deliveredTooManyTimes) + { + deadLetter(deliveryTag); + } + else + { + message.incrementDeliveryCount(); + } + } + else + { + requeue(deliveryTag); + } + } + } + } + + } + + } + @Override public void receiveChannelFlow(final boolean active) { @@ -3355,6 +3453,21 @@ public class AMQChannel resend(); } + @Override + public void receiveConfirmSelect(final boolean nowait) + { + if(_logger.isDebugEnabled()) + { + _logger.debug("RECV[" + _channelId + "] ConfirmSelect [ nowait: " + nowait + " ]"); + } + _confirmOnPublish = true; + + if(!nowait) + { + _connection.writeFrame(new AMQFrame(_channelId, ConfirmSelectOkBody.INSTANCE)); + } + } + private void closeChannel(final AMQConstant cause, final String message) { diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java index 413cf49eaf..49db24be52 100644 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java +++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java @@ -85,6 +85,7 @@ import org.apache.qpid.server.util.ConnectionScopedRuntimeException; import org.apache.qpid.server.util.ServerScopedRuntimeException; import org.apache.qpid.server.virtualhost.VirtualHostImpl; import org.apache.qpid.transport.Sender; +import org.apache.qpid.transport.SenderException; import org.apache.qpid.transport.TransportException; import org.apache.qpid.transport.network.NetworkConnection; import org.apache.qpid.util.BytesDataOutput; @@ -432,6 +433,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, String.valueOf(_closeWhenNoRoute)); serverProperties.setString(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED, String.valueOf(_broker.isMessageCompressionEnabled())); + serverProperties.setString(ConnectionStartProperties.QPID_CONFIRMED_PUBLISH_SUPPORTED, Boolean.TRUE.toString()); AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(), (short) pv.getActualMinorVersion(), @@ -1119,9 +1121,17 @@ public class AMQProtocolEngine implements ServerProtocolEngine, _currentClassId, _currentMethodId); - writeFrame(closeBody.generateFrame(0)); + try + { + writeFrame(closeBody.generateFrame(0)); + + _sender.close(); + } + catch(SenderException e) + { + // ignore + } - _sender.close(); } finally { diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java index bd7b070cd2..198b7fe21b 100644 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java +++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java @@ -21,6 +21,7 @@ package org.apache.qpid.server.protocol.v0_8; import java.util.Collection; +import java.util.Map; import java.util.Set; import org.apache.qpid.AMQException; @@ -63,7 +64,7 @@ public interface UnacknowledgedMessageMap Set getDeliveryTags(); Collection acknowledge(long deliveryTag, boolean multiple); - + void collect(long key, boolean multiple, Map msgs); } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java index 176eb5d0c4..bcf0721aab 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java @@ -46,6 +46,8 @@ import org.apache.qpid.common.ServerPropertyNames; import org.apache.qpid.configuration.ClientProperties; import org.apache.qpid.framing.ChannelOpenBody; import org.apache.qpid.framing.ChannelOpenOkBody; +import org.apache.qpid.framing.ConfirmSelectBody; +import org.apache.qpid.framing.ConfirmSelectOkBody; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.framing.ProtocolVersion; import org.apache.qpid.framing.TxSelectBody; @@ -68,6 +70,8 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate private final AMQConnection _conn; private boolean _messageCompressionSupported; private boolean _addrSyntaxSupported; + private boolean _confirmedPublishSupported; + private boolean _confirmedPublishNonTransactionalSupported; public void closeConnection(long timeout) throws JMSException, AMQException { @@ -94,6 +98,11 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate return ((cause instanceof ConnectException) || (cause instanceof UnresolvedAddressException)); } + public boolean isConfirmedPublishSupported() + { + return _confirmedPublishSupported; + } + public ProtocolVersion makeBrokerConnection(BrokerDetails brokerDetail) throws AMQException, IOException { if (_logger.isDebugEnabled()) @@ -146,6 +155,8 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate _conn.setConnected(true); _conn.logConnected(network.getLocalAddress(), network.getRemoteAddress()); _messageCompressionSupported = checkMessageCompressionSupported(); + _confirmedPublishSupported = checkConfirmedPublishSupported(); + _confirmedPublishNonTransactionalSupported = checkConfirmedPublishNonTransactionalSupported(); return null; } else @@ -155,6 +166,32 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate } + // RabbitMQ supports confirmed publishing, but only on non transactional sessions + private boolean checkConfirmedPublishNonTransactionalSupported() + { + FieldTable serverProperties = _conn.getProtocolHandler().getProtocolSession().getConnectionStartServerProperties(); + if( serverProperties != null + && serverProperties.containsKey("capabilities") + && serverProperties.get("capabilities") instanceof FieldTable) + { + FieldTable capabilities = serverProperties.getFieldTable("capabilities"); + if(capabilities.containsKey("publisher_confirms") + && capabilities.get("publisher_confirms") instanceof Boolean + && capabilities.getBoolean("publisher_confirms")) + { + return true; + } + else + { + return false; + } + } + else + { + return false; + } + } + public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode, final int prefetch) throws JMSException { @@ -266,9 +303,21 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate } TxSelectBody body = _conn.getProtocolHandler().getMethodRegistry().createTxSelectBody(); - // TODO: Be aware of possible changes to parameter order as versions change. + _conn.getProtocolHandler().syncWrite(body.generateFrame(channelId), TxSelectOkBody.class); } + boolean useConfirms = (_confirmedPublishSupported || (!transacted && _confirmedPublishNonTransactionalSupported)) + && "all".equals(_conn.getSyncPublish()); + if(useConfirms) + { + if (_logger.isDebugEnabled()) + { + _logger.debug("Issuing ConfirmSelect for " + channelId); + } + ConfirmSelectBody body = new ConfirmSelectBody(false); + + _conn.getProtocolHandler().syncWrite(body.generateFrame(channelId), ConfirmSelectOkBody.class); + } } public void failoverPrep() @@ -340,7 +389,7 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate } catch (IllegalStateException e) { - if (!(e.getMessage().startsWith("Fail-over interupted no-op failover support"))) + if (!(e.getMessage().startsWith("Fail-over interrupted no-op failover support"))) { throw e; } @@ -424,6 +473,14 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate } + private boolean checkConfirmedPublishSupported() + { + FieldTable serverProperties = _conn.getProtocolHandler().getProtocolSession().getConnectionStartServerProperties(); + return serverProperties != null + && Boolean.parseBoolean(serverProperties.getString(ConnectionStartProperties.QPID_CONFIRMED_PUBLISH_SUPPORTED)); + + } + public boolean isMessageCompressionSupported() { return _messageCompressionSupported; @@ -433,4 +490,9 @@ public class AMQConnectionDelegate_8_0 implements AMQConnectionDelegate { return _addrSyntaxSupported; } + + public boolean isConfirmedPublishNonTransactionalSupported() + { + return _confirmedPublishNonTransactionalSupported; + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java index 1d47ce9a07..f72d915c25 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java @@ -122,7 +122,7 @@ public abstract class BasicMessageProducer extends Closeable implements org.apac ? System.getProperty("qpid.default_mandatory") : "false")); - private PublishMode publishMode = PublishMode.ASYNC_PUBLISH_ALL; + private PublishMode _publishMode = PublishMode.ASYNC_PUBLISH_ALL; protected BasicMessageProducer(Logger logger,AMQConnection connection, AMQDestination destination, boolean transacted, int channelId, AMQSession session, long producerId, Boolean immediate, Boolean mandatory) throws AMQException @@ -165,16 +165,16 @@ public abstract class BasicMessageProducer extends Closeable implements org.apac // Support for deprecated option sync_persistence if (syncPub.equals("persistent") || _connection.getSyncPersistence()) { - publishMode = PublishMode.SYNC_PUBLISH_PERSISTENT; + _publishMode = PublishMode.SYNC_PUBLISH_PERSISTENT; } else if (syncPub.equals("all")) { - publishMode = PublishMode.SYNC_PUBLISH_ALL; + _publishMode = PublishMode.SYNC_PUBLISH_ALL; } if (_logger.isDebugEnabled()) { - _logger.debug("MessageProducer " + toString() + " using publish mode : " + publishMode); + _logger.debug("MessageProducer " + toString() + " using publish mode : " + _publishMode); } } @@ -720,12 +720,12 @@ public abstract class BasicMessageProducer extends Closeable implements org.apac protected PublishMode getPublishMode() { - return publishMode; + return _publishMode; } protected void setPublishMode(PublishMode publishMode) { - this.publishMode = publishMode; + _publishMode = publishMode; } Logger getLogger() diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java index 69d02566bf..e4c879aca8 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java @@ -32,14 +32,19 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.qpid.AMQException; +import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.message.AMQMessageDelegate_0_8; import org.apache.qpid.client.message.AbstractJMSMessage; import org.apache.qpid.client.message.QpidMessageProperties; import org.apache.qpid.client.protocol.AMQProtocolHandler; +import org.apache.qpid.client.protocol.BlockingMethodFrameListener; import org.apache.qpid.configuration.ClientProperties; import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.BasicAckBody; import org.apache.qpid.framing.BasicContentHeaderProperties; +import org.apache.qpid.framing.BasicNackBody; import org.apache.qpid.framing.BasicPublishBody; import org.apache.qpid.framing.CompositeAMQDataBlock; import org.apache.qpid.framing.ContentBody; @@ -211,9 +216,6 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer } - // TODO: This is a hacky way of getting the AMQP class-id for the Basic class - int classIfForBasic = getSession().getMethodRegistry().createBasicQosOkBody().getClazz(); - AMQFrame contentHeaderFrame = ContentHeaderBody.createAMQFrame(getChannelId(), contentHeaderProperties, size); @@ -232,7 +234,7 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer frames[0] = publishFrame; frames[1] = contentHeaderFrame; - CompositeAMQDataBlock compositeFrame = new CompositeAMQDataBlock(frames); + final CompositeAMQDataBlock compositeFrame = new CompositeAMQDataBlock(frames); try { @@ -246,7 +248,40 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer throw jmse; } - getConnection().getProtocolHandler().writeFrame(compositeFrame); + AMQConnectionDelegate_8_0 connectionDelegate80 = (AMQConnectionDelegate_8_0) (getConnection().getDelegate()); + + boolean useConfirms = getPublishMode() == PublishMode.SYNC_PUBLISH_ALL + && (connectionDelegate80.isConfirmedPublishSupported() + || (!getSession().isTransacted() && connectionDelegate80.isConfirmedPublishNonTransactionalSupported())); + + if(!useConfirms) + { + getConnection().getProtocolHandler().writeFrame(compositeFrame); + } + else + { + final PublishConfirmMessageListener frameListener = new PublishConfirmMessageListener(getChannelId()); + try + { + + getConnection().getProtocolHandler().writeCommandFrameAndWaitForReply(compositeFrame, + frameListener); + + if(frameListener.isRejected()) + { + throw new JMSException("The message was not accepted by the server (e.g. because the address was no longer valid)"); + } + } + catch (AMQException e) + { + throw new JMSAMQException(e); + } + catch (FailoverException e) + { + throw new JMSAMQException("Fail-over interrupted send. Status of the send is uncertain.", e); + + } + } } /** @@ -290,7 +325,7 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer private int calculateContentBodyFrameCount(ByteBuffer payload) { - // we substract one from the total frame maximum size to account for the end of frame marker in a body frame + // we subtract one from the total frame maximum size to account for the end of frame marker in a body frame // (0xCE byte). int frameCount; if ((payload == null) || (payload.remaining() == 0)) @@ -313,4 +348,42 @@ public class BasicMessageProducer_0_8 extends BasicMessageProducer { return (AMQSession_0_8) super.getSession(); } + + private static class PublishConfirmMessageListener extends BlockingMethodFrameListener + { + private boolean _rejected; + + /** + * Creates a new method listener, that filters incoming method to just those that match the specified channel id. + * + * @param channelId The channel id to filter incoming methods with. + */ + public PublishConfirmMessageListener(final int channelId) + { + super(channelId); + } + + @Override + public boolean processMethod(final int channelId, final AMQMethodBody frame) + { + if (frame instanceof BasicAckBody) + { + return true; + } + else if (frame instanceof BasicNackBody) + { + _rejected = true; + return true; + } + else + { + return false; + } + } + + public boolean isRejected() + { + return _rejected; + } + } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverNoopSupport.java b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverNoopSupport.java index a69e808880..da17bedcfd 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverNoopSupport.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverNoopSupport.java @@ -68,7 +68,7 @@ public class FailoverNoopSupport implements FailoverSupp } catch (FailoverException e) { - throw new IllegalStateException("Fail-over interupted no-op failover support. " + throw new IllegalStateException("Fail-over interrupted no-op failover support. " + "No-op support should only be used where the caller is certain fail-over cannot occur.", e); } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverRetrySupport.java b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverRetrySupport.java index ffe0baecd8..74bf9a54fd 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverRetrySupport.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverRetrySupport.java @@ -45,7 +45,7 @@ import org.apache.qpid.client.AMQConnection; *

* Wrapping a synchronous method in a FailoverRetrySupport will have the effect that the operation will not be * started during fail-over, but be delayed until any current fail-over has completed. Should a fail-over process want - * to start whilst waiting for the synchrnous reply, the FailoverRetrySupport will detect this and rety the operation + * to start whilst waiting for the synchronous reply, the FailoverRetrySupport will detect this and retry the operation * until it succeeds. Synchronous methods are usually coordinated with a * {@link org.apache.qpid.client.protocol.BlockingMethodFrameListener} which is notified when a fail-over process wants * to start and throws a FailoverException in response to this. @@ -53,12 +53,6 @@ import org.apache.qpid.client.AMQConnection; * Wrapping an asynchronous method in a FailoverRetrySupport will have the effect that the operation will not be * started during fail-over, but be delayed until any current fail-over has completed. *

- * TODO Another continuation. Could use an interface Continuation (as described in other todos) - * Then have a wrapping continuation (this), which blocks on an arbitrary - * Condition or Latch (specified in constructor call), that this blocks on before calling the wrapped Continuation. - * Must work on Java 1.4, so check retrotranslator works on Lock/Condition or latch first. Argument and return type - * to match wrapped condition as type parameters. Rename to AsyncConditionalContinuation or something like that. - *

* TODO InterruptedException not handled well. */ public class FailoverRetrySupport implements FailoverSupport diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java index e6eb2d814f..de2f2f52a9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java @@ -53,6 +53,7 @@ public class ClientMethodDispatcherImpl implements MethodDispatcher private static final Logger _logger = LoggerFactory.getLogger(ClientMethodDispatcherImpl.class); + private static interface DispatcherFactory { public ClientMethodDispatcherImpl createMethodDispatcher(AMQProtocolSession session); @@ -147,6 +148,13 @@ public class ClientMethodDispatcherImpl implements MethodDispatcher return false; } + @Override + public boolean dispatchConfirmSelectOk(final ConfirmSelectOkBody confirmSelectOkBody, final int channelId) + throws AMQException + { + return false; + } + public boolean dispatchBasicCancelOk(BasicCancelOkBody body, int channelId) throws AMQException { _basicCancelOkMethodHandler.methodReceived(_session, body, channelId); @@ -271,11 +279,19 @@ public class ClientMethodDispatcherImpl implements MethodDispatcher throw new AMQMethodNotImplementedException(body); } + @Override public boolean dispatchBasicAck(BasicAckBody body, int channelId) throws AMQException { - throw new AMQMethodNotImplementedException(body); + return false; } + @Override + public boolean dispatchBasicNack(final BasicNackBody basicNackBody, final int channelId) + { + return false; + } + + public boolean dispatchBasicCancel(BasicCancelBody body, int channelId) throws AMQException { throw new AMQMethodNotImplementedException(body); @@ -400,6 +416,12 @@ public class ClientMethodDispatcherImpl implements MethodDispatcher return false; } + @Override + public boolean dispatchConfirmSelect(final ConfirmSelectBody body, final int channelId) throws AMQException + { + throw new AMQMethodNotImplementedException(body); + } + public boolean dispatchExchangeBoundOk(ExchangeBoundOkBody body, int channelId) throws AMQException { _exchangeBoundOkMethodHandler.methodReceived(_session, body, channelId); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index bb98c0abbd..4886eabb90 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -661,7 +661,7 @@ public class AMQProtocolHandler implements ProtocolEngine * @param frame * @param listener the blocking listener. Note the calling thread will block. */ - public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQFrame frame, BlockingMethodFrameListener listener) + public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQDataBlock frame, BlockingMethodFrameListener listener) throws AMQException, FailoverException { return writeCommandFrameAndWaitForReply(frame, listener, DEFAULT_SYNC_TIMEOUT); @@ -674,7 +674,7 @@ public class AMQProtocolHandler implements ProtocolEngine * @param frame * @param listener the blocking listener. Note the calling thread will block. */ - public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQFrame frame, BlockingMethodFrameListener listener, + public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQDataBlock frame, BlockingMethodFrameListener listener, long timeout) throws AMQException, FailoverException { try diff --git a/qpid/java/common/src/main/java/org/apache/qpid/codec/ClientDecoder.java b/qpid/java/common/src/main/java/org/apache/qpid/codec/ClientDecoder.java index 5048193cac..9bdc1dd889 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/codec/ClientDecoder.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/codec/ClientDecoder.java @@ -209,6 +209,9 @@ public class ClientDecoder extends AMQDecoder options = new HashMap<>(); + options.put(ConnectionURL.OPTIONS_SYNC_PUBLISH, "all"); + _connection = getConnectionWithOptions(options); + } + + @Override + public void tearDown() throws Exception + { + _connection.close(); + super.tearDown(); + } + + public void testAnonPublisherUnknownDestination() throws Exception + { + Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer producer = session.createProducer(null); + try + { + producer.send(session.createQueue("direct://amq.direct/unknown/unknown"),session.createTextMessage("hello")); + fail("Send to unknown destination should result in error"); + } + catch (JMSException e) + { + // pass + } + } + + + public void testAnonPublisherUnknownDestinationTransactional() throws Exception + { + Session session = _connection.createSession(true, Session.SESSION_TRANSACTED); + MessageProducer producer = session.createProducer(null); + try + { + producer.send(session.createQueue("direct://amq.direct/unknown/unknown"),session.createTextMessage("hello")); + fail("Send to unknown destination should result in error"); + } + catch (JMSException e) + { + // pass + } + try + { + session.commit(); + } + catch (JMSException e) + { + fail("session should commit successfully even though the message was not sent"); + } + + } + + public void testQueueRemovedAfterConsumerCreated() throws JMSException + { + Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + TemporaryQueue queue = session.createTemporaryQueue(); + MessageProducer producer = session.createProducer(queue); + try + { + producer.send(session.createTextMessage("hello")); + } + catch (JMSException e) + { + fail("Send to temporary queue should succeed"); + } + + try + { + queue.delete(); + } + catch (JMSException e) + { + fail("temporary queue should be deletable"); + } + + try + { + producer.send(session.createTextMessage("hello")); + fail("Send to deleted temporary queue should not succeed"); + } + catch (JMSException e) + { + // pass + } + + + } +} diff --git a/qpid/java/test-profiles/Java010Excludes b/qpid/java/test-profiles/Java010Excludes index ce7c5fa151..136bc7918f 100755 --- a/qpid/java/test-profiles/Java010Excludes +++ b/qpid/java/test-profiles/Java010Excludes @@ -76,3 +76,5 @@ org.apache.qpid.systest.management.jmx.QueueManagementTest#testExclusiveQueueHas org.apache.qpid.test.unit.client.AMQSessionTest#testQueueDepthForQueueThatDoesNotExistLegacyBehaviour_08_091 org.apache.qpid.client.prefetch.PrefetchBehaviourTest#testPrefetchWindowExpandsOnReceiveTransaction + +org.apache.qpid.client.SyncPublishTest#* -- cgit v1.2.1