qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r943489 - in /qpid/trunk/qpid/cpp/src/qpid/client: SslConnector.cpp amqp0_10/ConnectionImpl.cpp amqp0_10/FailoverUpdates.cpp
Date Wed, 12 May 2010 13:41:41 GMT
Author: aconway
Date: Wed May 12 13:41:41 2010
New Revision: 943489

URL: http://svn.apache.org/viewvc?rev=943489&view=rev
Log:
Fixes to new API impl to support failover.

- client/SslConnector.cpp: throw ConnectionException for errors during connection.
- client/amqp0_10/ConnectionImpl.cpp: translate unknown exceptions to ConnectionError.
- client/amqp0_10/FailoverUpdates.cpp: interrupt receiver by closing session in dtor.

Modified:
    qpid/trunk/qpid/cpp/src/qpid/client/SslConnector.cpp
    qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
    qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/FailoverUpdates.cpp

Modified: qpid/trunk/qpid/cpp/src/qpid/client/SslConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SslConnector.cpp?rev=943489&r1=943488&r2=943489&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SslConnector.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SslConnector.cpp Wed May 12 13:41:41 2010
@@ -194,7 +194,7 @@ void SslConnector::connect(const std::st
         socket.connect(host, port);
     } catch (const std::exception& e) {
         socket.close();
-        throw;
+        throw ConnectionException(framing::connection::CLOSE_CODE_FRAMING_ERROR, e.what());
     }
 
     identifier = str(format("[%1% %2%]") % socket.getLocalPort() % socket.getPeerAddress());

Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp?rev=943489&r1=943488&r2=943489&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp Wed May 12 13:41:41 2010
@@ -211,7 +211,11 @@ void ConnectionImpl::open()
 {
     qpid::sys::AbsTime start = qpid::sys::now();
     qpid::sys::ScopedLock<qpid::sys::Semaphore> l(semaphore);
-    if (!connection.isOpen()) connect(start);
+    try {
+        if (!connection.isOpen()) connect(start);
+    }
+    catch (const types::Exception&) { throw; }
+    catch (const qpid::Exception& e) { throw messaging::ConnectionError(e.what()); }
 }
 
 bool expired(const qpid::sys::AbsTime& start, int64_t timeout)

Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/FailoverUpdates.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/FailoverUpdates.cpp?rev=943489&r1=943488&r2=943489&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/FailoverUpdates.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/FailoverUpdates.cpp Wed May 12 13:41:41 2010
@@ -42,42 +42,42 @@ struct FailoverUpdatesImpl : qpid::sys::
     Session session;
     Receiver receiver;
     qpid::sys::Thread thread;
-    volatile bool quit;
 
-    FailoverUpdatesImpl(Connection& c) : connection(c), quit(false)
+    FailoverUpdatesImpl(Connection& c) : connection(c)
     {
         session = connection.createSession("failover-updates");
         receiver = session.createReceiver("amq.failover");
         thread = qpid::sys::Thread(*this);
     }
 
+    ~FailoverUpdatesImpl() {
+        receiver.close();
+        session.close();
+        thread.join();
+    }
+
     void run()
     {
         try {
             Message message;
-            while (!quit && receiver.fetch(message)) {
+            while (receiver.fetch(message)) {
                 connection.setOption("reconnect-urls", message.getProperties()["amq.failover"]);
                 QPID_LOG(debug, "Set reconnect-urls to " << message.getProperties()["amq.failover"]);
                 session.acknowledge();
             }
-        } catch (const qpid::TransportFailure& e) {
+        }
+        catch (const ClosedException&) {}
+        catch (const qpid::TransportFailure& e) {
             QPID_LOG(warning, "Failover updates stopped on loss of connection. " <<
e.what());
-        } catch (const std::exception& e) {
+        }
+        catch (const std::exception& e) {
             QPID_LOG(warning, "Failover updates stopped due to exception: " << e.what());
         }
-        receiver.close();
-        session.close();
-    }
-
-    void wait()
-    {
-        quit = true;
-        thread.join();
     }
 };
 
 FailoverUpdates::FailoverUpdates(Connection& connection) : impl(new FailoverUpdatesImpl(connection))
{}
-FailoverUpdates::~FailoverUpdates() { if (impl) { impl->wait(); delete impl; } }
+FailoverUpdates::~FailoverUpdates() { if (impl) { delete impl; } }
 FailoverUpdates::FailoverUpdates(const FailoverUpdates&) : impl(0) {}
 FailoverUpdates& FailoverUpdates::operator=(const FailoverUpdates&) { return *this;
}
 



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message