diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java index 92d3cac08..194dcdb73 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpFixedProducer.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.concurrent.ScheduledFuture; +import org.apache.qpid.jms.JmsDestination; import org.apache.qpid.jms.message.JmsOutboundMessageDispatch; import org.apache.qpid.jms.meta.JmsConnectionInfo; import org.apache.qpid.jms.meta.JmsProducerInfo; @@ -250,7 +251,11 @@ private void applyDeliveryStateUpdate(InFlightSend send, Delivery delivery, Deli break; case Released: LOG.trace("Outcome of delivery was released: {}", delivery); - deliveryError = new ProviderDeliveryReleasedException("Delivery failed: released by receiver"); + if (isReleasedOutcomeSuccessful(send)) { + send.onSuccess(); + } else { + deliveryError = new ProviderDeliveryReleasedException("Delivery failed: released by receiver"); + } break; case Modified: LOG.trace("Outcome of delivery was modified: {}", delivery); @@ -266,6 +271,28 @@ private void applyDeliveryStateUpdate(InFlightSend send, Delivery delivery, Deli } } + /** + * A released outcome at the target means the transfer "was not and will not be acted upon" + * (AMQP 1.0 section 3.4.4). Unlike rejected or modified it carries no indication that + * anything went wrong, and for a topic it is the expected result of publishing when no + * subscription matched: Jakarta Messaging section 4.2.2 states that a subscription only + * receives a copy of a message sent to the topic while the subscription exists, so a + * message published to a topic with no matching subscription is simply discarded and the + * send succeeds. + * + * For a queue the same outcome means the message was not stored, so it is still reported as + * a send failure rather than silently losing a point-to-point message. + * + * @param send + * the in-flight send whose delivery was released by the remote. + * + * @return true if the released outcome should complete the send successfully. + */ + private boolean isReleasedOutcomeSuccessful(InFlightSend send) { + JmsDestination destination = send.getEnvelope().getDestination(); + return destination != null && destination.isTopic(); + } + public AmqpSession getSession() { return session; } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java index 0680941c7..9515a7cf1 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java @@ -1580,6 +1580,112 @@ private void doSyncSendMessageNotAcceptedTestImpl(ListDescribedType responseStat } } + @Test + @Timeout(20) + public void testSyncSendToTopicMessageReleasedCompletesSuccessfully() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + Connection connection = testFixture.establishConnecton(testPeer); + testPeer.expectBegin(); + testPeer.expectSenderAttach(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic topic = session.createTopic("myTopic"); + MessageProducer producer = session.createProducer(topic); + Message message = session.createTextMessage("content"); + + testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true); + testPeer.expectClose(); + + // A released outcome means the remote did not act on the transfer. For a topic that is + // just "no subscription took a copy of this message", which the JMS pub/sub model + // defines as a successful send, so no exception must escape here. + producer.send(message); + + connection.close(); + + testPeer.waitForAllHandlersToComplete(2000); + } + } + + @Test + @Timeout(20) + public void testAsyncSendToTopicMessageReleasedCompletesSuccessfully() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer); + + final CountDownLatch asyncError = new CountDownLatch(1); + + connection.setForceAsyncSend(true); + connection.setExceptionListener(new ExceptionListener() { + + @Override + public void onException(JMSException exception) { + LOG.debug("ExceptionListener got error: {}", exception.getMessage()); + asyncError.countDown(); + } + }); + + testPeer.expectBegin(); + testPeer.expectSenderAttach(); + testPeer.expectSenderAttach(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic topic = session.createTopic("myTopic"); + MessageProducer producer = session.createProducer(topic); + + // Create a second producer which allows for a safe wait for credit for the + // first producer without the need for a sleep. Otherwise the first producer + // might not do an actual async send due to not having received credit yet. + session.createProducer(topic); + + Message message = session.createTextMessage("content"); + + testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true); + + producer.send(message); + + testPeer.waitForAllHandlersToComplete(2000); + + assertFalse(asyncError.await(50, TimeUnit.MILLISECONDS), + "Released outcome for a topic send should not be reported as an async error"); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(2000); + } + } + + @Test + @Timeout(20) + public void testAsyncCompletionSendToTopicMessageReleasedCompletesSuccessfully() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + Connection connection = testFixture.establishConnecton(testPeer); + + testPeer.expectBegin(); + testPeer.expectSenderAttach(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic topic = session.createTopic("myTopic"); + MessageProducer producer = session.createProducer(topic); + Message message = session.createTextMessage("content"); + + testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true); + + TestJmsCompletionListener listener = new TestJmsCompletionListener(); + producer.send(message, listener); + + assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback"); + assertNull(listener.exception, "Released outcome for a topic send should complete the send"); + assertNotNull(listener.message); + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(2000); + } + } + @Test @Timeout(20) public void testAsyncSendMessageRejected() throws Exception {