QPID-6754: distinguish between null target and null address (depends on PROTON-1043); add some docs; fix client to set address to null, not the whole target.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1713529 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 37 insertions, 10 deletions

diff --git a/qpid/cpp/AMQP_1.0 b/qpid/cpp/AMQP_1.0
index 95f02b8e53..0e811a68f5 100644
--- a/qpid/cpp/AMQP_1.0
+++ b/qpid/cpp/AMQP_1.0
@@ -51,6 +51,9 @@ The name specified in the address supplied when creating a sender or
 receiver is used to set the 'address' field of the target or source
 respectively.
 
+The special address name '<null>' can be used to indicate that the
+address hould be left null.
+
 If the subject is specified for a sender it is used the default
 subject for messages sent without an explicit subject set.
 
@@ -177,6 +180,20 @@ be created on the broker and bound to the exchange. [See section on
 'Topics' below]. The name of the queue will be of the form
 <container-id>_<link-name>.
 
+If the target address of an attaching sender link is null, and the
+dynamic flag is not set, then the broker will route all messages
+received over this sender link using the 'to' field from the
+properties. If that matches the name of a queue, the message will be
+enqueued to that queue. If it doesn't match a queue but matches an
+exchange the message will be routed through that exchange. If it
+doesn't match either a queue or an exchange then it will be
+dropped. This behaviour is known as 'ANONYMOUS-RELAY', a capability
+which the broker advertises in the open frames it sends over every
+connection. To use it the sending process needs to have permission to
+access the ANONYMOUS-RELAY 'pseudo-exchange' (note: there isn't
+actually an exchange of that name, but this is how the capability is
+represented in the ACL permissions model).
+
 Outgoing links may have a filter set on their source. The filters
 currently supported by the broker are 'legacy-amqp-direct-binding',
 'legacy-amqp-topic-binding', 'legacy-amqp-headers-binding',
diff --git a/qpid/cpp/src/qpid/broker/amqp/Connection.cpp b/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
index abe3f8d1f9..2582877bb6 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
@@ -109,6 +109,7 @@ struct ConnectionTickerTask : public qpid::sys::TimerTask
         connection.requestIO();
     }
 };
+const std::string ANONYMOUS_RELAY("ANONYMOUS-RELAY");
 }
 
 Connection::Connection(qpid::sys::OutputControl& o, const std::string& i, BrokerContext& b, bool saslInUse, bool brokerInitiated)
@@ -351,6 +352,15 @@ void Connection::open()
                      << " remote=" << pn_transport_get_remote_idle_timeout(transport));
     }
 
+    pn_data_t* offered_capabilities = pn_connection_offered_capabilities(connection);
+    if (offered_capabilities) {
+        pn_data_put_array(offered_capabilities, false, PN_SYMBOL);
+        pn_data_enter(offered_capabilities);
+        pn_data_put_symbol(offered_capabilities, pn_bytes(ANONYMOUS_RELAY.size(), ANONYMOUS_RELAY.c_str()));
+        pn_data_exit(offered_capabilities);
+        pn_data_rewind(offered_capabilities);
+    }
+
     // QPID-6592: put self-identifying information into the connection
     // properties.  Use keys defined by the 0-10 spec, as AMQP 1.0 has yet to
     // define any.
diff --git a/qpid/cpp/src/qpid/broker/amqp/Session.cpp b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
index e0ca8f58e0..33d3373bd7 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Session.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Session.cpp
@@ -430,13 +430,7 @@ void Session::attach(pn_link_t* link)
         std::string name;
         if (pn_terminus_get_type(target) == PN_UNSPECIFIED) {
             pn_terminus_set_type(pn_link_target(link), PN_UNSPECIFIED);
-            authorise.access("ANONYMOUS-RELAY");
-            boost::shared_ptr<Incoming> r(new AnonymousRelay(connection.getBroker(), connection, *this, link));
-            incoming[link] = r;
-            if (connection.getBroker().isAuthenticating() && !connection.isLink())
-                r->verify(connection.getUserId(), connection.getBroker().getRealm());
-            QPID_LOG(debug, "Incoming link attached for ANONYMOUS-RELAY");
-            return;
+            throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "No target specified!");
         } else if (pn_terminus_get_type(target) == PN_COORDINATOR) {
             QPID_LOG(debug, "Received attach request for incoming link to transaction coordinator on " << this);
             boost::shared_ptr<Incoming> i(new IncomingToCoordinator(link, connection.getBroker(), *this));
@@ -446,6 +440,14 @@ void Session::attach(pn_link_t* link)
             name = generateName(link);
             QPID_LOG(debug, "Received attach request for incoming link to " << name);
             pn_terminus_set_address(pn_link_target(link), qualifyName(name).c_str());
+        } else if (pn_terminus_get_type(target) == PN_TARGET && !pn_terminus_get_address(target)) {
+            authorise.access("ANONYMOUS-RELAY");
+            boost::shared_ptr<Incoming> r(new AnonymousRelay(connection.getBroker(), connection, *this, link));
+            incoming[link] = r;
+            if (connection.getBroker().isAuthenticating() && !connection.isLink())
+                r->verify(connection.getUserId(), connection.getBroker().getRealm());
+            QPID_LOG(debug, "Incoming link attached for ANONYMOUS-RELAY");
+            return;
         } else {
             name  = pn_terminus_get_address(target);
             QPID_LOG(debug, "Received attach request for incoming link to " << name);
diff --git a/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp b/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
index 608d6c2883..513556ba53 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
@@ -612,9 +612,7 @@ void AddressHelper::configure(pn_link_t* link, pn_terminus_t* terminus, CheckMod
         //application expects a name to be generated
         pn_terminus_set_dynamic(terminus, true);
         setNodeProperties(terminus);
-    } else if (name == NULL_ADDRESS) {
-        pn_terminus_set_type(terminus, PN_UNSPECIFIED);
-    } else {
+    } else if (name != NULL_ADDRESS) {
         pn_terminus_set_address(terminus, name.c_str());
         if (createEnabled(mode)) {
             //application expects name of node to be as specified