1 files changed, 161 insertions, 41 deletions

diff --git a/cpp/src/qpid/client/ConnectionImpl.cpp b/cpp/src/qpid/client/ConnectionImpl.cpp
index 5e8596cacb..cede7f7310 100644
--- a/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -18,55 +18,97 @@
  * under the License.
  *
  */
-#include "ConnectionImpl.h"
-#include "Connector.h"
-#include "ConnectionSettings.h"
-#include "SessionImpl.h"
+#include "qpid/client/ConnectionImpl.h"
+#include "qpid/client/Connector.h"
+#include "qpid/client/ConnectionSettings.h"
+#include "qpid/client/SessionImpl.h"
 
 #include "qpid/log/Statement.h"
-#include "qpid/framing/constants.h"
+#include "qpid/Url.h"
+#include "qpid/framing/enum.h"
 #include "qpid/framing/reply_exceptions.h"
 
 #include <boost/bind.hpp>
 #include <boost/format.hpp>
 
-using namespace qpid::client;
+#include <limits>
+
+namespace qpid {
+namespace client {
+
 using namespace qpid::framing;
+using namespace qpid::framing::connection;
 using namespace qpid::sys;
-
 using namespace qpid::framing::connection;//for connection error codes
 
+// Get timer singleton  
+Timer& theTimer() {
+    static Mutex timerInitLock;
+    ScopedLock<Mutex> l(timerInitLock);
+
+    static qpid::sys::Timer t;
+    return t;
+}
+
+class HeartbeatTask : public TimerTask {
+    TimeoutHandler& timeout;
+
+    void fire() {
+        // If we ever get here then we have timed out
+        QPID_LOG(debug, "Traffic timeout");
+        timeout.idleIn();
+    }
+
+public:
+    HeartbeatTask(Duration p, TimeoutHandler& t) :
+        TimerTask(p),
+        timeout(t)
+    {}
+};
+
 ConnectionImpl::ConnectionImpl(framing::ProtocolVersion v, const ConnectionSettings& settings)
     : Bounds(settings.maxFrameSize * settings.bounds),
       handler(settings, v),
-      connector(new Connector(v, settings, this)), 
-      version(v)
+      version(v),
+      nextChannel(1)
 {
-    QPID_LOG(debug, "ConnectionImpl created for " << version);
+    QPID_LOG(debug, "ConnectionImpl created for " << version.toString());
     handler.in = boost::bind(&ConnectionImpl::incoming, this, _1);
     handler.out = boost::bind(&Connector::send, boost::ref(connector), _1);
     handler.onClose = boost::bind(&ConnectionImpl::closed, this,
-                                  NORMAL, std::string());
-    connector->setInputHandler(&handler);
-    connector->setShutdownHandler(this);
-
+                                  CLOSE_CODE_NORMAL, std::string());
     //only set error handler once  open
     handler.onError = boost::bind(&ConnectionImpl::closed, this, _1, _2);
+    handler.getSSF = boost::bind(&Connector::getSSF, boost::ref(connector));
 }
 
+const uint16_t ConnectionImpl::NEXT_CHANNEL = std::numeric_limits<uint16_t>::max();
+
 ConnectionImpl::~ConnectionImpl() {
     // Important to close the connector first, to ensure the
     // connector thread does not call on us while the destructor
     // is running.
-    connector->close(); 
+    if (connector) connector->close();
 }
 
-void ConnectionImpl::addSession(const boost::shared_ptr<SessionImpl>& session)
+void ConnectionImpl::addSession(const boost::shared_ptr<SessionImpl>& session, uint16_t channel)
 {
     Mutex::ScopedLock l(lock);
-    boost::weak_ptr<SessionImpl>& s = sessions[session->getChannel()];
-    if (s.lock()) throw SessionBusyException();
-    s = session;
+    for (uint16_t i = 0; i < NEXT_CHANNEL; i++) { //will at most search through channels once
+        uint16_t c = channel == NEXT_CHANNEL ? nextChannel++ : channel;
+        boost::weak_ptr<SessionImpl>& s = sessions[c];
+        boost::shared_ptr<SessionImpl> ss = s.lock();
+        if (!ss) {
+            //channel is free, we can assign it to this session
+            session->setChannel(c);
+            s = session;
+            return;
+        } else if (channel != NEXT_CHANNEL) {
+            //channel is taken and was requested explicitly so don't look for another
+            throw SessionBusyException(QPID_MSG("Channel " << ss->getChannel() << " attached to " << ss->getId()));
+        } //else channel is busy, but we can keep looking for a free one
+    }
+
 }
 
 void ConnectionImpl::handle(framing::AMQFrame& frame)
@@ -81,9 +123,11 @@ void ConnectionImpl::incoming(framing::AMQFrame& frame)
         Mutex::ScopedLock l(lock);
         s = sessions[frame.getChannel()].lock();
     }
-    if (!s)
-        throw NotAttachedException(QPID_MSG("Invalid channel: " << frame.getChannel()));
-    s->in(frame);
+    if (!s) {
+        QPID_LOG(info, "Dropping frame received on invalid channel: " << frame);
+    } else {
+        s->in(frame);
+    }
 }
 
 bool ConnectionImpl::isOpen() const 
@@ -92,57 +136,121 @@ bool ConnectionImpl::isOpen() const
 }
 
 
