diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java index ef075fc227..960942d0bd 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java @@ -20,6 +20,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import org.apache.activemq.artemis.api.core.ActiveMQExceptionType; import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.core.server.QueueQueryResult; import org.apache.activemq.artemis.core.transaction.Transaction; @@ -140,8 +141,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr try { SelectorParser.parse(selector); } catch (FilterException e) { - close(new ErrorCondition(AmqpError.INVALID_FIELD, e.getMessage())); - return; + throw new ActiveMQAMQPException(AmqpError.INVALID_FIELD, "Invalid filter", ActiveMQExceptionType.INVALID_FILTER_EXPRESSION); } supportedFilters.put(filter.getKey(), filter.getValue()); @@ -313,6 +313,9 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr @Override public void close(ErrorCondition condition) throws ActiveMQAMQPException { closed = true; + if (condition != null) { + sender.setCondition(condition); + } protonSession.removeSender(sender); synchronized (connection.getLock()) { sender.close(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveTest.java index 6c50b86639..aae265050e 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveTest.java @@ -45,6 +45,8 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.jms.JMSException; + /** * Test basic send and receive scenarios using only AMQP sender and receiver links. */ @@ -132,6 +134,24 @@ public class AmqpSendReceiveTest extends AmqpClientTestSupport { connection.close(); } + @Test(timeout = 60000) + public void testInvalidFilter() throws Exception { + AmqpClient client = createAmqpClient(); + + AmqpConnection connection = addConnection(client.connect()); + AmqpSession session = connection.createSession(); + + try { + session.createReceiver(getTestName(), "null = 'f''", true); + fail("should throw exception"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof JMSException); + //passed + } + + connection.close(); + } + @Test(timeout = 60000) public void testQueueReceiverReadMessage() throws Exception { sendMessages(getTestName(), 1);