QPID-664: Added getReceiver()/getSender() methods to session in new api.

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

14 files changed, 116 insertions, 3 deletions

diff --git a/qpid/cpp/include/qpid/messaging/Receiver.h b/qpid/cpp/include/qpid/messaging/Receiver.h
index 659a583547..2dc62b1598 100644
--- a/qpid/cpp/include/qpid/messaging/Receiver.h
+++ b/qpid/cpp/include/qpid/messaging/Receiver.h
@@ -110,6 +110,12 @@ class Receiver : public qpid::client::Handle<ReceiverImpl>
      * Cancels this receiver.
      */
     QPID_CLIENT_EXTERN void cancel();
+
+    /**
+     * Returns the name of this receiver.
+     */
+    QPID_CLIENT_EXTERN const std::string& getName() const;
+
   private:
   friend class qpid::client::PrivateImplRef<Receiver>;
 };
diff --git a/qpid/cpp/include/qpid/messaging/Sender.h b/qpid/cpp/include/qpid/messaging/Sender.h
index 9b83a04d60..29af3810f4 100644
--- a/qpid/cpp/include/qpid/messaging/Sender.h
+++ b/qpid/cpp/include/qpid/messaging/Sender.h
@@ -24,6 +24,7 @@
 #include "qpid/client/ClientImportExport.h"
 #include "qpid/client/Handle.h"
 #include "qpid/sys/IntegerTypes.h"
+#include <string>
 
 namespace qpid {
 namespace client {
@@ -68,6 +69,10 @@ class Sender : public qpid::client::Handle<SenderImpl>
      */
     QPID_CLIENT_EXTERN uint32_t pending();
 
+    /**
+     * Returns the name of this sender.
+     */
+    QPID_CLIENT_EXTERN const std::string& getName() const;
   private:
   friend class qpid::client::PrivateImplRef<Sender>;
 };
diff --git a/qpid/cpp/include/qpid/messaging/Session.h b/qpid/cpp/include/qpid/messaging/Session.h
index 4b23d355ec..edb6fa6331 100644
--- a/qpid/cpp/include/qpid/messaging/Session.h
+++ b/qpid/cpp/include/qpid/messaging/Session.h
@@ -21,6 +21,7 @@
  * under the License.
  *
  */
+#include "qpid/Exception.h"
 #include "qpid/client/ClientImportExport.h"
 #include "qpid/client/Handle.h"
 #include "qpid/sys/Time.h"
@@ -43,6 +44,11 @@ class Receiver;
 class SessionImpl;
 class Subscription;
 
+struct KeyError : qpid::Exception
+{
+    QPID_CLIENT_EXTERN KeyError(const std::string&);
+};
+
 /**
  * A session represents a distinct 'conversation' which can involve
  * sending and receiving messages to and from different addresses.
@@ -103,11 +109,30 @@ class Session : public qpid::client::Handle<SessionImpl>
      */
     QPID_CLIENT_EXTERN Receiver nextReceiver(qpid::sys::Duration timeout=qpid::sys::TIME_INFINITE);
     
-
+    /**
+     * Create a new sender through which messages can be sent to the
+     * specified address.
+     */
     QPID_CLIENT_EXTERN Sender createSender(const Address& address);
     QPID_CLIENT_EXTERN Sender createSender(const std::string& address);
+
+    /**
+     * Create a new receiver through which messages can be received
+     * from the specified address.
+     */
     QPID_CLIENT_EXTERN Receiver createReceiver(const Address& address);
     QPID_CLIENT_EXTERN Receiver createReceiver(const std::string& address);
+
+    /**
+     * Returns the sender with the specified name or throws KeyError
+     * if there is none for that name.
+     */
+    QPID_CLIENT_EXTERN Sender getSender(const std::string& name) const;
+    /**
+     * Returns the receiver with the specified name or throws KeyError
+     * if there is none for that name.
+     */
+    QPID_CLIENT_EXTERN Receiver getReceiver(const std::string& name) const;
   private:
   friend class qpid::client::PrivateImplRef<Session>;
 };
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
index 9d168725e6..ef16b26b91 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
@@ -129,4 +129,9 @@ void SenderImpl::cancelImpl()
     parent.senderCancelled(name);
 }
 
+const std::string& SenderImpl::getName() const
+{
+    return name;
+}
+
 }}} // namespace qpid::client::amqp0_10
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
index 60b196b21b..8140b3479a 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
@@ -54,6 +54,7 @@ class SenderImpl : public qpid::messaging::SenderImpl
     uint32_t getCapacity();
     uint32_t pending();
     void init(qpid::client::AsyncSession, AddressResolution&);