-void ConnectionImpl::open(const std::string& host, int port)
+void ConnectionImpl::open()
 {
-    QPID_LOG(info, "Connecting to " << host << ":" << port);
+    const std::string& protocol = handler.protocol;
+    const std::string& host = handler.host;
+    int port = handler.port;
+    QPID_LOG(info, "Connecting to " << protocol << ":" << host << ":" << port);
+
+    connector.reset(Connector::create(protocol, version, handler, this)); 
+    connector->setInputHandler(&handler);
+    connector->setShutdownHandler(this);
     connector->connect(host, port);
     connector->init();
-    handler.waitForOpen();    
+ 
+    // Enable heartbeat if requested
+    uint16_t heartbeat = static_cast<ConnectionSettings&>(handler).heartbeat;
+    if (heartbeat) {
+        // Set connection timeout to be 2x heart beat interval and setup timer
+        heartbeatTask = new HeartbeatTask(heartbeat * 2 * TIME_SEC, *this);
+        handler.setRcvTimeoutTask(heartbeatTask);
+        theTimer().add(heartbeatTask);
+    }
+
+    try {
+        handler.waitForOpen();
+    } catch (...) {
+        // Make sure the connector thread is joined.
+        connector->close();
+        throw;
+    }
+
+    // If the SASL layer has provided an "operational" userId for the connection,
+    // put it in the negotiated settings.
+    const std::string& userId(handler.getUserId());
+    if (!userId.empty())
+        handler.username = userId;
+
+    //enable security layer if one has been negotiated:
+    std::auto_ptr<SecurityLayer> securityLayer = handler.getSecurityLayer();
+    if (securityLayer.get()) {
+        QPID_LOG(debug, "Activating security layer");
+        connector->activateSecurityLayer(securityLayer);
+    } else {
+        QPID_LOG(debug, "No security layer in place");
+    }
 }
 
 void ConnectionImpl::idleIn()
 {
-    close();
+    connector->abort();
 }
 
 void ConnectionImpl::idleOut()
 {
-    AMQFrame frame(in_place<AMQHeartbeatBody>());
+    AMQFrame frame((AMQHeartbeatBody()));
     connector->send(frame);
 }
 
 void ConnectionImpl::close()
 {
-    if (!handler.isOpen()) return;
-    handler.close();
-    closed(NORMAL, "Closed by client");
+    if (heartbeatTask) 
+        heartbeatTask->cancel();
+    // close() must be idempotent and no-throw as it will often be called in destructors.
+    if (handler.isOpen()) {
+        try {
+            handler.close();
+            closed(CLOSE_CODE_NORMAL, "Closed by client");
+        } catch (...) {}
+    }
+    assert(!handler.isOpen());
 }
 
 
 template <class F> void ConnectionImpl::closeInternal(const F& f) {
-    connector->close();
-    for (SessionMap::iterator i = sessions.begin(); i != sessions.end(); ++i) {
+    if (heartbeatTask) {
+        heartbeatTask->cancel();
+    }
+    {
+        Mutex::ScopedUnlock u(lock);
+        connector->close();
+    }
+    //notifying sessions of failure can result in those session being
+    //deleted which in turn results in a call to erase(); this can
+    //even happen on this thread, when 's' goes out of scope
+    //below. Using a copy prevents the map being modified as we
+    //iterate through.
+    SessionMap copy;
+    sessions.swap(copy);
+    for (SessionMap::iterator i = copy.begin(); i != copy.end(); ++i) {
         boost::shared_ptr<SessionImpl> s = i->second.lock();
         if (s) f(s);
     }
-    sessions.clear();
 }
 
 void ConnectionImpl::closed(uint16_t code, const std::string& text) { 
     Mutex::ScopedLock l(lock);
-    setException(new ConnectionException(code, text));
+    setException(new ConnectionException(ConnectionHandler::convert(code), text));
     closeInternal(boost::bind(&SessionImpl::connectionClosed, _1, code, text));
 }
 
-static const std::string CONN_CLOSED("Connection closed by broker");
+static const std::string CONN_CLOSED("Connection closed");
 
 void ConnectionImpl::shutdown() {
-    Mutex::ScopedLock l(lock);
-    // FIXME aconway 2008-06-06: exception use, connection-forced is incorrect here.
-    setException(new ConnectionException(CONNECTION_FORCED, CONN_CLOSED));
+    if ( failureCallback )
+      failureCallback();
+
     if (handler.isClosed()) return;
-    handler.fail(CONN_CLOSED);
-    closeInternal(boost::bind(&SessionImpl::connectionBroke, _1, CONNECTION_FORCED, CONN_CLOSED));
+
+    // FIXME aconway 2008-06-06: exception use, amqp0-10 does not seem to have
+    // an appropriate close-code. connection-forced is not right.
+    bool isClosing = handler.isClosing();
+    handler.fail(CONN_CLOSED);//ensure connection is marked as failed before notifying sessions
+    Mutex::ScopedLock l(lock);
+    if (!isClosing) 
+        closeInternal(boost::bind(&SessionImpl::connectionBroke, _1, CONN_CLOSED));
+    setException(new TransportFailure(CONN_CLOSED));
 }
 
 void ConnectionImpl::erase(uint16_t ch) {
@@ -154,4 +262,16 @@ const ConnectionSettings& ConnectionImpl::getNegotiatedSettings()
 {
     return handler;
 }
-    
+
+std::vector<qpid::Url> ConnectionImpl::getInitialBrokers() {
+    return handler.knownBrokersUrls;
+}
+
+boost::shared_ptr<SessionImpl>  ConnectionImpl::newSession(const std::string& name, uint32_t timeout, uint16_t channel) {
+    boost::shared_ptr<SessionImpl> simpl(new SessionImpl(name, shared_from_this()));
+    addSession(simpl, channel);
+    simpl->open(timeout);
+    return simpl;
+}
+
+}} // namespace qpid::client