QPID-5568: HA C++ qpid::messaging AMQP 1.0 client failover logging is not clear

The qpid::messaging AMQP 1.0 protocol logging did not give clear information
about reconnection during failover.

This patch simplifies the reconnect logic by collapsing all known addresses from
broker URL and reconnect URLs into a single URL with no duplicates.

It rationalizes the info and notice logging as follows:

# Initial connection with multiple addresses, info logs show the
# full URL, each attempt to connect and the finally connected address.

[Messaging] info Starting connection to amqp:tcp:20.0.10.33:5672,tcp:20.0.10.34:5672,tcp:20.0.10.35:5672
[Messaging] info Connecting to tcp:20.0.10.33:5672
[Messaging] info Failed to connect to tcp:20.0.10.33:5672
[Messaging] info Connecting to tcp:20.0.10.34:5672
[Messaging] info Failed to connect to tcp:20.0.10.34:5672
[Messaging] info Connecting to tcp:20.0.10.35:5672
[Messaging] info Connected to tcp:20.0.10.35:5672

# Re-connection due to a failure. notice logs for the start of reconnection (with full URL)
# and eventual sucess (with individual address). info logs for individual connection attempts.

[Messaging] notice Auto-reconnecting to amqp:tcp:20.0.10.33:5672,tcp:20.0.10.34:5672,tcp:20.0.10.35:5672
[Messaging] info Connecting to tcp:20.0.10.33:5672
[Messaging] info Failed to connect to tcp:20.0.10.33:5672
....
[Messaging] info Connected to tcp:20.0.10.33:5672
[Messaging] notice Auto-reconnected to amqp:tcp:20.0.10.33:5672

The idea here is that there are no logs by default (info is not on by default)
for "normal" behavior, but failover does get a (short) notice log by default.
By turning on info logs you can follow the detailed blow-by-blow of failover
without getting drowned in the detail of debug logs.

Note that final failure to connect is signalled to the application via an exception.
There was not previously any log message for that and I didn't add one.

Additional changes: updated qpid-ping test client to use the messaging library.

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

1 files changed, 46 insertions, 39 deletions

diff --git a/qpid/cpp/src/tests/qpid-ping.cpp b/qpid/cpp/src/tests/qpid-ping.cpp
index 52331499e7..f9b1ec17f1 100644
--- a/qpid/cpp/src/tests/qpid-ping.cpp
+++ b/qpid/cpp/src/tests/qpid-ping.cpp
@@ -20,68 +20,75 @@
 
  */
 
-#include "TestOptions.h"
-#include "qpid/client/SubscriptionManager.h"
-#include "qpid/client/Connection.h"
-#include "qpid/client/AsyncSession.h"
-#include "qpid/sys/Time.h"
-#include "qpid/sys/Thread.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/sys/Monitor.h"
-#include "qpid/framing/Uuid.h"
+#include <qpid/messaging/Address.h>
+#include <qpid/messaging/Connection.h>
+#include "qpid/messaging/Duration.h"
+#include <qpid/messaging/Message.h>
+#include <qpid/messaging/Sender.h>
+#include <qpid/messaging/Receiver.h>
+#include <qpid/messaging/Session.h>
+#include <qpid/Msg.h>
+#include <qpid/Options.h>
+#include <qpid/types/Uuid.h>
 #include <string>
 #include <iostream>
 
-using std::cerr;
-using std::cout;
-using std::endl;
-using std::exception;
-using std::string;
-using namespace qpid::client::arg; // For keyword args
-using qpid::client::AsyncSession;
-using qpid::client::Connection;
-using qpid::client::Message;
-using qpid::client::SubscriptionManager;
-using qpid::framing::Uuid;
+using namespace std;
+using namespace qpid::messaging;
+using qpid::types::Uuid;
 
-namespace qpid {
-namespace tests {
+namespace {
 
-struct PingOptions : public qpid::TestOptions {
-    int timeout;                // Timeout in seconds.
+struct PingOptions : public qpid::Options {
+    string url;
+    string address;
+    string message;
+    string connectionOptions;
+    double timeout;             // Timeout in seconds.
     bool quiet;                 // No output
-    PingOptions() : timeout(1), quiet(false) {
+
+    PingOptions() :
+        url("127.0.0.1"),
+        address(Uuid(true).str()+";{create:always}"),
+        message(Uuid(true).str()),
+        timeout(1),
+        quiet(false)
+    {
+        using qpid::optValue;
         addOptions()
+            ("broker,b", qpid::optValue(url, "URL"), "url of broker to connect to.")
+            ("address,a", qpid::optValue(address, "ADDRESS"), "address to use.")
+            ("message,m", optValue(message, "MESSAGE"), "message text to send.")
+            ("connection-options", optValue(connectionOptions, "OPTIONS"), "options for the connection.")
             ("timeout,t", optValue(timeout, "SECONDS"), "Max time to wait.")
             ("quiet,q", optValue(quiet), "Don't print anything to stderr/stdout.");
     }
 };
 
-}} // namespace qpid::tests
+} // namespace
 
 int main(int argc, char** argv) {
+    Connection connection;
+    PingOptions opts;
     try {
-        qpid::tests::PingOptions opts;
         opts.parse(argc, argv);
-        opts.con.heartbeat = (opts.timeout+1)/2;
-        Connection connection;
-        opts.open(connection);
+        connection = Connection(opts.url, opts.connectionOptions);
+        connection.open();
         if (!opts.quiet) cout << "Opened connection." << endl;
-        AsyncSession s = connection.newSession();
-        string qname(Uuid(true).str());
-        s.queueDeclare(queue=qname, autoDelete=true, exclusive=true);
-        s.messageTransfer(content=Message("hello", qname));
+        Session s = connection.createSession();
+        s.createSender(opts.address).send(Message(opts.message));
         if (!opts.quiet) cout << "Sent message." << endl;
-        SubscriptionManager subs(s);
-        subs.get(qname);
+        Message m = s.createReceiver(opts.address).
+            fetch(Duration(opts.timeout*1000));
+        if (m.getContent() != opts.message)
+            throw qpid::Exception(qpid::Msg() << "Expected " << opts.message
+                                  << " but received " << m.getContent());
         if (!opts.quiet) cout << "Received message." << endl;
-        s.sync();
-        s.close();
         connection.close();
-        if (!opts.quiet) cout << "Success." << endl;
         return 0;
     } catch (const exception& e) {
         cerr << "Error: " << e.what() << endl;
+        connection.close();
         return 1;
     }
 }