You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2019/08/06 18:58:34 UTC
[qpid-proton] branch master updated: PROTON-2040: Allow connection
options to be updated for improved reconnect - Added connection method -
update_options() - Added connection options - reconnect_url This allows the
Address Url used to reconnect to be controlled - Moved failover_urls option
to connection options directly - Improve the behaviour of failover urls If
failover urls are set they will be tried immediately after the initial
connection url rather than retrying that first.
This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
The following commit(s) were added to refs/heads/master by this push:
new 4f5a20f PROTON-2040: Allow connection options to be updated for improved reconnect - Added connection method - update_options() - Added connection options - reconnect_url This allows the Address Url used to reconnect to be controlled - Moved failover_urls option to connection options directly - Improve the behaviour of failover urls If failover urls are set they will be tried immediately after the initial connection url rather than retrying that first.
4f5a20f is described below
commit 4f5a20ff0bc0c038e511292dddb7ab93245d9879
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Mon May 6 21:22:06 2019 -0400
PROTON-2040: Allow connection options to be updated for improved reconnect
- Added connection method - update_options()
- Added connection options - reconnect_url
This allows the Address Url used to reconnect to be controlled
- Moved failover_urls option to connection options directly
- Improve the behaviour of failover urls
If failover urls are set they will be tried immediately after the initial
connection url rather than retrying that first.
---
c/src/core/connection_driver.c | 5 +-
cpp/include/proton/connection.hpp | 15 ++-
cpp/include/proton/connection_options.hpp | 31 ++++-
cpp/include/proton/reconnect_options.hpp | 3 +
cpp/src/connection.cpp | 5 +
cpp/src/connection_options.cpp | 46 +++++++-
cpp/src/contexts.cpp | 5 +-
cpp/src/contexts.hpp | 11 +-
cpp/src/proactor_container_impl.cpp | 67 ++++++-----
cpp/src/proactor_container_impl.hpp | 2 -
cpp/src/reconnect_options.cpp | 2 +-
cpp/src/reconnect_options_impl.hpp | 11 +-
cpp/src/reconnect_test.cpp | 184 +++++++++++++++++++++++++++++-
13 files changed, 335 insertions(+), 52 deletions(-)
diff --git a/c/src/core/connection_driver.c b/c/src/core/connection_driver.c
index 7120f2c..b7d71fe 100644
--- a/c/src/core/connection_driver.c
+++ b/c/src/core/connection_driver.c
@@ -77,12 +77,13 @@ int pn_connection_driver_bind(pn_connection_driver_t *d) {
pn_connection_t *pn_connection_driver_release_connection(pn_connection_driver_t *d) {
if (d->transport) { /* Make sure transport is closed and unbound */
- pn_connection_driver_close(d);
- pn_transport_unbind(d->transport);
+ pn_connection_driver_close(d);
+ pn_transport_unbind(d->transport);
}
pn_connection_t *c = d->connection;
if (c) {
d->connection = NULL;
+ pn_connection_reset(c);
pn_connection_collect(c, NULL); /* Disconnect from the collector */
}
return c;
diff --git a/cpp/include/proton/connection.hpp b/cpp/include/proton/connection.hpp
index ee90a04..2dcfb16 100644
--- a/cpp/include/proton/connection.hpp
+++ b/cpp/include/proton/connection.hpp
@@ -175,12 +175,25 @@ PN_CPP_CLASS_EXTERN connection : public internal::object<pn_connection_t>, publi
/// execute code safely in the event-handler thread.
PN_CPP_EXTERN void wake() const;
- /// **Unsettled API** - true if this connection has been automatically
+ /// **Unsettled API** - True if this connection has been automatically
/// re-connected.
///
/// @see reconnect_options, messaging_handler
PN_CPP_EXTERN bool reconnected() const;
+ /// **Unsettled API** - Update the connection options for this connection
+ ///
+ /// This method can be used to update the connection options used with this
+ /// connection. Usually the connection options are only used during the initial
+ /// connection attempt so this ability is only useful when automatically
+ /// reconnect is enabled and you wish to change the connection options before
+ /// the next reconnect attempt. This would usually be in the handler for the
+ /// `on_transport_error` event @see messaging_handler.
+ ///
+ /// @note Connection options supplied in the parameter will be merged with the
+ /// existing parameters as if `connection_options::update()` was used.
+ PN_CPP_EXTERN void update_options(const connection_options&);
+
/// @cond INTERNAL
friend class internal::factory<connection>;
friend class container;
diff --git a/cpp/include/proton/connection_options.hpp b/cpp/include/proton/connection_options.hpp
index 6da610c..d139251 100644
--- a/cpp/include/proton/connection_options.hpp
+++ b/cpp/include/proton/connection_options.hpp
@@ -168,14 +168,41 @@ class connection_options {
/// **Unsettled API** - Set the SASL configuration path.
PN_CPP_EXTERN connection_options& sasl_config_path(const std::string&);
- /// **Unsettled API** - Set reconnect and failover options.
- PN_CPP_EXTERN connection_options& reconnect(const reconnect_options &);
+ /// **Unsettled API** - Set reconnect timing options.
+ ///
+ /// If reconnect timing options are set, but no specific reconnect or failover urls are set
+ /// then the original url used for the connection will be used as the reconnect url.
+ PN_CPP_EXTERN connection_options& reconnect(const reconnect_options&);
+
+ /// **Unsettled API** - Set reconnect URL.
+ ///
+ /// If the connection to the primary connection url fails then try to reconnect using
+ /// the reconnect url.
+ ///
+ /// If this option is set then it will by default set the default reconnect timing options.
+ ///
+ /// If this option is changed using @ref connection::update_options before a reconnection
+ /// attempt (for example in the @ref messaging_handler::on_transport_error callback) then
+ /// the new value of the reconnect url will be used for the reconnect attempt rather then
+ /// any previous value.
+ ///
+ /// If both the failover_urls and reconnect_url options are set then the behavior is not defined.
+ PN_CPP_EXTERN connection_options& reconnect_url(const std::string&);
+
+ /// **Unsettled API** - Set Fail-over URLs.
+ ///
+ /// If the connection to the primary connection url fails then try each of the fail-over
+ /// connection urls in turn.
+ ///
+ /// If both the failover_urls and reconnect_url options are set then the behavior is not defined.
+ PN_CPP_EXTERN connection_options& failover_urls(const std::vector<std::string>&);
/// Update option values from values set in other.
PN_CPP_EXTERN connection_options& update(const connection_options& other);
private:
void apply_unbound(connection&) const;
+ void apply_reconnect_urls(pn_connection_t* pnc) const;
void apply_unbound_client(pn_transport_t*) const;
void apply_unbound_server(pn_transport_t*) const;
messaging_handler* handler() const;
diff --git a/cpp/include/proton/reconnect_options.hpp b/cpp/include/proton/reconnect_options.hpp
index 2206eec..fab9fb8 100644
--- a/cpp/include/proton/reconnect_options.hpp
+++ b/cpp/include/proton/reconnect_options.hpp
@@ -75,8 +75,10 @@ class reconnect_options {
/// meaning no limit.
PN_CPP_EXTERN reconnect_options& max_attempts(int);
+ /// Deprecated - use connection_options::failover_urls
/// Alternative connection URLs used for failover. There are none
/// by default.
+ PN_CPP_DEPRECATED("use connection_options::failover_urls()")
PN_CPP_EXTERN reconnect_options& failover_urls(const std::vector<std::string>& conn_urls);
private:
@@ -84,6 +86,7 @@ class reconnect_options {
internal::pn_unique_ptr<impl> impl_;
/// @cond INTERNAL
+ friend class connection_options;
friend class container;
/// @endcond
};
diff --git a/cpp/src/connection.cpp b/cpp/src/connection.cpp
index f0219cd..c6686ae 100644
--- a/cpp/src/connection.cpp
+++ b/cpp/src/connection.cpp
@@ -198,4 +198,9 @@ bool connection::reconnected() const {
return (rc && rc->reconnected_);
}
+void connection::update_options(const connection_options& options) {
+ connection_context& cc = connection_context::get(pn_object());
+ cc.connection_options_->update(options);
+}
+
} // namespace proton
diff --git a/cpp/src/connection_options.cpp b/cpp/src/connection_options.cpp
index d47d049..2bf281d 100644
--- a/cpp/src/connection_options.cpp
+++ b/cpp/src/connection_options.cpp
@@ -33,6 +33,7 @@
#include "messaging_adapter.hpp"
#include "msg.hpp"
#include "proton_bits.hpp"
+#include "reconnect_options_impl.hpp"
#include "ssl_options_impl.hpp"
#include <proton/connection.h>
@@ -62,7 +63,9 @@ class connection_options::impl {
option<std::string> password;
option<std::vector<symbol> > offered_capabilities;
option<std::vector<symbol> > desired_capabilities;
- option<reconnect_options> reconnect;
+ option<reconnect_options_base> reconnect;
+ option<std::string> reconnect_url;
+ option<std::vector<std::string> > failover_urls;
option<class ssl_client_options> ssl_client_options;
option<class ssl_server_options> ssl_server_options;
option<bool> sasl_enabled;
@@ -85,8 +88,14 @@ class connection_options::impl {
bool uninit = c.uninitialized();
if (!uninit) return;
- if (reconnect.set)
- connection_context::get(pnc).reconnect_context_.reset(new reconnect_context(reconnect.value));
+ if (reconnect.set || reconnect_url.set || failover_urls.set) {
+ // Transfer reconnect options from connection options to reconnect contexts
+ // to stop the reconnect context being reset every retry unless there are new options
+ connection_context::get(pnc).reconnect_context_.reset(
+ new reconnect_context(
+ reconnect.set ? reconnect.value : reconnect_options_base()));
+ reconnect.set = false;
+ }
if (container_id.set)
pn_connection_set_container(pnc, container_id.value.c_str());
if (virtual_host.set)
@@ -101,6 +110,24 @@ class connection_options::impl {
value(pn_connection_desired_capabilities(pnc)) = desired_capabilities.value;
}
+ void apply_reconnect_urls(pn_connection_t* pnc) {
+ connection_context& cc = connection_context::get(pnc);
+ // If there no reconnect delay parameters create default
+ if ((reconnect_url.set || failover_urls.set) && !cc.reconnect_context_)
+ cc.reconnect_context_.reset(new reconnect_context(reconnect_options_base()));
+ if (reconnect_url.set) {
+ reconnect_url.set = false;
+ cc.reconnect_url_ = reconnect_url.value;
+ cc.reconnect_context_->current_url_ = -1;
+ }
+ if (failover_urls.set) {
+ failover_urls.set = false;
+ cc.failover_urls_ = failover_urls.value;
+ cc.reconnect_context_->current_url_ = 0;
+ }
+
+ }
+
void apply_transport(pn_transport_t* pnt) {
if (max_frame_size.set)
pn_transport_set_max_frame(pnt, max_frame_size.value);
@@ -167,6 +194,8 @@ class connection_options::impl {
offered_capabilities.update(x.offered_capabilities);
desired_capabilities.update(x.desired_capabilities);
reconnect.update(x.reconnect);
+ reconnect_url.update(x.reconnect_url);
+ failover_urls.update(x.failover_urls);
ssl_client_options.update(x.ssl_client_options);
ssl_server_options.update(x.ssl_server_options);
sasl_enabled.update(x.sasl_enabled);
@@ -208,7 +237,15 @@ connection_options& connection_options::user(const std::string &user) { impl_->u
connection_options& connection_options::password(const std::string &password) { impl_->password = password; return *this; }
connection_options& connection_options::offered_capabilities(const std::vector<symbol> &caps) { impl_->offered_capabilities = caps; return *this; }
connection_options& connection_options::desired_capabilities(const std::vector<symbol> &caps) { impl_->desired_capabilities = caps; return *this; }
-connection_options& connection_options::reconnect(const reconnect_options &r) { impl_->reconnect = r; return *this; }
+connection_options& connection_options::reconnect(const reconnect_options &r) {
+ if (!r.impl_->failover_urls.empty()) {
+ impl_->failover_urls = r.impl_->failover_urls;
+ }
+ impl_->reconnect = *r.impl_;
+ return *this;
+}
+connection_options& connection_options::reconnect_url(const std::string& u) { impl_->reconnect_url = u; return *this; }
+connection_options& connection_options::failover_urls(const std::vector<std::string>& us) { impl_->failover_urls = us; return *this; }
connection_options& connection_options::ssl_client_options(const class ssl_client_options &c) { impl_->ssl_client_options = c; return *this; }
connection_options& connection_options::ssl_server_options(const class ssl_server_options &c) { impl_->ssl_server_options = c; return *this; }
connection_options& connection_options::sasl_enabled(bool b) { impl_->sasl_enabled = b; return *this; }
@@ -218,6 +255,7 @@ connection_options& connection_options::sasl_config_name(const std::string &n) {
connection_options& connection_options::sasl_config_path(const std::string &p) { impl_->sasl_config_path = p; return *this; }
void connection_options::apply_unbound(connection& c) const { impl_->apply_unbound(c); }
+void connection_options::apply_reconnect_urls(pn_connection_t *c) const { impl_->apply_reconnect_urls(c); }
void connection_options::apply_unbound_client(pn_transport_t *t) const { impl_->apply_sasl(t); impl_->apply_ssl(t, true); impl_->apply_transport(t); }
void connection_options::apply_unbound_server(pn_transport_t *t) const { impl_->apply_sasl(t); impl_->apply_ssl(t, false); impl_->apply_transport(t); }
diff --git a/cpp/src/contexts.cpp b/cpp/src/contexts.cpp
index 9689de2..65d883a 100644
--- a/cpp/src/contexts.cpp
+++ b/cpp/src/contexts.cpp
@@ -22,6 +22,7 @@
#include "contexts.hpp"
#include "msg.hpp"
+#include "reconnect_options_impl.hpp"
#include "proton_bits.hpp"
#include "proton/connection_options.hpp"
@@ -71,8 +72,8 @@ connection_context::connection_context() :
container(0), default_session(0), link_gen(0), handler(0), listener_context_(0)
{}
-reconnect_context::reconnect_context(const reconnect_options& ro) :
- reconnect_options_(new reconnect_options(ro)), retries_(0), current_url_(-1), reconnected_(false)
+reconnect_context::reconnect_context(const reconnect_options_base& ro) :
+ reconnect_options_(ro), retries_(0), current_url_(-1), reconnected_(false)
{}
listener_context::listener_context() : listen_handler_(0) {}
diff --git a/cpp/src/contexts.hpp b/cpp/src/contexts.hpp
index 4a046a4..6660b55 100644
--- a/cpp/src/contexts.hpp
+++ b/cpp/src/contexts.hpp
@@ -22,6 +22,8 @@
*
*/
+#include "reconnect_options_impl.hpp"
+
#include "proton/work_queue.hpp"
#include "proton/message.hpp"
#include "proton/internal/pn_unique_ptr.hpp"
@@ -91,19 +93,22 @@ class connection_context : public context {
io::link_namer* link_gen; // Link name generator.
messaging_handler* handler;
- std::string connected_address_;
+ std::string reconnect_url_;
+ std::vector<std::string> failover_urls_;
internal::pn_unique_ptr<connection_options> connection_options_;
internal::pn_unique_ptr<reconnect_context> reconnect_context_;
listener_context* listener_context_;
work_queue work_queue_;
};
+class reconnect_options_base;
+
// This is not a context object on its own, but an optional part of connection
class reconnect_context {
public:
- reconnect_context(const reconnect_options& ro);
+ reconnect_context(const reconnect_options_base& ro);
- internal::pn_unique_ptr<const reconnect_options> reconnect_options_;
+ const reconnect_options_base reconnect_options_;
duration delay_;
int retries_;
int current_url_;
diff --git a/cpp/src/proactor_container_impl.cpp b/cpp/src/proactor_container_impl.cpp
index a559f03..8cab1c5 100644
--- a/cpp/src/proactor_container_impl.cpp
+++ b/cpp/src/proactor_container_impl.cpp
@@ -166,13 +166,18 @@ void container::impl::remove_work_queue(container::impl::container_work_queue* l
work_queues_.erase(l);
}
-void container::impl::setup_connection_lh(const url& url, pn_connection_t *pnc) {
- pn_connection_set_container(pnc, id_.c_str());
- pn_connection_set_hostname(pnc, url.host().c_str());
+namespace {
+void default_url_options(connection_options& opts, const url& url) {
+ opts.virtual_host(url.host());
if (!url.user().empty())
- pn_connection_set_user(pnc, url.user().c_str());
+ opts.user(url.user());
if (!url.password().empty())
- pn_connection_set_password(pnc, url.password().c_str());
+ opts.password(url.password());
+ // If scheme is amqps then use default tls settings
+ if (url.scheme()==url.AMQPS) {
+ opts.ssl_client_options(ssl_client_options());
+ }
+}
}
pn_connection_t* container::impl::make_connection_lh(
@@ -183,10 +188,8 @@ pn_connection_t* container::impl::make_connection_lh(
throw proton::error("container is stopping");
connection_options opts;
- // If scheme is amqps then use default tls settings
- if (url.scheme()==url.AMQPS) {
- opts.ssl_client_options(ssl_client_options());
- }
+ opts.container_id(id_);
+ default_url_options(opts, url);
opts.update(client_connection_options_);
opts.update(user_opts);
messaging_handler* mh = opts.handler();
@@ -196,11 +199,10 @@ pn_connection_t* container::impl::make_connection_lh(
cc.container = &container_;
cc.handler = mh;
cc.work_queue_ = new container::impl::connection_work_queue(*container_.impl_, pnc);
- cc.connected_address_ = url;
+ cc.reconnect_url_ = url;
cc.connection_options_.reset(new connection_options(opts));
- setup_connection_lh(url, pnc);
- make_wrapper(pnc).open(opts);
+ make_wrapper(pnc).open(*cc.connection_options_);
return pnc; // 1 refcount from pn_connection()
}
@@ -216,12 +218,13 @@ pn_connection_t* container::impl::make_connection_lh(
// after start_connection() which is undefined!
//
void container::impl::start_connection(const url& url, pn_connection_t *pnc) {
- char caddr[PN_MAX_ADDR];
- pn_proactor_addr(caddr, sizeof(caddr), url.host().c_str(), url.port().c_str());
pn_transport_t* pnt = pn_transport();
connection_context& cc = connection_context::get(pnc);
connection_options& co = *cc.connection_options_;
co.apply_unbound_client(pnt);
+
+ char caddr[PN_MAX_ADDR];
+ pn_proactor_addr(caddr, sizeof(caddr), url.host().c_str(), url.port().c_str());
pn_proactor_connect2(proactor_, pnc, pnt, caddr); // Takes ownership of pnc, pnt
}
@@ -237,26 +240,33 @@ void container::impl::reconnect(pn_connection_t* pnc) {
connection_context& cc = connection_context::get(pnc);
reconnect_context& rc = *cc.reconnect_context_.get();
- const reconnect_options::impl& roi = *rc.reconnect_options_->impl_;
+
+ connection_options& co = *cc.connection_options_;
+ co.apply_reconnect_urls(pnc);
// Figure out next connection url to try
// rc.current_url_ == -1 means try the url specified in connect, not a failover url
- const proton::url url(rc.current_url_==-1 ? cc.connected_address_ : roi.failover_urls[rc.current_url_]);
+ const proton::url url(rc.current_url_==-1 ? cc.reconnect_url_ : cc.failover_urls_[rc.current_url_]);
// XXXX Debug:
- //std::cout << "Retries: " << rc.retries_ << " Delay: " << rc.delay_ << " Trying: " << url << std::endl;
-
- setup_connection_lh(url, pnc);
- make_wrapper(pnc).open(*cc.connection_options_);
- start_connection(url, pnc);
+ //std::cout << "Retries: " << rc.retries_ << " Delay: " << rc.delay_ << " Trying: " << url << "@" << rc.current_url_ << std::endl;
+ ++rc.current_url_;
// Did we go through all the urls?
- if (rc.current_url_==int(roi.failover_urls.size())-1) {
+ if (rc.current_url_==int(cc.failover_urls_.size())) {
rc.current_url_ = -1;
++rc.retries_;
- } else {
- ++rc.current_url_;
}
+
+ connection_options opts;
+ opts.container_id(id_);
+ default_url_options(opts, url);
+ opts.update(co);
+ messaging_handler* mh = opts.handler();
+ cc.handler = mh;
+
+ make_wrapper(pnc).open(co);
+ start_connection(url, pnc);
}
namespace {
@@ -273,16 +283,15 @@ duration random_between(duration, duration max)
return max;
}
#endif
-}
-duration container::impl::next_delay(reconnect_context& rc) {
+duration next_delay(reconnect_context& rc) {
// If we've not retried before do it immediately
if (rc.retries_==0) return duration(0);
// If we haven't tried all failover urls yet this round do it immediately
if (rc.current_url_!=-1) return duration(0);
- const reconnect_options::impl& roi = *rc.reconnect_options_->impl_;
+ const reconnect_options_base& roi = rc.reconnect_options_;
if (rc.retries_==1) {
rc.delay_ = roi.delay;
} else {
@@ -290,6 +299,7 @@ duration container::impl::next_delay(reconnect_context& rc) {
}
return random_between(roi.delay, rc.delay_);
}
+}
void container::impl::reset_reconnect(pn_connection_t* pnc) {
connection_context& cc = connection_context::get(pnc);
@@ -299,6 +309,7 @@ void container::impl::reset_reconnect(pn_connection_t* pnc) {
rc->delay_ = 0;
rc->retries_ = 0;
+ // set retry to the initial url next
rc->current_url_ = -1;
}
@@ -321,7 +332,7 @@ bool container::impl::can_reconnect(pn_connection_t* pnc) {
// If reconnect not enabled just fail
if (!rc) return false;
- const reconnect_options::impl& roi = *rc->reconnect_options_->impl_;
+ const reconnect_options_base& roi = rc->reconnect_options_;
pn_transport_t* t = pn_connection_transport(pnc);
pn_condition_t* condition = pn_transport_condition(t);
diff --git a/cpp/src/proactor_container_impl.hpp b/cpp/src/proactor_container_impl.hpp
index 0f0ef60..b58ab0e 100644
--- a/cpp/src/proactor_container_impl.hpp
+++ b/cpp/src/proactor_container_impl.hpp
@@ -103,10 +103,8 @@ class container::impl {
class container_work_queue;
pn_listener_t* listen_common_lh(const std::string&);
pn_connection_t* make_connection_lh(const url& url, const connection_options&);
- void setup_connection_lh(const url& url, pn_connection_t *pnc);
void start_connection(const url& url, pn_connection_t* c);
void reconnect(pn_connection_t* pnc);
- duration next_delay(reconnect_context& rc);
bool can_reconnect(pn_connection_t* pnc);
void setup_reconnect(pn_connection_t* pnc);
void reset_reconnect(pn_connection_t* pnc);
diff --git a/cpp/src/reconnect_options.cpp b/cpp/src/reconnect_options.cpp
index a469d3f..890cf36 100644
--- a/cpp/src/reconnect_options.cpp
+++ b/cpp/src/reconnect_options.cpp
@@ -25,7 +25,7 @@
namespace proton {
reconnect_options::reconnect_options() : impl_(new impl()) {}
-reconnect_options::reconnect_options(const reconnect_options& x) : impl_(new impl()) {
+reconnect_options::reconnect_options(const reconnect_options& x) : impl_(new impl) {
*this = x;
}
reconnect_options::~reconnect_options() {}
diff --git a/cpp/src/reconnect_options_impl.hpp b/cpp/src/reconnect_options_impl.hpp
index 2aca82d..4861292 100644
--- a/cpp/src/reconnect_options_impl.hpp
+++ b/cpp/src/reconnect_options_impl.hpp
@@ -23,20 +23,25 @@
*/
#include "proton/duration.hpp"
+#include "proton/internal/pn_unique_ptr.hpp"
+#include "proton/reconnect_options.hpp"
#include <string>
#include <vector>
namespace proton {
-
-class reconnect_options::impl {
+class reconnect_options_base {
public:
- impl() : delay(10), delay_multiplier(2.0), max_delay(duration::FOREVER), max_attempts(0) {}
+ reconnect_options_base() : delay(10), delay_multiplier(2.0), max_delay(duration::FOREVER), max_attempts(0) {}
duration delay;
float delay_multiplier;
duration max_delay;
int max_attempts;
+};
+
+class reconnect_options::impl : public reconnect_options_base {
+ public:
std::vector<std::string> failover_urls;
};
diff --git a/cpp/src/reconnect_test.cpp b/cpp/src/reconnect_test.cpp
index 72def2a..17907f1 100644
--- a/cpp/src/reconnect_test.cpp
+++ b/cpp/src/reconnect_test.cpp
@@ -148,7 +148,7 @@ class tester : public proton::messaging_handler, public waiter {
std::vector<std::string> urls;
urls.push_back(s2->url());
urls.push_back(s3->url());
- container_.connect(s1->url(), proton::connection_options().reconnect(proton::reconnect_options().failover_urls(urls)));
+ container_.connect(s1->url(), proton::connection_options().failover_urls(urls));
}
void on_connection_open(proton::connection& c) PN_CPP_OVERRIDE {
@@ -185,7 +185,7 @@ class tester : public proton::messaging_handler, public waiter {
container_.run();
ASSERT_EQUAL(1, start_count);
ASSERT_EQUAL(3, open_count);
- // Could be > 3, unpredicatble number reconnects while listener comes up.
+ // Could be > 3, unpredictable number reconnects while listener comes up.
ASSERT(2 < transport_error_count);
// Last reconnect fails before opening links
ASSERT(link_open_count > 1);
@@ -316,12 +316,188 @@ int test_auth_fail_reconnect() {
return 0;
}
-int main(int argc, char** argv) {
+class test_reconnect_url : public proton::messaging_handler {
+public:
+ test_reconnect_url()
+ : errors_(0), container_(*this, "test_reconnect_update") {}
+
+ proton::reconnect_options ropts() {
+ // Fast as we can to avoid needless test slowness.
+ return proton::reconnect_options().delay(proton::duration::MILLISECOND);
+ }
+
+ proton::connection_options copts() { return proton::connection_options(); }
+
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ // Never actually connects, keeps re-trying to bogus hostnames with
+ // changing options.
+ c.connect("nosuchhost0",
+ copts()
+ .reconnect(ropts())
+ .virtual_host("vhost0")
+ .user("user0")
+ .reconnect_url("hahaha1"));
+ }
+
+ void on_transport_error(proton::transport &t) PN_CPP_OVERRIDE {
+ switch (++errors_) {
+ case 1:
+ ASSERT_SUBSTRING("nosuchhost0", t.error().what()); // First failure
+ break;
+ case 2: {
+ ASSERT_SUBSTRING("hahaha1",t.error().what()); // Second failure
+ ASSERT_EQUAL("user0", t.connection().user());
+ break;
+ }
+ case 3:
+ ASSERT_SUBSTRING("hahaha1", t.error().what()); // Still trying reconnect url
+ t.connection().update_options(copts().reconnect_url("nosuchhost1"));
+ // Verify changing reconnect options does not affect other options.
+ ASSERT_EQUAL("user0", t.connection().user());
+ break;
+ case 4:
+ ASSERT_SUBSTRING("nosuchhost1", t.error().what()); // Re-try new reconnect url
+ break;
+ default:
+ t.connection().container().stop();
+ }
+ }
+
+ void run() { container_.run(); }
+
+private:
+ int errors_;
+ proton::container container_;
+};
+
+// Verify we can change connection options for reconnect on_transport_error()
+class test_reconnect_update_failover : public proton::messaging_handler {
+public:
+ test_reconnect_update_failover()
+ : errors_(0), container_(*this, "test_reconnect_update") {}
+
+ proton::reconnect_options ropts() {
+ // Fast as we can to avoid needless test slowness.
+ return proton::reconnect_options().delay(proton::duration::MILLISECOND);
+ }
+
+ proton::connection_options copts() { return proton::connection_options(); }
+
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ // Never actually connects, keeps re-trying to bogus hostnames with
+ // changing options.
+ c.connect("nosuchhost0", copts().reconnect(ropts()).virtual_host("vhost0").user("user0"));
+ }
+
+ void on_transport_error(proton::transport &t) PN_CPP_OVERRIDE {
+ switch (++errors_) {
+ case 1:
+ ASSERT_SUBSTRING("nosuchhost0", t.error().what()); // First failure
+ break;
+ case 2: {
+ ASSERT_SUBSTRING("nosuchhost0",t.error().what()); // Second failure
+ std::vector<std::string> urls;
+ urls.push_back("nosuchhost1");
+ // Update the failover list
+ t.connection().update_options(copts().failover_urls(urls));
+ // Verify changing reconnect options does not affect other options.
+ ASSERT_EQUAL("user0", t.connection().user());
+ break;
+ }
+ case 3:
+ ASSERT_SUBSTRING("nosuchhost1", t.error().what()); // Using failover host
+ // Change a non-reconnect option should not affect reconnect
+ t.connection().update_options(copts().user("user1"));
+ break;
+ case 4:
+ ASSERT_SUBSTRING("nosuchhost0", t.error().what()); // Back to original url
+ ASSERT_EQUAL("user1", t.connection().user());
+ break;
+ case 5:
+ ASSERT_SUBSTRING("nosuchhost1", t.error().what()); // Still have failover
+ break;
+ default:
+ t.connection().container().stop();
+ }
+ }
+
+ void run() { container_.run(); }
+
+private:
+ int errors_;
+ proton::container container_;
+};
+
+class test_reconnect_update_simple : public proton::messaging_handler {
+public:
+ test_reconnect_update_simple()
+ : errors_(0), container_(*this, "test_reconnect_update") {}
+
+ proton::reconnect_options ropts() {
+ // Fast as we can to avoid needless test slowness.
+ return proton::reconnect_options().delay(proton::duration::MILLISECOND);
+ }
+
+ proton::connection_options copts() { return proton::connection_options(); }
+
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ // Never actually connects, keeps re-trying to bogus hostnames with
+ // changing options.
+ c.connect("nosuchhost0", copts().reconnect(ropts()).virtual_host("vhost0").user("user0"));
+ }
+
+ void on_transport_error(proton::transport &t) PN_CPP_OVERRIDE {
+ switch (++errors_) {
+ case 1:
+ ASSERT_SUBSTRING("nosuchhost0", t.error().what()); // First failure
+ break;
+ case 2: {
+ ASSERT_SUBSTRING("nosuchhost0",t.error().what()); // Second failure
+ t.connection().update_options(copts().reconnect_url("nosuchhost1"));
+ // Verify changing reconnect options does not affect other options.
+ ASSERT_EQUAL("user0", t.connection().user());
+ break;
+ }
+ case 3:
+ ASSERT_SUBSTRING("nosuchhost1", t.error().what()); // Re-try original
+ t.connection().update_options(copts().reconnect_url("notsuchahostatall"));
+ break;
+ case 4:
+ ASSERT_SUBSTRING("notsuchahostatall", t.error().what()); // Re-try new reconnect url
+ break;
+ case 5:
+ ASSERT_SUBSTRING("notsuchahostatall", t.error().what()); // Re-try new reconnect url
+ // Change a non-reconnect option should not affect reconnect
+ t.connection().update_options(copts().user("user1"));
+ break;
+ case 6:
+ ASSERT_SUBSTRING("notsuchahostatall", t.error().what()); // Same reconnect url
+ ASSERT_EQUAL("user1", t.connection().user());
+ t.connection().update_options(copts().reconnect_url("nosuchhost1"));
+ break;
+ case 7:
+ ASSERT_SUBSTRING("nosuchhost1", t.error().what());
+ break;
+ default:
+ t.connection().container().stop();
+ }
+ }
+
+ void run() { container_.run(); }
+
+private:
+ int errors_;
+ proton::container container_;
+};
+
+int main(int argc, char **argv) {
int failed = 0;
RUN_ARGV_TEST(failed, test_failover_simple());
RUN_ARGV_TEST(failed, test_stop_reconnect());
RUN_ARGV_TEST(failed, test_auth_fail_reconnect());
RUN_ARGV_TEST(failed, test_reconnecting_close().run());
+ RUN_ARGV_TEST(failed, test_reconnect_url().run());
+ RUN_ARGV_TEST(failed, test_reconnect_update_failover().run());
+ RUN_ARGV_TEST(failed, test_reconnect_update_simple().run());
return failed;
}
-
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org