You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/08/30 10:06:33 UTC
[1/2] qpid-broker-j git commit: QPID-7602: [Java Broker] Unify
address parsing functionality
Repository: qpid-broker-j
Updated Branches:
refs/heads/master d8c7308a5 -> aefd3e9a2
QPID-7602: [Java Broker] Unify address parsing functionality
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/cf3baaf3
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/cf3baaf3
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/cf3baaf3
Branch: refs/heads/master
Commit: cf3baaf3fabff4a44d81532c822dd9df2780ac92
Parents: d8c7308
Author: Lorenz Quack <lq...@apache.org>
Authored: Tue Aug 29 15:34:21 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Wed Aug 30 11:01:26 2017 +0100
----------------------------------------------------------------------
.../server/exchange/DefaultDestination.java | 32 ++------
.../qpid/server/model/DestinationAddress.java | 79 ++++++++++++++++++++
.../qpid/server/model/NamedAddressSpace.java | 2 +
...stractNonConnectionAcceptingVirtualHost.java | 6 ++
.../server/virtualhost/AbstractVirtualHost.java | 7 +-
.../virtualhost/QueueManagingVirtualHost.java | 2 -
.../v1_0/AnonymousRelayDestination.java | 56 +++-----------
.../protocol/v1_0/NodeReceivingDestination.java | 38 ++++++----
.../qpid/server/protocol/v1_0/Session_1_0.java | 75 +++++--------------
.../MessageConverter_1_0_to_v0_10.java | 33 +++-----
.../PropertyConverter_1_0_to_0_10Test.java | 3 +
.../PropertyConverter_1_0_to_0_8Test.java | 3 +
.../management/amqp/ManagementAddressSpace.java | 6 ++
13 files changed, 167 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java b/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
index c784c75..7e281f8 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
@@ -27,6 +27,7 @@ import org.apache.qpid.server.message.MessageSender;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.DestinationAddress;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.PermissionedObject;
@@ -98,34 +99,11 @@ public class DefaultDestination implements MessageDestination, PermissionedObjec
{
RoutingResult<M> result = new RoutingResult<>(message);
- if (routingAddress != null && !routingAddress.trim().equals(""))
+ DestinationAddress destinationAddress = new DestinationAddress(_virtualHost, routingAddress);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+ if (messageDestination != null)
{
- final MessageDestination dest = _virtualHost.getAttainedMessageDestination(routingAddress);
- if (dest == null)
- {
- routingAddress = _virtualHost.getLocalAddress(routingAddress);
- if (routingAddress.contains("/") && !routingAddress.startsWith("/"))
- {
- String[] parts = routingAddress.split("/", 2);
- Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, parts[0]);
- if (exchange != null)
- {
- result.add(exchange.route(message, parts[1], instanceProperties));
- }
- }
- else if (!routingAddress.contains("/"))
- {
- Exchange<?> exchange = _virtualHost.getAttainedChildFromAddress(Exchange.class, routingAddress);
- if (exchange != null)
- {
- result.add(exchange.route(message, "", instanceProperties));
- }
- }
- }
- else
- {
- result.add(dest.route(message, routingAddress, instanceProperties));
- }
+ result.add(messageDestination.route(message,destinationAddress.getRoutingKey(), instanceProperties));
}
return result;
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/model/DestinationAddress.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/DestinationAddress.java b/broker-core/src/main/java/org/apache/qpid/server/model/DestinationAddress.java
new file mode 100644
index 0000000..2a73ec3
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/DestinationAddress.java
@@ -0,0 +1,79 @@
+/*
+ * 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.server.model;
+
+import org.apache.qpid.server.message.MessageDestination;
+
+public class DestinationAddress
+{
+ private final MessageDestination _messageDestination;
+ private final String _routingKey;
+ private final String _routingAddress;
+
+ public DestinationAddress(NamedAddressSpace addressSpace, String routingAddress)
+ {
+ MessageDestination destination = null;
+ String routingKey = routingAddress;
+ if (routingAddress != null && !routingAddress.trim().equals(""))
+ {
+ String localRoutingAddress = addressSpace.getLocalAddress(routingAddress);
+ if (!localRoutingAddress.contains("/"))
+ {
+ destination = addressSpace.getAttainedMessageDestination(localRoutingAddress);
+ if (destination != null)
+ {
+ routingKey = "";
+ }
+ }
+ else if (!localRoutingAddress.startsWith("/"))
+ {
+ String[] parts = localRoutingAddress.split("/", 2);
+ destination = addressSpace.getAttainedMessageDestination(parts[0]);
+ if (destination instanceof Exchange)
+ {
+ routingKey = parts[1];
+ }
+ else
+ {
+ destination = null;
+ }
+ }
+ }
+ _routingAddress = routingAddress == null ? "" : routingAddress;
+ _messageDestination = destination;
+ _routingKey = routingKey == null ? "" : routingKey;
+ }
+
+ public MessageDestination getMessageDestination()
+ {
+ return _messageDestination;
+ }
+
+ public String getRoutingKey()
+ {
+ return _routingKey;
+ }
+
+ public String getRoutingAddress()
+ {
+ return _routingAddress;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/model/NamedAddressSpace.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/NamedAddressSpace.java b/broker-core/src/main/java/org/apache/qpid/server/model/NamedAddressSpace.java
index 5076446..8a1785d 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/NamedAddressSpace.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/NamedAddressSpace.java
@@ -76,4 +76,6 @@ public interface NamedAddressSpace extends Named
Collection<? extends Connection<?>> getConnections();
List<String> getGlobalAddressDomains();
+
+ String getLocalAddress(String routingAddress);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractNonConnectionAcceptingVirtualHost.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractNonConnectionAcceptingVirtualHost.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractNonConnectionAcceptingVirtualHost.java
index 5be1888..ed251f1 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractNonConnectionAcceptingVirtualHost.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractNonConnectionAcceptingVirtualHost.java
@@ -167,6 +167,12 @@ public abstract class AbstractNonConnectionAcceptingVirtualHost<X extends Abstra
}
@Override
+ public String getLocalAddress(final String routingAddress)
+ {
+ return routingAddress;
+ }
+
+ @Override
public boolean isActive()
{
return false;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index 4db5344..29d2ad9 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -1498,18 +1498,17 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
@Override
public String getLocalAddress(final String routingAddress)
{
- String localAddress = routingAddress;
if(getGlobalAddressDomains() != null)
{
for(String domain : getGlobalAddressDomains())
{
- if(localAddress.length() > routingAddress.length() - domain.length() && routingAddress.startsWith(domain + "/"))
+ if(routingAddress.startsWith(domain + "/"))
{
- localAddress = routingAddress.substring(domain.length());
+ return routingAddress.substring(domain.length() + 1);
}
}
}
- return localAddress;
+ return routingAddress;
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
index 58d897f..a2f6d5e 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueManagingVirtualHost.java
@@ -275,8 +275,6 @@ public interface QueueManagingVirtualHost<X extends QueueManagingVirtualHost<X>>
Queue<?> getAttainedQueue(String name);
- String getLocalAddress(String routingAddress);
-
<T extends ConfiguredObject<?>> T getAttainedChildFromAddress(Class<T> childClass,
String address);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/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
index fd53dc8..fa83516 100644
--- 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
@@ -26,11 +26,9 @@ import java.util.Arrays;
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.DestinationAddress;
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;
@@ -76,53 +74,19 @@ public class AnonymousRelayDestination implements ReceivingDestination
{
final ReceivingDestination destination;
final String routingAddress = message.getTo();
- if (!routingAddress.startsWith("/") && routingAddress.contains("/"))
+ DestinationAddress destinationAddress = new DestinationAddress(_addressSpace, routingAddress);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+ if (messageDestination != null)
{
- String[] parts = routingAddress.split("/", 2);
- MessageDestination exchangeDestination = _addressSpace.getAttainedMessageDestination(parts[0]);
- if (exchangeDestination instanceof Exchange)
- {
- destination = new NodeReceivingDestination(exchangeDestination,
- _target.getDurable(),
- _target.getExpiryPolicy(),
- parts[0],
- _target.getCapabilities(),
- _eventLogger);
- }
- else
- {
- destination = null;
- }
+ destination = new NodeReceivingDestination(destinationAddress,
+ _target.getDurable(),
+ _target.getExpiryPolicy(),
+ _target.getCapabilities(),
+ _eventLogger);
}
else
{
- MessageDestination messageDestination = _addressSpace.getAttainedMessageDestination(routingAddress);
- if(messageDestination == null)
- {
- // TODO - should we do this... if the queue is not being advertised as a destination, shouldn't we
- // respect that?
-
- // Covers the unlikely case where there is no attained destination with the given address, but there is
- // a queue with that address
- MessageSource source = _addressSpace.getAttainedMessageSource(routingAddress);
- if (source instanceof Queue)
- {
- messageDestination = (Queue<?>) source;
- }
- }
- if (messageDestination != null)
- {
- destination = new NodeReceivingDestination(messageDestination,
- _target.getDurable(),
- _target.getExpiryPolicy(),
- routingAddress,
- _target.getCapabilities(),
- _eventLogger);
- }
- else
- {
- destination = null;
- }
+ destination = null;
}
final Outcome outcome;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/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 7bb9044..192a618 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
@@ -32,6 +32,7 @@ import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.RejectType;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.DestinationAddress;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.protocol.v1_0.type.Outcome;
import org.apache.qpid.server.protocol.v1_0.type.Symbol;
@@ -56,23 +57,33 @@ public class NodeReceivingDestination implements ReceivingDestination
private TerminusDurability _durability;
private TerminusExpiryPolicy _expiryPolicy;
private final String _address;
- private String _routingAddress;
+ private final String _routingAddress;
- public NodeReceivingDestination(MessageDestination destination,
- TerminusDurability durable,
- TerminusExpiryPolicy expiryPolicy,
- final String address, final Symbol[] capabilities,
+ public NodeReceivingDestination(final DestinationAddress destinationAddress,
+ final TerminusDurability durable,
+ final TerminusExpiryPolicy expiryPolicy,
+ final Symbol[] capabilities,
final EventLogger eventLogger)
{
- _destination = destination;
+ _destination = destinationAddress.getMessageDestination();
_durability = durable;
_expiryPolicy = expiryPolicy;
- _address = address;
+
_eventLogger = eventLogger;
- _discardUnroutable = destination instanceof Exchange
- && ((capabilities != null && Arrays.asList(capabilities).contains(DISCARD_UNROUTABLE))
- || ((Exchange)destination).getUnroutableMessageBehaviour() == Exchange.UnroutableMessageBehaviour.DISCARD);
+ if (_destination instanceof Exchange)
+ {
+ _discardUnroutable = ((capabilities != null && Arrays.asList(capabilities).contains(DISCARD_UNROUTABLE))
+ || ((Exchange)_destination).getUnroutableMessageBehaviour() == Exchange.UnroutableMessageBehaviour.DISCARD);
+ _routingAddress = destinationAddress.getRoutingKey();
+ _address = _destination.getName();
+ }
+ else
+ {
+ _discardUnroutable = false;
+ _routingAddress = "";
+ _address = destinationAddress.getRoutingAddress();
+ }
}
@Override
@@ -84,7 +95,7 @@ public class NodeReceivingDestination implements ReceivingDestination
@Override
public Outcome send(final ServerMessage<?> message, final ServerTransaction txn, final SecurityToken securityToken)
{
- final String routingAddress = _routingAddress == null ? getRoutingAddress(message) : _routingAddress;
+ final String routingAddress = "".equals(_routingAddress) ? getRoutingAddress(message) : _routingAddress;
_destination.authorisePublish(securityToken, Collections.singletonMap("routingKey", routingAddress));
final InstanceProperties instanceProperties =
@@ -212,9 +223,4 @@ public class NodeReceivingDestination implements ReceivingDestination
capabilities[1] = DELAYED_DELIVERY;
return capabilities;
}
-
- public void setRoutingAddress(final String routingAddress)
- {
- _routingAddress = routingAddress;
- }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/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 5d9eb4e..2c44be0 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
@@ -59,6 +59,7 @@ import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.DestinationAddress;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Queue;
@@ -617,8 +618,14 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
if (Boolean.TRUE.equals(target.getDynamic()))
{
MessageDestination tempDestination = createDynamicDestination(link, target.getDynamicNodeProperties(), target.getCapabilities());
- // TODO: avoid NPE
- target.setAddress(tempDestination.getName());
+ if(tempDestination != null)
+ {
+ target.setAddress(_primaryDomain + tempDestination.getName());
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.INTERNAL_ERROR, "Cannot create dynamic destination");
+ }
}
String addr = target.getAddress();
@@ -626,57 +633,18 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
{
destination = new AnonymousRelayDestination(getAddressSpace(), target, _connection.getEventLogger());
}
- else if (!addr.startsWith("/") && addr.contains("/"))
- {
- String[] parts = addr.split("/", 2);
- Exchange<?> exchange = getExchange(parts[0]);
- if (exchange != null)
- {
-
-
- destination =
- new NodeReceivingDestination(exchange,
- target.getDurable(),
- target.getExpiryPolicy(),
- parts[0],
- target.getCapabilities(),
- _connection.getEventLogger());
- ((NodeReceivingDestination)destination).setRoutingAddress(parts[1]);
-
- }
- else
- {
- destination = null;
- }
- }
else
{
- MessageDestination messageDestination =
- getAddressSpace().getAttainedMessageDestination(addr);
-
- if(messageDestination == null)
- {
- // TODO - should we do this... if the queue is not being advertised as a destination, shouldn't we
- // respect that?
-
- // Covers the unlikely case where there is no attained destination with the given address, but there is
- // a queue with that address
- MessageSource source = getAddressSpace().getAttainedMessageSource(addr);
- if (source instanceof Queue)
- {
- messageDestination = (Queue<?>) source;
- }
- }
+ DestinationAddress destinationAddress = new DestinationAddress(getAddressSpace(), addr);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
if (messageDestination != null)
{
- destination =
- new NodeReceivingDestination(messageDestination,
- target.getDurable(),
- target.getExpiryPolicy(),
- addr,
- target.getCapabilities(),
- _connection.getEventLogger());
+ destination = new NodeReceivingDestination(destinationAddress,
+ target.getDurable(),
+ target.getExpiryPolicy(),
+ target.getCapabilities(),
+ _connection.getEventLogger());
}
else
{
@@ -825,11 +793,11 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
private MessageDestination createDynamicDestination(final Link_1_0<?, ?> link,
Map properties,
- final Symbol[] capabilities)
+ final Symbol[] capabilities) throws AmqpErrorException
{
final Set<Symbol> capabilitySet = capabilities == null ? Collections.emptySet() : Sets.newHashSet(capabilities);
boolean isTopic = capabilitySet.contains(Symbol.valueOf("temporary-topic")) || capabilitySet.contains(Symbol.valueOf("topic"));
- final String destName = _primaryDomain + (isTopic ? "TempTopic" : "TempQueue") + UUID.randomUUID().toString();
+ final String destName = (isTopic ? "TempTopic" : "TempQueue") + UUID.randomUUID().toString();
try
{
Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(link, properties, destName);
@@ -846,12 +814,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
}
catch (AccessControlException e)
{
- Error error = new Error();
- error.setCondition(AmqpError.UNAUTHORIZED_ACCESS);
- error.setDescription(e.getMessage());
-
- _connection.close(error);
- return null;
+ throw new AmqpErrorException(AmqpError.UNAUTHORIZED_ACCESS, e.getMessage());
}
catch (AbstractConfiguredObject.DuplicateNameException e)
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java b/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
index c10a232..f80a913 100644
--- a/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
+++ b/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
@@ -42,6 +42,7 @@ import java.util.UUID;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.model.DestinationAddress;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Queue;
@@ -294,30 +295,14 @@ public class MessageConverter_1_0_to_v0_10 implements MessageConverter<Message_1
private ReplyTo getReplyTo(final NamedAddressSpace addressSpace, final String origReplyTo)
{
- String exchange, routingKey;
- if (origReplyTo.startsWith("/"))
- {
- exchange = "";
- routingKey = origReplyTo;
- }
- else if (origReplyTo.contains("/"))
- {
- String[] parts = origReplyTo.split("/", 2);
- exchange = parts[0];
- routingKey = parts[1];
- }
- else if (addressSpace.getAttainedMessageDestination(origReplyTo) instanceof Exchange)
- {
- exchange = origReplyTo;
- routingKey = "";
- }
- else
- {
- exchange = "";
- routingKey = origReplyTo;
- }
- return new ReplyTo(ensureStr8("reply-to[\"exchange\"]", exchange),
- ensureStr8("reply-to[\"routing-key\"]", routingKey));
+ DestinationAddress destinationAddress = new DestinationAddress(addressSpace, origReplyTo);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+ return new ReplyTo(ensureStr8("reply-to[\"exchange\"]",
+ messageDestination instanceof Exchange ? messageDestination.getName() : ""),
+ ensureStr8("reply-to[\"routing-key\"]",
+ messageDestination instanceof Queue
+ ? messageDestination.getName()
+ : destinationAddress.getRoutingKey()));
}
private void setExchangeAndRoutingKeyOnDeliveryProperties(final DeliveryProperties deliveryProps,
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/PropertyConverter_1_0_to_0_10Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/PropertyConverter_1_0_to_0_10Test.java b/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/PropertyConverter_1_0_to_0_10Test.java
index a2515c2..ab60a13 100644
--- a/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/PropertyConverter_1_0_to_0_10Test.java
+++ b/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/PropertyConverter_1_0_to_0_10Test.java
@@ -20,6 +20,8 @@
package org.apache.qpid.server.protocol.converter.v0_10_v1_0;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.AdditionalAnswers.returnsFirstArg;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -70,6 +72,7 @@ public class PropertyConverter_1_0_to_0_10Test extends QpidTestCase
{
super.setUp();
_namedAddressSpace = mock(NamedAddressSpace.class);
+ when(_namedAddressSpace.getLocalAddress(anyString())).then(returnsFirstArg());
_messageConverter = new MessageConverter_1_0_to_v0_10();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/PropertyConverter_1_0_to_0_8Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/PropertyConverter_1_0_to_0_8Test.java b/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/PropertyConverter_1_0_to_0_8Test.java
index efd35fd..0bb47a7 100644
--- a/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/PropertyConverter_1_0_to_0_8Test.java
+++ b/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/test/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/PropertyConverter_1_0_to_0_8Test.java
@@ -20,6 +20,8 @@
package org.apache.qpid.server.protocol.converter.v0_8_v1_0;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.AdditionalAnswers.returnsFirstArg;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -65,6 +67,7 @@ public class PropertyConverter_1_0_to_0_8Test extends QpidTestCase
{
super.setUp();
_namedAddressSpace = mock(NamedAddressSpace.class);
+ when(_namedAddressSpace.getLocalAddress(anyString())).then(returnsFirstArg());
_messageConverter = new MessageConverter_1_0_to_v0_8();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/cf3baaf3/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
----------------------------------------------------------------------
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
index a98c8e6..ea2975a 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
@@ -366,6 +366,12 @@ public class ManagementAddressSpace implements NamedAddressSpace
return Collections.emptyList();
}
+ @Override
+ public String getLocalAddress(final String routingAddress)
+ {
+ return routingAddress;
+ }
+
private class DefaultDestination implements MessageDestination
{
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/2] qpid-broker-j git commit: QPID-7602: [Java Broker] Convert
reply-to from internal format into AMQP 0-x
Posted by lq...@apache.org.
QPID-7602: [Java Broker] Convert reply-to from internal format into AMQP 0-x
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/aefd3e9a
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/aefd3e9a
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/aefd3e9a
Branch: refs/heads/master
Commit: aefd3e9a231cd0015d5d65e63fd7f2f113eba2fb
Parents: cf3baaf
Author: Lorenz Quack <lq...@apache.org>
Authored: Tue Aug 29 15:35:03 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Wed Aug 30 11:01:27 2017 +0100
----------------------------------------------------------------------
.../MessageConverter_Internal_to_v0_10.java | 44 ++++++++---
...PropertyConverter_Internal_to_v0_10Test.java | 77 ++++++++++++++++++
.../v0_8/MessageConverter_Internal_to_v0_8.java | 58 ++++++++++++--
.../PropertyConverter_Internal_to_v0_8Test.java | 83 ++++++++++++++++++++
.../v0_8_v1_0/MessageConverter_1_0_to_v0_8.java | 48 +++++++++--
5 files changed, 287 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/aefd3e9a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
index 700a0bc..4c74784 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
@@ -27,21 +27,26 @@ import java.util.Collections;
import java.util.UUID;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.internal.InternalMessageHeader;
import org.apache.qpid.server.message.mimecontentconverter.MimeContentConverterRegistry;
import org.apache.qpid.server.message.mimecontentconverter.ObjectToMimeContentConverter;
+import org.apache.qpid.server.model.DestinationAddress;
+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.plugin.MessageConverter;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.protocol.converter.MessageConversionException;
-import org.apache.qpid.server.protocol.v0_10.transport.EncoderUtils;
-import org.apache.qpid.server.protocol.v0_10.transport.MessageDeliveryMode;
-import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties;
+import org.apache.qpid.server.protocol.v0_10.transport.EncoderUtils;
import org.apache.qpid.server.protocol.v0_10.transport.Header;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageDeliveryMode;
import org.apache.qpid.server.protocol.v0_10.transport.MessageDeliveryPriority;
import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties;
+import org.apache.qpid.server.protocol.v0_10.transport.ReplyTo;
+import org.apache.qpid.server.store.StoredMessage;
@PluggableService
public class MessageConverter_Internal_to_v0_10 implements MessageConverter<InternalMessage, MessageTransferMessage>
@@ -64,7 +69,7 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
@Override
public MessageTransferMessage convert(InternalMessage serverMsg, NamedAddressSpace addressSpace)
{
- return new MessageTransferMessage(convertToStoredMessage(serverMsg), null);
+ return new MessageTransferMessage(convertToStoredMessage(serverMsg, addressSpace), null);
}
@Override
@@ -73,7 +78,8 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
}
- private StoredMessage<MessageMetaData_0_10> convertToStoredMessage(final InternalMessage serverMsg)
+ private StoredMessage<MessageMetaData_0_10> convertToStoredMessage(final InternalMessage serverMsg,
+ final NamedAddressSpace addressSpace)
{
Object messageBody = serverMsg.getMessageBody();
ObjectToMimeContentConverter converter = MimeContentConverterRegistry.getBestFitObjectToMimeContentConverter(messageBody);
@@ -82,9 +88,8 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
mimeType = improveMimeType(serverMsg, mimeType);
- final MessageMetaData_0_10 messageMetaData_0_10 = convertMetaData(serverMsg,
- mimeType,
- messageContent.length);
+ final MessageMetaData_0_10 messageMetaData_0_10 =
+ convertMetaData(serverMsg, addressSpace, mimeType, messageContent.length);
final int metadataSize = messageMetaData_0_10.getStorableSize();
return new StoredMessage<MessageMetaData_0_10>()
@@ -162,7 +167,10 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
return mimeType;
}
- private MessageMetaData_0_10 convertMetaData(InternalMessage serverMsg, final String bodyMimeType, final int size)
+ private MessageMetaData_0_10 convertMetaData(final InternalMessage serverMsg,
+ final NamedAddressSpace addressSpace,
+ final String bodyMimeType,
+ final int size)
{
DeliveryProperties deliveryProps = new DeliveryProperties();
MessageProperties messageProps = new MessageProperties();
@@ -218,10 +226,28 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
messageProps.setUserId(bytes);
}
}
+
+ final String origReplyTo = messageHeader.getReplyTo();
+ if (origReplyTo != null && !origReplyTo.equals(""))
+ {
+ messageProps.setReplyTo(getReplyTo(addressSpace, origReplyTo));
+ }
+
Header header = new Header(deliveryProps, messageProps, null);
return new MessageMetaData_0_10(header, size, serverMsg.getArrivalTime());
}
+ private ReplyTo getReplyTo(final NamedAddressSpace addressSpace, final String origReplyTo)
+ {
+ DestinationAddress destinationAddress = new DestinationAddress(addressSpace, origReplyTo);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+ final String exchange = ensureStr8("reply-to[\"exchange\"]", messageDestination instanceof Exchange
+ ? messageDestination.getName() : "");
+ final String routingKey = ensureStr8("reply-to[\"routing-key\"]", messageDestination instanceof Queue
+ ? messageDestination.getName() : destinationAddress.getRoutingKey());
+ return new ReplyTo(exchange, routingKey);
+ }
+
private void validateValue(final Object value, final String path)
{
if (!EncoderUtils.isEncodable(value))
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/aefd3e9a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/PropertyConverter_Internal_to_v0_10Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/PropertyConverter_Internal_to_v0_10Test.java b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/PropertyConverter_Internal_to_v0_10Test.java
index d56ca24..69962fc 100644
--- a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/PropertyConverter_Internal_to_v0_10Test.java
+++ b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/PropertyConverter_Internal_to_v0_10Test.java
@@ -22,7 +22,9 @@
package org.apache.qpid.server.protocol.v0_10;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.AdditionalAnswers.returnsFirstArg;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -43,9 +45,12 @@ import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.internal.InternalMessageHeader;
import org.apache.qpid.server.message.internal.InternalMessageMetaData;
import org.apache.qpid.server.message.internal.InternalMessageMetaDataType;
+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.converter.MessageConversionException;
import org.apache.qpid.server.protocol.v0_10.transport.MessageDeliveryMode;
+import org.apache.qpid.server.protocol.v0_10.transport.ReplyTo;
import org.apache.qpid.server.protocol.v0_8.AMQShortString;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -60,6 +65,7 @@ public class PropertyConverter_Internal_to_v0_10Test extends QpidTestCase
{
super.setUp();
_addressSpace = mock(NamedAddressSpace.class);
+ when(_addressSpace.getLocalAddress(anyString())).then(returnsFirstArg());
_messageConverter = new MessageConverter_Internal_to_v0_10();
}
@@ -354,6 +360,77 @@ public class PropertyConverter_Internal_to_v0_10Test extends QpidTestCase
convertedMessage.getHeader().getMessageProperties().getContentLength());
}
+ public void testReplyToConversionWhenQueueIsSpecified() throws IOException
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ final String replyTo = "myTestQueue";
+ final Queue queue = mock(Queue.class);
+ when(queue.getName()).thenReturn(replyTo);
+ when(_addressSpace.getAttainedMessageDestination(replyTo)).thenReturn(queue);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ MessageTransferMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ final ReplyTo convertedReplyTo =
+ convertedMessage.getHeader().getMessageProperties().getReplyTo();
+ assertEquals("Unexpected exchange", "", convertedReplyTo.getExchange());
+ assertEquals("Unexpected routing key", replyTo, convertedReplyTo.getRoutingKey());
+ }
+
+ public void testReplyToConversionWhenExchangeIsSpecified() throws IOException
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ final String replyTo = "myTestExchange";
+ final Exchange exchange = mock(Exchange.class);
+ when(exchange.getName()).thenReturn(replyTo);
+ when(_addressSpace.getAttainedMessageDestination(replyTo)).thenReturn(exchange);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ MessageTransferMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ final ReplyTo convertedReplyTo =
+ convertedMessage.getHeader().getMessageProperties().getReplyTo();
+ assertEquals("Unexpected exchange", replyTo, convertedReplyTo.getExchange());
+ assertEquals("Unexpected routing key", "", convertedReplyTo.getRoutingKey());
+ }
+
+ public void testReplyToConversionWhenExchangeAndRoutingKeyAreSpecified() throws IOException
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ final String exchangeName = "testExchnageName";
+ final String routingKey = "testRoutingKey";
+ final String replyTo = String.format("%s/%s", exchangeName, routingKey);
+ final Exchange exchange = mock(Exchange.class);
+ when(exchange.getName()).thenReturn(exchangeName);
+ when(_addressSpace.getAttainedMessageDestination(exchangeName)).thenReturn(exchange);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ MessageTransferMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ final ReplyTo convertedReplyTo =
+ convertedMessage.getHeader().getMessageProperties().getReplyTo();
+ assertEquals("Unexpected exchange", exchangeName, convertedReplyTo.getExchange());
+ assertEquals("Unexpected routing key", routingKey, convertedReplyTo.getRoutingKey());
+ }
+
+ public void testReplyToConversionWhenReplyToCannotBeResolved() throws IOException
+ {
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ final String replyTo = "direct://amq.direct//test?routingkey='test'";
+ when(header.getReplyTo()).thenReturn(replyTo);
+ InternalMessage originalMessage = createTestMessage(header);
+
+ MessageTransferMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ final ReplyTo convertedReplyTo =
+ convertedMessage.getHeader().getMessageProperties().getReplyTo();
+ assertEquals("Unexpected exchange", "", convertedReplyTo.getExchange());
+ assertEquals("Unexpected routing key", replyTo, convertedReplyTo.getRoutingKey());
+ }
+
private InternalMessage createTestMessage(final AMQMessageHeader header) throws IOException
{
return createTestMessage(header, null, false, System.currentTimeMillis());
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/aefd3e9a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
index a8ff769..1678a8d 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
@@ -26,10 +26,15 @@ import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.exchange.ExchangeDefaults;
+import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.mimecontentconverter.MimeContentConverterRegistry;
import org.apache.qpid.server.message.mimecontentconverter.ObjectToMimeContentConverter;
+import org.apache.qpid.server.model.DestinationAddress;
+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.plugin.MessageConverter;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.protocol.converter.MessageConversionException;
@@ -58,7 +63,7 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
@Override
public AMQMessage convert(InternalMessage serverMsg, NamedAddressSpace addressSpace)
{
- return new AMQMessage(convertToStoredMessage(serverMsg), null);
+ return new AMQMessage(convertToStoredMessage(serverMsg, addressSpace), null);
}
@Override
@@ -67,7 +72,8 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
}
- private StoredMessage<MessageMetaData> convertToStoredMessage(final InternalMessage serverMsg)
+ private StoredMessage<MessageMetaData> convertToStoredMessage(final InternalMessage serverMsg,
+ final NamedAddressSpace addressSpace)
{
Object messageBody = serverMsg.getMessageBody();
ObjectToMimeContentConverter converter = MimeContentConverterRegistry.getBestFitObjectToMimeContentConverter(messageBody);
@@ -85,9 +91,8 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
mimeType = improveMimeType(serverMsg, mimeType);
- final MessageMetaData messageMetaData_0_8 = convertMetaData(serverMsg,
- mimeType,
- messageContent.length);
+ final MessageMetaData messageMetaData_0_8 =
+ convertMetaData(serverMsg, addressSpace, mimeType, messageContent.length);
final int metadataSize = messageMetaData_0_8.getStorableSize();
return new StoredMessage<MessageMetaData>()
@@ -164,7 +169,10 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
return mimeType;
}
- private MessageMetaData convertMetaData(final InternalMessage serverMsg, final String bodyMimeType, final int size)
+ private MessageMetaData convertMetaData(final InternalMessage serverMsg,
+ final NamedAddressSpace addressSpace,
+ final String bodyMimeType,
+ final int size)
{
MessagePublishInfo publishInfo = new MessagePublishInfo(AMQShortString.EMPTY_STRING,
@@ -181,7 +189,7 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
props.setExpiration(serverMsg.getExpiration());
props.setMessageId(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getMessageId()));
props.setPriority(serverMsg.getMessageHeader().getPriority());
- props.setReplyTo(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getReplyTo()));
+ props.setReplyTo(convertToShortStringForProperty("reply-to", getReplyTo(serverMsg, addressSpace)));
props.setTimestamp(serverMsg.getMessageHeader().getTimestamp());
props.setUserId(convertToOptionalAMQPShortString(serverMsg.getMessageHeader().getUserId()));
@@ -209,6 +217,42 @@ public class MessageConverter_Internal_to_v0_8 implements MessageConverter<Inter
return new MessageMetaData(publishInfo, chb, serverMsg.getArrivalTime());
}
+ private String getReplyTo(final InternalMessage serverMsg, final NamedAddressSpace addressSpace)
+ {
+ String replyTo = serverMsg.getMessageHeader().getReplyTo();
+
+ if (replyTo != null)
+ {
+ DestinationAddress destinationAddress = new DestinationAddress(addressSpace, replyTo);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+
+ final String replyToBindingUrl;
+ if (messageDestination instanceof Exchange)
+ {
+ Exchange<?> exchange = (Exchange<?>) messageDestination;
+ replyToBindingUrl = String.format("%s://%s//?routingkey='%s'",
+ exchange.getType(),
+ exchange.getName(),
+ destinationAddress.getRoutingKey());
+ }
+ else if (messageDestination instanceof Queue)
+ {
+ replyToBindingUrl = String.format("%s:////%s",
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
+ messageDestination.getName());
+ }
+ else
+ {
+ replyToBindingUrl = String.format("%s:////?routingkey='%s'",
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
+ destinationAddress.getRoutingKey());
+ }
+
+ return replyToBindingUrl;
+ }
+ return null;
+ }
+
private AMQShortString convertToOptionalAMQPShortString(final String stringValue)
{
AMQShortString result;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/aefd3e9a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
index b279bf6..0d8baee 100644
--- a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
+++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/PropertyConverter_Internal_to_v0_8Test.java
@@ -21,7 +21,9 @@
*/
package org.apache.qpid.server.protocol.v0_8;
+import static org.mockito.AdditionalAnswers.returnsFirstArg;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -30,12 +32,15 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.internal.InternalMessageHeader;
import org.apache.qpid.server.message.internal.InternalMessageMetaData;
import org.apache.qpid.server.message.internal.InternalMessageMetaDataType;
+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.converter.MessageConversionException;
import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
import org.apache.qpid.server.store.StoredMessage;
@@ -52,6 +57,7 @@ public class PropertyConverter_Internal_to_v0_8Test extends QpidTestCase
super.setUp();
_messageConverter = new MessageConverter_Internal_to_v0_8();
_addressSpace = mock(NamedAddressSpace.class);
+ when(_addressSpace.getLocalAddress(anyString())).then(returnsFirstArg());
}
public void testDurableTrueConversion()
@@ -273,6 +279,83 @@ public class PropertyConverter_Internal_to_v0_8Test extends QpidTestCase
}
}
+ public void testReplyToConversionWhenQueueIsSpecified()
+ {
+ final String replyTo = "testQueue";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ Queue queue = mock(Queue.class);
+ when(queue.getName()).thenReturn(replyTo);
+ when(_addressSpace.getAttainedMessageDestination(replyTo)).thenReturn(queue);
+
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected reply-to",
+ "direct:////" + replyTo,
+ convertedMessage.getContentHeaderBody().getProperties().getReplyToAsString());
+ }
+
+ public void testReplyToConversionWhenExchangeIsSpecified()
+ {
+ final String replyTo = "testExchange";
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ Exchange exchange = mock(Exchange.class);
+ when(exchange.getName()).thenReturn(replyTo);
+ when(exchange.getType()).thenReturn(ExchangeDefaults.FANOUT_EXCHANGE_CLASS);
+
+ when(_addressSpace.getAttainedMessageDestination(replyTo)).thenReturn(exchange);
+
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected reply-to",
+ "fanout://" + replyTo + "//?routingkey=''",
+ convertedMessage.getContentHeaderBody().getProperties().getReplyToAsString());
+ }
+
+ public void testReplyToConversionWhenExchangeAndRoutingKeyAreSpecified()
+ {
+ final String exchangeName = "testExchange";
+ final String routingKey = "testKey";
+ final String replyTo = String.format("%s/%s", exchangeName, routingKey);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getReplyTo()).thenReturn(replyTo);
+ Exchange exchange = mock(Exchange.class);
+ when(exchange.getName()).thenReturn(exchangeName);
+ when(exchange.getType()).thenReturn(ExchangeDefaults.TOPIC_EXCHANGE_CLASS);
+
+ when(_addressSpace.getAttainedMessageDestination(exchangeName)).thenReturn(exchange);
+
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected reply-to",
+ "topic://" + exchangeName + "//?routingkey='" + routingKey + "'",
+ convertedMessage.getContentHeaderBody().getProperties().getReplyToAsString());
+ }
+
+ public void testReplyToConversionWhenNonExistingExchangeAndRoutingKeyAreSpecified()
+ {
+ final String exchangeName = "testExchange";
+ final String routingKey = "testKey";
+ final String replyTo = String.format("%s/%s", exchangeName, routingKey);
+ final AMQMessageHeader header = mock(AMQMessageHeader.class);
+ when(header.getReplyTo()).thenReturn(replyTo);
+
+ InternalMessage originalMessage = createTestMessage(header);
+
+ AMQMessage convertedMessage = _messageConverter.convert(originalMessage, _addressSpace);
+
+ assertEquals("Unexpected reply-to",
+ "direct:////?routingkey='" + replyTo + "'",
+ convertedMessage.getContentHeaderBody().getProperties().getReplyToAsString());
+ }
+
private InternalMessage createTestMessage(final AMQMessageHeader header)
{
return createTestMessage(header, null, false);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/aefd3e9a/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java b/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
index 73e6b65..b309a56 100644
--- a/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
+++ b/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
@@ -39,7 +39,10 @@ import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.model.DestinationAddress;
+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.plugin.MessageConverter;
@@ -343,16 +346,47 @@ public class MessageConverter_1_0_to_v0_8 implements MessageConverter<Message_1_
private AMQShortString getReplyTo(final Message_1_0 serverMsg, final NamedAddressSpace addressSpace)
{
- // TODO : QPID-7602 - we probably need to look up the replyTo object and construct the correct BURL based on that
final String replyTo = serverMsg.getMessageHeader().getReplyTo();
- try
- {
- return AMQShortString.valueOf(replyTo);
- }
- catch (IllegalArgumentException e)
+
+ if (replyTo != null)
{
- throw new MessageConversionException("Could not convert message from 1.0 to 0-8 because conversion of 'reply-to' failed.", e);
+ DestinationAddress destinationAddress = new DestinationAddress(addressSpace, replyTo);
+ MessageDestination messageDestination = destinationAddress.getMessageDestination();
+
+ final String replyToBindingUrl;
+ if (messageDestination instanceof Exchange)
+ {
+ Exchange<?> exchange = (Exchange<?>) messageDestination;
+ replyToBindingUrl = String.format("%s://%s//?routingkey='%s'",
+ exchange.getType(),
+ exchange.getName(),
+ destinationAddress.getRoutingKey());
+ }
+ else if (messageDestination instanceof Queue)
+ {
+ replyToBindingUrl = String.format("%s:////%s",
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
+ messageDestination.getName());
+ }
+ else
+ {
+ replyToBindingUrl = String.format("%s:////?routingkey='%s'",
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
+ destinationAddress.getRoutingKey());
+ }
+
+ try
+ {
+ return AMQShortString.valueOf(replyToBindingUrl);
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new MessageConversionException(
+ "Could not convert message from 1.0 to 0-8 because conversion of 'reply-to' failed.",
+ e);
+ }
}
+ return null;
}
private AMQShortString getCorrelationIdAsShortString(final Message_1_0 serverMsg)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org