You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2017/06/08 15:08:17 UTC
qpid-broker-j git commit: QPID-7635: If ANONYMOUS-RELAY finds the
destination it should defer the delivery outcome to the destination
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 9c149b3af -> 205909415
QPID-7635: If ANONYMOUS-RELAY finds the destination it should defer the delivery outcome to the destination
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/20590941
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/20590941
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/20590941
Branch: refs/heads/master
Commit: 205909415d664c3a15d02ab53e6aa163129c4a5b
Parents: 9c149b3
Author: Alex Rudyy <or...@apache.org>
Authored: Thu Jun 8 16:08:02 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Thu Jun 8 16:08:02 2017 +0100
----------------------------------------------------------------------
.../v1_0/AnonymousRelayDestination.java | 204 +++++++++++++++++++
.../protocol/v1_0/ExchangeDestination.java | 18 +-
.../protocol/v1_0/NodeReceivingDestination.java | 19 +-
.../server/protocol/v1_0/QueueDestination.java | 15 +-
.../protocol/v1_0/ReceivingDestination.java | 6 +-
.../qpid/server/protocol/v1_0/Session_1_0.java | 12 +-
.../v1_0/StandardReceivingLinkEndpoint.java | 6 +-
.../org/apache/qpid/client/SyncPublishTest.java | 42 ----
.../qpid/systest/AnonymousProducerTest.java | 116 +++++++++++
test-profiles/Java010Excludes | 6 +-
10 files changed, 343 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AnonymousRelayDestination.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AnonymousRelayDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AnonymousRelayDestination.java
new file mode 100644
index 0000000..af07fec
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AnonymousRelayDestination.java
@@ -0,0 +1,204 @@
+package org.apache.qpid.server.protocol.v1_0;/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+import static org.apache.qpid.server.protocol.v1_0.Session_1_0.DELAYED_DELIVERY;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.logging.messages.ExchangeMessages;
+import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.NamedAddressSpace;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.protocol.v1_0.type.Outcome;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
+import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
+import org.apache.qpid.server.security.SecurityToken;
+import org.apache.qpid.server.txn.ServerTransaction;
+
+public class AnonymousRelayDestination implements ReceivingDestination
+{
+ private final Target _target;
+ private final NamedAddressSpace _addressSpace;
+ private final EventLogger _eventLogger;
+ private final boolean _discardUnroutable;
+
+ AnonymousRelayDestination(final NamedAddressSpace addressSpace,
+ final Target target,
+ final EventLogger eventLogger)
+ {
+ _addressSpace = addressSpace;
+ _target = target;
+ _eventLogger = eventLogger;
+ _discardUnroutable = target.getCapabilities() != null && Arrays.asList(target.getCapabilities())
+ .contains(DISCARD_UNROUTABLE);
+ }
+
+ @Override
+ public Symbol[] getCapabilities()
+ {
+ return new Symbol[]{DELAYED_DELIVERY};
+ }
+
+ @Override
+ public Outcome[] getOutcomes()
+ {
+ return new Outcome[0];
+ }
+
+ @Override
+ public Outcome send(final ServerMessage<?> message,
+ final String routingAddress,
+ final ServerTransaction txn,
+ final SecurityToken securityToken)
+ {
+ final ReceivingDestination destination;
+ if (!routingAddress.startsWith("/") && routingAddress.contains("/"))
+ {
+ String[] parts = routingAddress.split("/", 2);
+ MessageDestination exchangeDestination = _addressSpace.getAttainedMessageDestination(parts[0]);
+ if (exchangeDestination instanceof Exchange)
+ {
+ Symbol[] capabilities = _target.getCapabilities();
+ destination = new ExchangeDestination(((Exchange<?>) exchangeDestination),
+ null,
+ _target.getDurable(),
+ _target.getExpiryPolicy(),
+ parts[0],
+ parts[1],
+ capabilities != null ? Arrays.asList(capabilities) : Collections.emptyList());
+ }
+ else
+ {
+ destination = null;
+ }
+ }
+ else
+ {
+ MessageDestination messageDestination = _addressSpace.getAttainedMessageDestination(routingAddress);
+ if (messageDestination != null)
+ {
+ destination = new NodeReceivingDestination(messageDestination,
+ _target.getDurable(),
+ _target.getExpiryPolicy(),
+ routingAddress,
+ _target.getCapabilities(),
+ _eventLogger);
+ }
+ else
+ {
+ MessageSource source = _addressSpace.getAttainedMessageSource(routingAddress);
+ if (source instanceof Queue)
+ {
+ destination = new QueueDestination((Queue<?>) source, routingAddress);
+ }
+ else
+ {
+ destination = null;
+ }
+ }
+ }
+
+ final Outcome outcome;
+ if (destination == null)
+ {
+ if (_discardUnroutable)
+ {
+ _eventLogger.message(ExchangeMessages.DISCARDMSG("", routingAddress));
+ outcome = new Accepted();
+ }
+ else
+ {
+ outcome = createdRejectedOutcome(AmqpError.NOT_FOUND, "Unknown destination '" + routingAddress + '"');
+ }
+ }
+ else
+ {
+ outcome = destination.send(message, routingAddress, txn, securityToken);
+ }
+ return outcome;
+ }
+
+ @Override
+ public int getCredit()
+ {
+ // TODO - fix
+ return 20000;
+ }
+
+ @Override
+ public String getRoutingAddress(final Message_1_0 message)
+ {
+ String routingAddress;
+ MessageMetaData_1_0.MessageHeader_1_0 messageHeader = message.getMessageHeader();
+ final String to = messageHeader.getTo();
+ if (to != null)
+ {
+ routingAddress = to;
+ }
+ else if (messageHeader.getHeader("routing-key") instanceof String)
+ {
+ routingAddress = (String) messageHeader.getHeader("routing-key");
+ }
+ else if (messageHeader.getHeader("routing_key") instanceof String)
+ {
+ routingAddress = (String) messageHeader.getHeader("routing_key");
+ }
+ else if (messageHeader.getSubject() != null)
+ {
+ routingAddress = messageHeader.getSubject();
+ }
+ else
+ {
+ routingAddress = "";
+ }
+
+ return routingAddress;
+ }
+
+ @Override
+ public String getAddress()
+ {
+ return "";
+ }
+
+ @Override
+ public MessageDestination getMessageDestination()
+ {
+ return null;
+ }
+
+ private Outcome createdRejectedOutcome(AmqpError errorCode, String errorMessage)
+ {
+ Rejected rejected = new Rejected();
+ final Error notFoundError = new Error(errorCode, errorMessage);
+ rejected.setError(notFoundError);
+ return rejected;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
index 8a0b8bf..252a111 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
@@ -29,7 +29,6 @@ import java.util.List;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
@@ -44,7 +43,6 @@ import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public class ExchangeDestination extends QueueDestination
{
@@ -100,8 +98,10 @@ public class ExchangeDestination extends QueueDestination
public Outcome send(final ServerMessage<?> message,
final String routingAddress,
ServerTransaction txn,
- final Action<MessageInstance> action)
+ final SecurityToken securityToken)
{
+ _exchange.authorisePublish(securityToken, Collections.singletonMap("routingKey", routingAddress));
+
final InstanceProperties instanceProperties =
new InstanceProperties()
{
@@ -128,7 +128,7 @@ public class ExchangeDestination extends QueueDestination
final RoutingResult result = _exchange.route(message,
routingAddress,
instanceProperties);
- int enqueues = result.send(txn, action);
+ int enqueues = result.send(txn, null);
if(enqueues == 0)
{
@@ -153,16 +153,6 @@ public class ExchangeDestination extends QueueDestination
}
@Override
- public void authorizePublish(final SecurityToken securityToken,
- final String routingAddress)
- {
- _exchange.authorisePublish(securityToken,
- Collections.<String,Object>singletonMap("routingKey", routingAddress));
-
-
- }
-
- @Override
public String getRoutingAddress(final Message_1_0 message)
{
String routingAddress;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
index 9bc6323..2595b1c 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/NodeReceivingDestination.java
@@ -29,7 +29,6 @@ import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Exchange;
@@ -43,7 +42,6 @@ import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public class NodeReceivingDestination implements ReceivingDestination
{
@@ -82,8 +80,10 @@ public class NodeReceivingDestination implements ReceivingDestination
public Outcome send(final ServerMessage<?> message,
final String routingAddress, ServerTransaction txn,
- final Action<MessageInstance> action)
+ final SecurityToken securityToken)
{
+ _destination.authorisePublish(securityToken, Collections.singletonMap("routingKey", routingAddress));
+
final InstanceProperties instanceProperties =
new InstanceProperties()
{
@@ -108,7 +108,7 @@ public class NodeReceivingDestination implements ReceivingDestination
}};
RoutingResult result = _destination.route(message, routingAddress, instanceProperties);
- int enqueues = result.send(txn, action);
+ int enqueues = result.send(txn, null);
if(enqueues == 0)
{
@@ -152,19 +152,10 @@ public class NodeReceivingDestination implements ReceivingDestination
}
@Override
- public void authorizePublish(final SecurityToken securityToken,
- final String routingAddress)
- {
- _destination.authorisePublish(securityToken,
- Collections.<String, Object>singletonMap("routingKey", routingAddress));
-
- }
-
- @Override
public String getRoutingAddress(final Message_1_0 message)
{
- MessageMetaData_1_0.MessageHeader_1_0 messageHeader = message.getMessageHeader();
String routingAddress;
+ MessageMetaData_1_0.MessageHeader_1_0 messageHeader = message.getMessageHeader();
final String to = messageHeader.getTo();
if (to != null
&& (_destination.getName() == null || _destination.getName().trim().equals("")))
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
index 834bd51..e133206 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
@@ -23,7 +23,6 @@ package org.apache.qpid.server.protocol.v1_0;
import java.util.Collections;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Queue;
@@ -32,7 +31,6 @@ import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public class QueueDestination extends MessageSourceDestination implements ReceivingDestination
{
@@ -53,8 +51,9 @@ public class QueueDestination extends MessageSourceDestination implements Receiv
public Outcome send(final ServerMessage<?> message,
final String routingAddress, ServerTransaction txn,
- final Action<MessageInstance> action)
+ final SecurityToken securityToken)
{
+ getQueue().authorisePublish(securityToken, Collections.emptyMap());
txn.enqueue(getMessageSource(), message, new ServerTransaction.EnqueueAction()
{
@@ -65,7 +64,7 @@ public class QueueDestination extends MessageSourceDestination implements Receiv
{
try
{
- getQueue().enqueue(message, action, records[0]);
+ getQueue().enqueue(message, null, records[0]);
}
finally
{
@@ -101,14 +100,6 @@ public class QueueDestination extends MessageSourceDestination implements Receiv
}
@Override
- public void authorizePublish(final SecurityToken securityToken,
- final String routingAddress)
- {
-
- getQueue().authorisePublish(securityToken, Collections.<String, Object>emptyMap());
- }
-
- @Override
public String getAddress()
{
return _address;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingDestination.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingDestination.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingDestination.java
index 89c030b..6763f10 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingDestination.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingDestination.java
@@ -21,14 +21,12 @@
package org.apache.qpid.server.protocol.v1_0;
import org.apache.qpid.server.message.MessageDestination;
-import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
import org.apache.qpid.server.protocol.v1_0.type.Symbol;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
public interface ReceivingDestination extends Destination
{
@@ -42,7 +40,7 @@ public interface ReceivingDestination extends Destination
Outcome send(ServerMessage<?> message,
final String routingAddress,
ServerTransaction txn,
- final Action<MessageInstance> postEnqueueAction);
+ final SecurityToken securityToken);
int getCredit();
@@ -50,7 +48,5 @@ public interface ReceivingDestination extends Destination
String getAddress();
- void authorizePublish(SecurityToken securityToken, final String routingAddress);
-
MessageDestination getMessageDestination();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index f75df81..e2044e6 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -701,11 +701,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
String addr = target.getAddress();
if (addr == null || "".equals(addr.trim()))
{
- MessageDestination messageDestination = getAddressSpace().getDefaultDestination();
- destination = new NodeReceivingDestination(messageDestination, target.getDurable(),
- target.getExpiryPolicy(), "",
- target.getCapabilities(),
- _connection.getEventLogger());
+ destination = new AnonymousRelayDestination(getAddressSpace(), target, _connection.getEventLogger());
}
else if (!addr.startsWith("/") && addr.contains("/"))
{
@@ -761,11 +757,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
destination = null;
}
- if (destination != null)
- {
- target.setCapabilities(destination.getCapabilities());
- }
- else
+ if (destination == null)
{
throw new AmqpErrorException(AmqpError.NOT_FOUND,
String.format("Could not find destination for target '%s'", target));
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
index 0282813..199ecdf 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java
@@ -276,9 +276,9 @@ public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
final long arrivalTime = serverMessage.getArrivalTime();
session.getAMQPConnection()
.checkAuthorizedMessagePrincipal(serverMessage.getMessageHeader().getUserId());
- getReceivingDestination().authorizePublish(session.getSecurityToken(), routingAddress);
- Outcome outcome = getReceivingDestination().send(serverMessage, routingAddress, transaction, null);
+ Outcome outcome = getReceivingDestination().send(serverMessage, routingAddress, transaction,
+ session.getSecurityToken());
Source source = getSource();
DeliveryState resultantState;
@@ -592,6 +592,8 @@ public class StandardReceivingLinkEndpoint extends AbstractReceivingLinkEndpoint
Source source = (Source) attach.getSource();
Target target = getTarget();
+ // TODO: This seems a bit weird. Similar code is in attachReceived.
+ // We also seem to send back a different target than we are using.
final ReceivingDestination destination = getSession().getReceivingDestination(getLink(), getTarget());
target.setCapabilities(destination.getCapabilities());
setCapabilities(Arrays.asList(destination.getCapabilities()));
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java
----------------------------------------------------------------------
diff --git a/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java b/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java
index d4268e5..14716ac 100644
--- a/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java
+++ b/systests/src/test/java/org/apache/qpid/client/SyncPublishTest.java
@@ -48,48 +48,6 @@ public class SyncPublishTest extends QpidBrokerTestCase
super.tearDown();
}
- public void testAnonPublisherUnknownDestination() throws Exception
- {
- Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- MessageProducer producer = session.createProducer(null);
- try
- {
- final Queue queue = session.createQueue(isBroker10() ? "amq.direct/unknown" : "direct://amq.direct/unknown/unknown");
- producer.send(queue, 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
- {
- final Queue queue = session.createQueue(isBroker10() ? "amq.direct/unknown" : "direct://amq.direct/unknown/unknown");
- producer.send(queue,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);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/systests/src/test/java/org/apache/qpid/systest/AnonymousProducerTest.java
----------------------------------------------------------------------
diff --git a/systests/src/test/java/org/apache/qpid/systest/AnonymousProducerTest.java b/systests/src/test/java/org/apache/qpid/systest/AnonymousProducerTest.java
new file mode 100644
index 0000000..2550fdb
--- /dev/null
+++ b/systests/src/test/java/org/apache/qpid/systest/AnonymousProducerTest.java
@@ -0,0 +1,116 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.systest;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class AnonymousProducerTest extends QpidBrokerTestCase
+{
+
+ public void testPublishIntoDestinationBoundWithNotMatchingFilter() throws Exception
+ {
+ final Connection connection = getConnection();
+ connection.start();
+
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer messageProducer = session.createProducer(null);
+
+ Topic topic = createTopicOnFanout(connection, getTestName());
+
+ session.createConsumer(topic, "id>1");
+ TextMessage test = session.createTextMessage("testMessage");
+ test.setIntProperty("id", 1);
+ try
+ {
+ messageProducer.send(topic, test);
+ session.commit();
+ }
+ catch (JMSException e)
+ {
+ fail("Exception should not be thrown: " + e.getMessage());
+ }
+ }
+
+ public void testPublishIntoNonExistingTopic() throws Exception
+ {
+ final Connection connection = getConnection();
+ connection.start();
+
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer messageProducer = session.createProducer(null);
+
+ try
+ {
+ messageProducer.send(createTopicOnFanout(connection, "nonExistingTopic"),
+ session.createTextMessage("testMessage"));
+ session.commit();
+ }
+ catch (JMSException e)
+ {
+ fail("Message should be silently discarded. Exception should not be thrown: " + e.getMessage());
+ }
+ }
+
+ public void testPublishIntoNonExistingQueue() throws Exception
+ {
+ final Connection connection = getConnection();
+ connection.start();
+
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer messageProducer = session.createProducer(null);
+
+ try
+ {
+ messageProducer.send(session.createQueue("nonExistingQueue"), session.createTextMessage("testMessage"));
+ session.commit();
+ fail("Expected exception was not thrown");
+ }
+ catch (JMSException e)
+ {
+ // pass
+ }
+ }
+
+ public void testSyncPublishIntoNonExistingQueue() throws Exception
+ {
+ Session session = getConnectionWithSyncPublishing().createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(null);
+ try
+ {
+ final Queue queue = session.createQueue("nonExistingQueue");
+ producer.send(queue, session.createTextMessage("hello"));
+ fail("Send to unknown destination should result in error");
+ }
+ catch (JMSException e)
+ {
+ // pass
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/20590941/test-profiles/Java010Excludes
----------------------------------------------------------------------
diff --git a/test-profiles/Java010Excludes b/test-profiles/Java010Excludes
index e1bfec1..19463d6 100755
--- a/test-profiles/Java010Excludes
+++ b/test-profiles/Java010Excludes
@@ -69,8 +69,6 @@ org.apache.qpid.test.unit.client.AMQSessionTest#testQueueDepthForQueueThatDoesNo
org.apache.qpid.client.prefetch.PrefetchBehaviourTest#testPrefetchWindowExpandsOnReceiveTransaction
-org.apache.qpid.client.SyncPublishTest#*
-
// QPID-6722: Race client side means that session close can end in exception when failover is in progress.
org.apache.qpid.client.failover.FailoverBehaviourTest#testConnectionCloseInterruptsFailover
org.apache.qpid.client.failover.AddressBasedFailoverBehaviourTest#testConnectionCloseInterruptsFailover
@@ -89,3 +87,7 @@ org.apache.qpid.systests.jms_2_0.*
// Exclude 1.0 protocol tests
org.apache.qpid.tests.protocol.v1_0.*
+
+// QPID-7818: Messages published into non existing queue are discarded on 0-10 path
+org.apache.qpid.systest.AnonymousProducerTest#testPublishIntoNonExistingQueue
+org.apache.qpid.systest.AnonymousProducerTest#testSyncPublishIntoNonExistingQueue
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org