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/06/16 11:35:51 UTC
qpid-broker-j git commit: QPID-7635: [Java Broker] Refactor
resolution of routing address for AMQP 1.0
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 1a1aed8af -> d79e5b6ac
QPID-7635: [Java Broker] Refactor resolution of routing address for AMQP 1.0
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/d79e5b6a
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/d79e5b6a
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/d79e5b6a
Branch: refs/heads/master
Commit: d79e5b6ac5d771dee5ea63e0fe7cb3eac8123112
Parents: 1a1aed8
Author: Lorenz Quack <lq...@apache.org>
Authored: Fri Jun 16 10:35:23 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Fri Jun 16 10:39:13 2017 +0100
----------------------------------------------------------------------
.../qpid/server/message/ServerMessage.java | 2 -
.../message/internal/InternalMessage.java | 15 +---
.../message/AbstractServerMessageTest.java | 8 +-
.../server/store/TestMessageMetaDataType.java | 8 +-
.../qpid/server/txn/MockServerMessage.java | 6 --
.../protocol/v0_10/MessageTransferMessage.java | 14 +---
.../qpid/server/protocol/v0_8/AMQMessage.java | 13 +---
.../v1_0/AnonymousRelayDestination.java | 2 +-
.../protocol/v1_0/ExchangeDestination.java | 26 ++++++-
.../qpid/server/protocol/v1_0/Message_1_0.java | 77 +++++---------------
.../protocol/v1_0/NodeReceivingDestination.java | 2 +-
.../protocol/v1_0/ReceivingDestination.java | 11 ++-
12 files changed, 64 insertions(+), 120 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-core/src/main/java/org/apache/qpid/server/message/ServerMessage.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/ServerMessage.java b/broker-core/src/main/java/org/apache/qpid/server/message/ServerMessage.java
index 693afd6..d648b63 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/ServerMessage.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/ServerMessage.java
@@ -55,6 +55,4 @@ public interface ServerMessage<T extends StorableMessageMetaData> extends Enqueu
Object getConnectionReference();
boolean isResourceAcceptable(TransactionLogResource resource);
-
- String getRoutingAddress(String destinationAddress, String initialDestinationRoutingAddress);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java b/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
index 5de6127..78e930d 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
@@ -48,7 +48,7 @@ public class InternalMessage extends AbstractServerMessageImpl<InternalMessage,
private static final String NON_AMQP_MESSAGE = "Non-AMQP Message";
private final Object _messageBody;
private InternalMessageHeader _header;
- private String _initialRoutingAddress;
+ private String _initialRoutingAddress = "";
InternalMessage(final StoredMessage<InternalMessageMetaData> handle,
@@ -118,17 +118,6 @@ public class InternalMessage extends AbstractServerMessageImpl<InternalMessage,
return true;
}
- @Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- String initialRoutingAddress = getInitialRoutingAddress();
- if(initialRoutingAddress != null && destinationAddress != null && initialRoutingAddress.startsWith(destinationAddress+"/"))
- {
- initialRoutingAddress = initialRoutingAddress.substring(destinationAddress.length() + 1);
- }
- return initialRoutingAddress;
- }
-
public Object getMessageBody()
{
return _messageBody;
@@ -292,6 +281,6 @@ public class InternalMessage extends AbstractServerMessageImpl<InternalMessage,
public void setInitialRoutingAddress(final String initialRoutingAddress)
{
- _initialRoutingAddress = initialRoutingAddress;
+ _initialRoutingAddress = initialRoutingAddress == null ? "" : initialRoutingAddress;
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-core/src/test/java/org/apache/qpid/server/message/AbstractServerMessageTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/message/AbstractServerMessageTest.java b/broker-core/src/test/java/org/apache/qpid/server/message/AbstractServerMessageTest.java
index c3dd9aa..1b2cb8f 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/message/AbstractServerMessageTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/message/AbstractServerMessageTest.java
@@ -44,7 +44,7 @@ public class AbstractServerMessageTest extends QpidTestCase
@Override
public String getInitialRoutingAddress()
{
- return null;
+ return "";
}
@Override
@@ -76,12 +76,6 @@ public class AbstractServerMessageTest extends QpidTestCase
{
return true;
}
-
- @Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- return null;
- }
}
private TransactionLogResource createQueue(String name)
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
index cd7da55..24fa263 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
@@ -124,12 +124,6 @@ public class TestMessageMetaDataType implements MessageMetaDataType<TestMessageM
}
@Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- return null;
- }
-
- @Override
public long getExpiration()
{
return 0;
@@ -156,7 +150,7 @@ public class TestMessageMetaDataType implements MessageMetaDataType<TestMessageM
@Override
public String getInitialRoutingAddress()
{
- return null;
+ return "";
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java b/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
index 4b14faf..fcb7c92 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
@@ -139,12 +139,6 @@ class MockServerMessage implements ServerMessage
}
@Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- return null;
- }
-
- @Override
public long getArrivalTime()
{
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
index 8af13d8..07860f5 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
@@ -50,7 +50,8 @@ public class MessageTransferMessage extends AbstractServerMessageImpl<MessageTra
public String getInitialRoutingAddress()
{
- return getMetaData().getRoutingKey();
+ final String routingKey = getMetaData().getRoutingKey();
+ return routingKey == null ? "" : routingKey;
}
public AMQMessageHeader getMessageHeader()
@@ -86,17 +87,6 @@ public class MessageTransferMessage extends AbstractServerMessageImpl<MessageTra
return true;
}
- @Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- String initialRoutingAddress = getInitialRoutingAddress();
- if(initialRoutingAddress != null && destinationAddress != null && initialRoutingAddress.startsWith(destinationAddress+"/"))
- {
- initialRoutingAddress = initialRoutingAddress.substring(destinationAddress.length() + 1);
- }
- return initialRoutingAddress;
- }
-
public Header getHeader()
{
return getMetaData().getHeader();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
index d05d312..aaf22a2 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
@@ -69,7 +69,7 @@ public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMet
return routingKey.toString();
}
}
- return null;
+ return "";
}
public AMQMessageHeader getMessageHeader()
@@ -93,17 +93,6 @@ public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMet
return true;
}
- @Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- String initialRoutingAddress = getInitialRoutingAddress();
- if(initialRoutingAddress != null && destinationAddress != null && initialRoutingAddress.startsWith(destinationAddress+"/"))
- {
- initialRoutingAddress = initialRoutingAddress.substring(destinationAddress.length() + 1);
- }
- return initialRoutingAddress;
- }
-
public boolean isImmediate()
{
return getMessagePublishInfo().isImmediate();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/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 bf114df..9d8c819 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
@@ -76,7 +76,7 @@ public class AnonymousRelayDestination implements ReceivingDestination
public Outcome send(final ServerMessage<?> message, final ServerTransaction txn, final SecurityToken securityToken)
{
final ReceivingDestination destination;
- final String routingAddress = message.getRoutingAddress(null, null);
+ final String routingAddress = message.getInitialRoutingAddress();
if (!routingAddress.startsWith("/") && routingAddress.contains("/"))
{
String[] parts = routingAddress.split("/", 2);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/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 29a1679..575dafd 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
@@ -97,7 +97,7 @@ public class ExchangeDestination extends QueueDestination
public Outcome send(final ServerMessage<?> message, final ServerTransaction txn, final SecurityToken securityToken)
{
- final String routingAddress = message.getRoutingAddress(_exchange.getName(), _initialRoutingAddress);
+ final String routingAddress = getRoutingAddress(message);
_exchange.authorisePublish(securityToken, Collections.singletonMap("routingKey", routingAddress));
final InstanceProperties instanceProperties =
@@ -150,6 +150,30 @@ public class ExchangeDestination extends QueueDestination
return _exchange;
}
+ private String getRoutingAddress(final ServerMessage<?> message)
+ {
+ String routingAddress;
+ if (_initialRoutingAddress == null)
+ {
+ return ReceivingDestination.getRoutingAddress(message, _exchange.getName());
+ }
+ else
+ {
+ String initialRoutingAddress = message.getInitialRoutingAddress();
+ if (initialRoutingAddress.startsWith(_exchange.getName() + "/" + _initialRoutingAddress + "/"))
+ {
+ routingAddress = initialRoutingAddress.substring(2
+ + _exchange.getName().length()
+ + _initialRoutingAddress.length());
+ }
+ else
+ {
+ routingAddress = _initialRoutingAddress;
+ }
+ }
+ return routingAddress;
+ }
+
TerminusDurability getDurability()
{
return _durability;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
index f308e78..c29fa09 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
@@ -72,15 +72,31 @@ public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageM
public String getInitialRoutingAddress()
{
- Object routingKey = getMessageHeader().getHeader("routing-key");
- if(routingKey != null)
+ MessageMetaData_1_0.MessageHeader_1_0 messageHeader = getMessageHeader();
+ String routingAddress = null;
+ 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)
{
- return routingKey.toString();
+ routingAddress = (String) messageHeader.getHeader("routing_key");
+ }
+ else if (messageHeader.getSubject() != null)
+ {
+ routingAddress = messageHeader.getSubject();
}
else
{
- return getMessageHeader().getTo();
+ routingAddress = "";
}
+
+ return routingAddress;
}
private MessageMetaData_1_0 getMessageMetaData()
@@ -118,59 +134,6 @@ public class Message_1_0 extends AbstractServerMessageImpl<Message_1_0, MessageM
return getMessageHeader().getNotValidBefore() == 0L || resourceSupportsDeliveryDelay(resource);
}
- @Override
- public String getRoutingAddress(final String destinationAddress, final String initialDestinationRoutingAddress)
- {
- String routingAddress;
- MessageMetaData_1_0.MessageHeader_1_0 messageHeader = getMessageHeader();
- if (initialDestinationRoutingAddress == null)
- {
- final String to = messageHeader.getTo();
- if (to != null && (destinationAddress == null || destinationAddress.trim().equals("")))
- {
- routingAddress = to;
- }
- else if (to != null && to.startsWith(destinationAddress + "/"))
- {
- routingAddress = to.substring(1 + destinationAddress.length());
- }
- else if (to != null && !to.equals(destinationAddress))
- {
- 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 = "";
- }
- }
- else
- {
- if (messageHeader.getTo() != null
- && messageHeader.getTo().startsWith(destinationAddress + "/" + initialDestinationRoutingAddress + "/"))
- {
- final int prefixLength = 2 + destinationAddress.length() + initialDestinationRoutingAddress.length();
- routingAddress = messageHeader.getTo().substring(prefixLength);
- }
- else
- {
- routingAddress = initialDestinationRoutingAddress;
- }
- }
- return routingAddress;
- }
-
private boolean resourceSupportsDeliveryDelay(final TransactionLogResource resource)
{
return resource instanceof Queue && ((Queue<?>)resource).isHoldOnPublishEnabled();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/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 88908c5..ac598d6 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
@@ -80,7 +80,7 @@ public class NodeReceivingDestination implements ReceivingDestination
public Outcome send(final ServerMessage<?> message, final ServerTransaction txn, final SecurityToken securityToken)
{
- final String routingAddress = message.getRoutingAddress(_destination.getName(), null);
+ final String routingAddress = ReceivingDestination.getRoutingAddress(message, _address);
_destination.authorisePublish(securityToken, Collections.singletonMap("routingKey", routingAddress));
final InstanceProperties instanceProperties =
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/d79e5b6a/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 502cc47..debf559 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
@@ -23,7 +23,6 @@ package org.apache.qpid.server.protocol.v1_0;
import org.apache.qpid.server.message.MessageDestination;
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;
@@ -44,4 +43,14 @@ public interface ReceivingDestination extends Destination
String getAddress();
MessageDestination getMessageDestination();
+
+ static String getRoutingAddress(final ServerMessage<?> message, final String destinationName)
+ {
+ String initialRoutingAddress = message.getInitialRoutingAddress();
+ if (destinationName != null && initialRoutingAddress.startsWith(destinationName + "/"))
+ {
+ initialRoutingAddress = initialRoutingAddress.substring(destinationName.length() + 1);
+ }
+ return initialRoutingAddress;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org