+    const std::string& getName() const;
 
   private:
     SessionImpl& parent;
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
index 4d40bd6c50..3a8992d503 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
@@ -39,6 +39,7 @@
 #include <boost/intrusive_ptr.hpp>
 
 using qpid::messaging::Filter;
+using qpid::messaging::KeyError;
 using qpid::messaging::MessageImplAccess;
 using qpid::messaging::Sender;
 using qpid::messaging::Receiver;
@@ -176,6 +177,28 @@ Sender SessionImpl::createSenderImpl(const qpid::messaging::Address& address)
     return sender;
 }
 
+Sender SessionImpl::getSender(const std::string& name) const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    Senders::const_iterator i = senders.find(name);
+    if (i == senders.end()) {
+        throw KeyError(name);
+    } else {
+        return i->second;
+    }    
+}
+
+Receiver SessionImpl::getReceiver(const std::string& name) const
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    Receivers::const_iterator i = receivers.find(name);
+    if (i == receivers.end()) {
+        throw KeyError(name);
+    } else {
+        return i->second;
+    }
+}
+
 SessionImpl& SessionImpl::convert(qpid::messaging::Session& s)
 {
     boost::intrusive_ptr<SessionImpl> impl = getImplPtr<qpid::messaging::Session, SessionImpl>(s);
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
index f3018b9685..872f66801d 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
@@ -65,6 +65,9 @@ class SessionImpl : public qpid::messaging::SessionImpl
     qpid::messaging::Sender createSender(const qpid::messaging::Address& address);
     qpid::messaging::Receiver createReceiver(const qpid::messaging::Address& address);
 
+    qpid::messaging::Sender getSender(const std::string& name) const;
+    qpid::messaging::Receiver getReceiver(const std::string& name) const;
+
     bool nextReceiver(qpid::messaging::Receiver& receiver, qpid::sys::Duration timeout);
     qpid::messaging::Receiver nextReceiver(qpid::sys::Duration timeout);
 
@@ -99,7 +102,7 @@ class SessionImpl : public qpid::messaging::SessionImpl
     typedef std::map<std::string, qpid::messaging::Receiver> Receivers;
     typedef std::map<std::string, qpid::messaging::Sender> Senders;
 
-    qpid::sys::Mutex lock;
+    mutable qpid::sys::Mutex lock;
     ConnectionImpl& connection;
     qpid::client::Session session;
     AddressResolution resolver;
diff --git a/qpid/cpp/src/qpid/messaging/Receiver.cpp b/qpid/cpp/src/qpid/messaging/Receiver.cpp
index ac5d93319f..17bad6baed 100644
--- a/qpid/cpp/src/qpid/messaging/Receiver.cpp
+++ b/qpid/cpp/src/qpid/messaging/Receiver.cpp
@@ -47,5 +47,5 @@ uint32_t Receiver::getCapacity() { return impl->getCapacity(); }
 uint32_t Receiver::available() { return impl->available(); }
 uint32_t Receiver::pendingAck() { return impl->pendingAck(); }
 void Receiver::cancel() { impl->cancel(); }
-
+const std::string& Receiver::getName() const { return impl->getName(); }
 }} // namespace qpid::messaging
diff --git a/qpid/cpp/src/qpid/messaging/ReceiverImpl.h b/qpid/cpp/src/qpid/messaging/ReceiverImpl.h
index 5a057ba34e..3d60437aba 100644
--- a/qpid/cpp/src/qpid/messaging/ReceiverImpl.h
+++ b/qpid/cpp/src/qpid/messaging/ReceiverImpl.h
@@ -46,6 +46,7 @@ class ReceiverImpl : public virtual qpid::RefCounted
     virtual uint32_t available() = 0;
     virtual uint32_t pendingAck() = 0;
     virtual void cancel() = 0;
+    virtual const std::string& getName() const = 0;
 };
 }} // namespace qpid::messaging
 
diff --git a/qpid/cpp/src/qpid/messaging/Sender.cpp b/qpid/cpp/src/qpid/messaging/Sender.cpp
index 62b2944701..6087af3514 100644
--- a/qpid/cpp/src/qpid/messaging/Sender.cpp
+++ b/qpid/cpp/src/qpid/messaging/Sender.cpp
@@ -43,5 +43,6 @@ void Sender::cancel() { impl->cancel(); }
 void Sender::setCapacity(uint32_t c) { impl->setCapacity(c); }
 uint32_t Sender::getCapacity() { return impl->getCapacity(); }
 uint32_t Sender::pending() { return impl->pending(); }
+const std::string& Sender::getName() const { return impl->getName(); }
 
 }} // namespace qpid::messaging
diff --git a/qpid/cpp/src/qpid/messaging/SenderImpl.h b/qpid/cpp/src/qpid/messaging/SenderImpl.h
index fa3794ca4e..058ecf96f8 100644
--- a/qpid/cpp/src/qpid/messaging/SenderImpl.h
+++ b/qpid/cpp/src/qpid/messaging/SenderImpl.h
@@ -40,6 +40,7 @@ class SenderImpl : public virtual qpid::RefCounted
     virtual void setCapacity(uint32_t) = 0;
     virtual uint32_t getCapacity() = 0;
     virtual uint32_t pending() = 0;
+    virtual const std::string& getName() const = 0;
   private:
 };
 }} // namespace qpid::messaging
diff --git a/qpid/cpp/src/qpid/messaging/Session.cpp b/qpid/cpp/src/qpid/messaging/Session.cpp
index 53e85d53b1..5c702051d7 100644
--- a/qpid/cpp/src/qpid/messaging/Session.cpp
+++ b/qpid/cpp/src/qpid/messaging/Session.cpp
@@ -89,4 +89,15 @@ Receiver Session::nextReceiver(qpid::sys::Duration timeout)
 uint32_t Session::available() { return impl->available(); }
 uint32_t Session::pendingAck() { return impl->pendingAck(); }
 
+Sender Session::getSender(const std::string& name) const
+{ 
+    return impl->getSender(name); 
+}
+Receiver Session::getReceiver(const std::string& name) const
+{ 
+    return impl->getReceiver(name); 
+}
+
+KeyError::KeyError(const std::string& msg) : Exception(msg) {}
+
 }} // namespace qpid::messaging
diff --git a/qpid/cpp/src/qpid/messaging/SessionImpl.h b/qpid/cpp/src/qpid/messaging/SessionImpl.h
index b68baf821c..c4139a27ea 100644
--- a/qpid/cpp/src/qpid/messaging/SessionImpl.h
+++ b/qpid/cpp/src/qpid/messaging/SessionImpl.h
@@ -54,6 +54,8 @@ class SessionImpl : public virtual qpid::RefCounted
     virtual Receiver nextReceiver(qpid::sys::Duration timeout) = 0;
     virtual uint32_t available() = 0;
     virtual uint32_t pendingAck() = 0;
+    virtual Sender getSender(const std::string& name) const = 0;
+    virtual Receiver getReceiver(const std::string& name) const = 0;
   private:
 };
 }} // namespace qpid::messaging
diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
index ec0e174fc8..16455bc974 100644
--- a/qpid/cpp/src/tests/MessagingSessionTests.cpp
+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
@@ -694,6 +694,35 @@ QPID_AUTO_TEST_CASE(testAssertPolicyQueue)
     fix.admin.deleteQueue("q");
 }
 
+QPID_AUTO_TEST_CASE(testGetSender)
+{
+    QueueFixture fix;
+    std::string name = fix.session.createSender(fix.queue).getName();
+    Sender sender = fix.session.getSender(name);
+    BOOST_CHECK_EQUAL(name, sender.getName());
+    Message out(Uuid(true).str());
+    sender.send(out);
+    Message in;
+    BOOST_CHECK(fix.session.createReceiver(fix.queue).fetch(in));
+    BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+    BOOST_CHECK_THROW(fix.session.getSender("UnknownSender"), qpid::messaging::KeyError);
+}
+
+
+QPID_AUTO_TEST_CASE(testGetReceiver)
+{
+    QueueFixture fix;
+    std::string name = fix.session.createReceiver(fix.queue).getName();
+    Receiver receiver = fix.session.getReceiver(name);
+    BOOST_CHECK_EQUAL(name, receiver.getName());
+    Message out(Uuid(true).str());
+    fix.session.createSender(fix.queue).send(out);
+    Message in;
+    BOOST_CHECK(receiver.fetch(in));
+    BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
+    BOOST_CHECK_THROW(fix.session.getReceiver("UnknownReceiver"), qpid::messaging::KeyError);
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
 }} // namespace qpid::tests