qpid-dispatch git commit: DISPATCH-1107: Add router connection id interrouter Opens
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 447937f37 -> df7667e8a
DISPATCH-1107: Add router connection id interrouter Opens
This feature helps disambiguate interrouter connections.
Topology tests generate many connections and resulting logs have the
connection ids. There is nothing topology-specific about this feature.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/df7667e8
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/df7667e8
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/df7667e8
Branch: refs/heads/master
Commit: df7667e8a9fa2ad6a2dd2ca91fe6f4a45b4d4f3c
Parents: 447937f
Author: Chuck Rolke
Authored: Mon Sep 10 16:23:48 2018 -0400
Committer: Chuck Rolke
Committed: Mon Sep 10 16:23:48 2018 -0400
--
include/qpid/dispatch/amqp.h | 1 +
src/amqp.c | 1 +
src/server.c | 5
tests/system_tests_topology_disposition.py | 37 -
4 files changed, 43 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/df7667e8/include/qpid/dispatch/amqp.h
--
diff --git a/include/qpid/dispatch/amqp.h b/include/qpid/dispatch/amqp.h
index 393cf57..3820726 100644
--- a/include/qpid/dispatch/amqp.h
+++ b/include/qpid/dispatch/amqp.h
@@ -139,6 +139,7 @@ extern const char * const
QD_CONNECTION_PROPERTY_PRODUCT_KEY;
extern const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE;
extern const char * const QD_CONNECTION_PROPERTY_VERSION_KEY;
extern const char * const QD_CONNECTION_PROPERTY_COST_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_CONN_ID;
extern const char * const QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY;
extern const char * const QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY;
extern const char * const QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/df7667e8/src/amqp.c
--
diff --git a/src/amqp.c b/src/amqp.c
index fcad53a..16550a7 100644
--- a/src/amqp.c
+++ b/src/amqp.c
@@ -42,6 +42,7 @@ const char * const QD_CONNECTION_PROPERTY_PRODUCT_KEY
= "product";
const char * const QD_CONNECTION_PROPERTY_PRODUCT_VALUE =
"qpid-dispatch-router";
const char * const QD_CONNECTION_PROPERTY_VERSION_KEY = "version";
const char * const QD_CONNECTION_PROPERTY_COST_KEY =
"qd.inter-router-cost";
+const char * const QD_CONNECTION_PROPERTY_CONN_ID = "qd.conn-id";
const char * const QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY =
"failover-server-list";
const char * const QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY =
"network-host";
const char * const QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY = "port";
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/df7667e8/src/server.c
--
diff --git a/src/server.c b/src/server.c
index ab9ca92..a2bb8ea 100644
--- a/src/server.c
+++ b/src/server.c
@@ -447,6 +447,11 @@ static void decorate_connection(qd_server_t *qd_server,
pn_connection_t *conn, c
pn_data_put_string(pn_connection_properties(conn),
pn_bytes(strlen(QPID_DISPATCH_VERSION),
QPID_DISPATCH_VERSION));
+pn_data_put_symbol(pn_connection_properties(conn),
+ pn_bytes(strlen(QD_CONNECTION_PROPERTY_CONN_ID),
QD_CONNECTION_PROPERTY_CONN_ID));
+qd_connection_t *qd_conn = pn_connection_get_context(conn);
+pn_data_put_int(pn_connection_properties(conn), qd_conn->connection_id);
+
if (config && config->inter_router_cost > 1) {
pn_data_put_symbol(pn_connection_properties(conn),
pn_bytes(strlen(QD_CONNECTION_PROPERTY_COST_KEY),
QD_CONNECTION_PROPERTY_COST_KEY));
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/df7667e8/tests/system_tests_topology_disposition.py
--
diff --git a/tests/system_tests_topology_disposition.py
b/tests/system_tests_topology_disposition.py
index 1b0d922..a701c6d 100644
--- a/tests/system_tests_topology_disposition.py
+++ b/tests/system_tests_topology_disposition.py
@@ -370,7 +370,8 @@ class TopologyDispositionTests ( TestCase ):
# 1 means skip that test.
cls.skip = { 'test_01' : 0,
- 'test_02' : 0
+ 'test_02' : 0,
+ 'test_03' : 0
}
@@ -399,6 +400,40 @@ class TopologyDispositionTests ( TestCase ):
self.assertEqual ( None, test.error )
+def test_03_connection_id_propagation
[2/6] qpid-proton git commit: PROTON-1798: [c] Fix benign race in broker.c example found by tsan
PROTON-1798: [c] Fix benign race in broker.c example found by tsan
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/c6db6358
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/c6db6358
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/c6db6358
Branch: refs/heads/master
Commit: c6db635838f0abb67eb37bf565d4072870e1fe9d
Parents: 407711a
Author: Alan Conway
Authored: Tue Sep 4 19:32:59 2018 -0400
Committer: Alan Conway
Committed: Fri Sep 7 11:26:25 2018 -0400
--
c/examples/broker.c | 16
1 file changed, 8 insertions(+), 8 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/c6db6358/c/examples/broker.c
--
diff --git a/c/examples/broker.c b/c/examples/broker.c
index 69dc536..6ffe8ed 100644
--- a/c/examples/broker.c
+++ b/c/examples/broker.c
@@ -216,7 +216,6 @@ typedef struct broker_t {
size_t threads;
const char *container_id; /* AMQP container-id */
queues_t queues;
- bool finished;
pn_ssl_domain_t *ssl_domain;
} broker_t;
@@ -276,14 +275,14 @@ static void session_unsub(broker_t *b, pn_session_t *ssn)
{
static void check_condition(pn_event_t *e, pn_condition_t *cond) {
if (pn_condition_is_set(cond)) {
-fprintf(stderr, "%s: %s: %s\n", pn_event_type_name(pn_event_type(e)),
-pn_condition_get_name(cond), pn_condition_get_description(cond));
+printf("%s: %s: %s\n", pn_event_type_name(pn_event_type(e)),
+ pn_condition_get_name(cond), pn_condition_get_description(cond));
}
}
const int WINDOW=5; /* Very small incoming credit window, to show flow control
in action */
-static void handle(broker_t* b, pn_event_t* e) {
+static bool handle(broker_t* b, pn_event_t* e) {
pn_connection_t *c = pn_event_connection(e);
switch (pn_event_type(e)) {
@@ -418,25 +417,26 @@ static void handle(broker_t* b, pn_event_t* e) {
break;
case PN_PROACTOR_INTERRUPT:
-b->finished = true;
pn_proactor_interrupt(b->proactor); /* Pass along the interrupt to the
other threads */
-break;
+return false;
default:
break;
}
+ return true;
}
static void* broker_thread(void *void_broker) {
broker_t *b = (broker_t*)void_broker;
+ bool finished = false;
do {
pn_event_batch_t *events = pn_proactor_wait(b->proactor);
pn_event_t *e;
while ((e = pn_event_batch_next(events))) {
- handle(b, e);
+if (!handle(b, e)) finished = true;
}
pn_proactor_done(b->proactor, events);
- } while(!b->finished);
+ } while(!finished);
return NULL;
}
-
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
[4/6] qpid-proton git commit: PROTON-1798: cmake runtime-check improvements
PROTON-1798: cmake runtime-check improvements
Usage notes:
- new CMake variable: RUNTIME_CHECK, choose from [memcheck helgrind asan tsan
OFF]
- defaults to 'memcheck' if available, else OFF
- old ENABLE_ variables for valgrind/sanitizers are deprecated
- example_test scripts check for stderr output including from killed processes
Implementation details:
- moved all runtime-check setup code to seprate runtime-check.cmake
- tool-agnostic internal CMake variables for running tests
- removed all valgrind-specific code outside of runtime-check.cmake
- example_test.py check stderr as well as exit status to catch broker issues.
NOTE: asan,tsan not yet working for python/ruby bindings, they are disabled in
san builds. See tests/preload_asan.sh for current status of the work.
NOTE: Some python soak tests for obscure messenger features were removed, they
have faulty start-up timing logic and can fail under valgrind. We can restore
them if needed but we'll need to fix the -X feature of msgr-recv to report ready
only after connections are remote open.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/27edd9ac
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/27edd9ac
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/27edd9ac
Branch: refs/heads/master
Commit: 27edd9aca3b2b1078089ceaa2a387f0016ac6f0a
Parents: 7885bd3
Author: Alan Conway
Authored: Fri Sep 7 13:20:42 2018 -0400
Committer: Alan Conway
Committed: Mon Sep 10 15:41:24 2018 -0400
--
CMakeLists.txt | 71 +++
c/examples/CMakeLists.txt | 7 +-
c/examples/example_test.py | 58 +---
c/tests/CMakeLists.txt | 4 +-
c/tests/fuzz/CMakeLists.txt | 2 +-
c/tests/threaderciser.supp | 18
c/tests/threaderciser.tsupp | 5 -
cpp/CMakeLists.txt | 2 +-
cpp/examples/CMakeLists.txt | 14 ++-
cpp/examples/broker.cpp | 4 +-
cpp/examples/example_test.py| 42 ++---
misc/config.sh.in | 6 --
python/CMakeLists.txt | 8 +-
python/tests/proton_tests/common.py | 35 +--
python/tests/proton_tests/soak.py | 83 ++---
python/tests/proton_tests/ssl.py| 33 ---
python/tox.ini.in | 2 +-
ruby/CMakeLists.txt | 8 +-
runtime_check.cmake | 123 +
scripts/env.py | 5 -
tests/preload_asan.sh | 51 +++
tests/py/test_subprocess.py | 105 +
tests/valgrind.supp | 151 +++
23 files changed, 517 insertions(+), 320 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/27edd9ac/CMakeLists.txt
--
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 560dc05..105f22e 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -38,6 +38,9 @@ find_package (CyrusSASL)
enable_testing ()
+# Set up runtime checks (valgrind, sanitizers etc.)
+include(runtime_check.cmake)
+
## Variables used across components
set (PN_ENV_SCRIPT "${PYTHON_EXECUTABLE}" "${CMAKE_SOURCE_DIR}/scripts/env.py")
@@ -188,23 +191,6 @@ if (CMAKE_C_COMPILER_ID MATCHES "Clang")
set (CXX_WARNING_FLAGS "${COMPILE_WARNING_FLAGS} -Wno-c++98-compat
-Wno-c++98-compat-pedantic -Wno-float-equal -Wno-padded -Wno-sign-conversion
-Wno-switch-enum -Wno-weak-vtables -Wno-exit-time-destructors
-Wno-global-constructors -Wno-shorten-64-to-32 -Wno-documentation
-Wno-documentation-unknown-command -Wno-old-style-cast -Wno-missing-noreturn")
endif()
-# Sanitizer flags apply to to both GNU and clang, C and C++
-if(ENABLE_SANITIZERS)
- set(SANITIZE_FLAGS "-g -fno-omit-frame-pointer -fsanitize=address
-fsanitize=leak -fsanitize=undefined")
-endif()
-if(ENABLE_TSAN)
- set(SANITIZE_FLAGS "-g -fno-omit-frame-pointer -fsanitize=thread")
-endif()
-if (SANITIZE_FLAGS)
- mark_as_advanced(SANITIZE_FLAGS)
- if(CMAKE_C_COMPILER_ID MATCHES "Clang|GNU")
-set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${SANITIZE_FLAGS}")
- endif()
- if(CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU")
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${SANITIZE_FLAGS}")
- endif()
-endif()
-
if (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
if (NOT CMAKE_OSX_ARCHITECTURES)
set(CMAKE_OSX_ARCHITECTURES "x86_64")
@@ -300,41 +286,6 @@ endforeach()
set (PROTON_SHARE ${SHARE_INSTALL_DIR}/proton-${PN_VERSION})
# End of variables used during install
-# Can't do valgrind and coverage at athe same time - coverage takes precedence
-if (CMAKE_BUILD_TYPE MATCHES "Coverage")
- message(STATUS "Building for coverage analysis; no run-time error
[1/6] qpid-proton git commit: PROTON-1798: [cpp] add library destructors for main classes
Repository: qpid-proton
Updated Branches:
refs/heads/master 407711aac -> 6c765bc66
PROTON-1798: [cpp] add library destructors for main classes
Add library destructors to anchor vtables and typeinfo for connection, session,
sender, recever and delivery.
Without them, the ubsan sanitizer reports mismatched types due to different
weak vtable symbols in scope at library and executable link time.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/7885bd3b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/7885bd3b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/7885bd3b
Branch: refs/heads/master
Commit: 7885bd3b558dcf801e92ea866090d1110cc36aa1
Parents: e5aac00
Author: Alan Conway
Authored: Wed Sep 5 14:12:20 2018 -0400
Committer: Alan Conway
Committed: Fri Sep 7 11:26:25 2018 -0400
--
cpp/include/proton/connection.hpp | 2 ++
cpp/include/proton/delivery.hpp | 2 ++
cpp/include/proton/receiver.hpp | 2 ++
cpp/include/proton/sender.hpp | 2 ++
cpp/include/proton/session.hpp| 2 ++
cpp/src/connection.cpp| 2 ++
cpp/src/delivery.cpp | 1 +
cpp/src/receiver.cpp | 2 ++
cpp/src/sender.cpp| 2 ++
cpp/src/session.cpp | 2 ++
10 files changed, 19 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/include/proton/connection.hpp
--
diff --git a/cpp/include/proton/connection.hpp
b/cpp/include/proton/connection.hpp
index 77bf96d..de9c904 100644
--- a/cpp/include/proton/connection.hpp
+++ b/cpp/include/proton/connection.hpp
@@ -51,6 +51,8 @@ PN_CPP_CLASS_EXTERN connection : public
internal::object, publi
/// Create an empty connection.
connection() : internal::object(0) {}
+PN_CPP_EXTERN ~connection();
+
PN_CPP_EXTERN bool uninitialized() const;
PN_CPP_EXTERN bool active() const;
PN_CPP_EXTERN bool closed() const;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/include/proton/delivery.hpp
--
diff --git a/cpp/include/proton/delivery.hpp b/cpp/include/proton/delivery.hpp
index 61a2fbe..c1823d9 100644
--- a/cpp/include/proton/delivery.hpp
+++ b/cpp/include/proton/delivery.hpp
@@ -44,6 +44,8 @@ class delivery : public transfer {
public:
delivery() {}
+PN_CPP_EXTERN ~delivery();
+
/// Return the receiver for this delivery.
PN_CPP_EXTERN class receiver receiver() const;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/include/proton/receiver.hpp
--
diff --git a/cpp/include/proton/receiver.hpp b/cpp/include/proton/receiver.hpp
index 6c8137d..7f78401 100644
--- a/cpp/include/proton/receiver.hpp
+++ b/cpp/include/proton/receiver.hpp
@@ -47,6 +47,8 @@ PN_CPP_CLASS_EXTERN receiver : public link {
/// Create an empty receiver.
receiver() {}
+PN_CPP_EXTERN ~receiver();
+
/// Open the receiver.
///
/// @see endpoint_lifecycle
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/include/proton/sender.hpp
--
diff --git a/cpp/include/proton/sender.hpp b/cpp/include/proton/sender.hpp
index 840032c..5f7c7b5 100644
--- a/cpp/include/proton/sender.hpp
+++ b/cpp/include/proton/sender.hpp
@@ -46,6 +46,8 @@ PN_CPP_CLASS_EXTERN sender : public link {
/// Create an empty sender.
sender() {}
+PN_CPP_EXTERN ~sender();
+
/// Open the sender.
///
/// @see endpoint_lifecycle
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/include/proton/session.hpp
--
diff --git a/cpp/include/proton/session.hpp b/cpp/include/proton/session.hpp
index 78a1fd4..7f289f5 100644
--- a/cpp/include/proton/session.hpp
+++ b/cpp/include/proton/session.hpp
@@ -49,6 +49,8 @@ PN_CPP_CLASS_EXTERN session : public
internal::object, public endp
/// Create an empty session.
session() : internal::object(0) {}
+PN_CPP_EXTERN ~session();
+
PN_CPP_EXTERN bool uninitialized() const;
PN_CPP_EXTERN bool active() const;
PN_CPP_EXTERN bool closed() const;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/7885bd3b/cpp/src/connection.cpp
--
diff --git a/cpp/src/connection.cpp b/cpp/src/connection.cpp
index b2dadae..223bb0c 100644
--- a/cpp/src/connection.cpp
+++ b/cpp/src/connection.cpp
@@ -45,6 +45,8 @@
namespace proton {
+connection::~connection() {}
+
transport connection::transport()
[5/6] qpid-proton git commit: PROTON-1930: [cpp] Fix race condition in container_test.cpp
PROTON-1930: [cpp] Fix race condition in container_test.cpp
Test was starting container, opening connection and then checking for
["start", "open"] sequence to be set by handlers.
Sometimes the sequence was instead ["open", "start"], which is legal since the
events are generated in different handler contexts.
Fixed by starting container, waiting for "start", opening connection, then
waiting for "open"
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/d3c11354
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/d3c11354
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/d3c11354
Branch: refs/heads/master
Commit: d3c113548146b56cdd8b2a31c2d4dd7ba5dddbfd
Parents: 27edd9a
Author: Alan Conway
Authored: Mon Sep 10 12:35:52 2018 -0400
Committer: Alan Conway
Committed: Mon Sep 10 15:50:37 2018 -0400
--
cpp/src/container_test.cpp | 28 +++-
1 file changed, 15 insertions(+), 13 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d3c11354/cpp/src/container_test.cpp
--
diff --git a/cpp/src/container_test.cpp b/cpp/src/container_test.cpp
index 6f7caa4..0e74aaa 100644
--- a/cpp/src/container_test.cpp
+++ b/cpp/src/container_test.cpp
@@ -366,17 +366,18 @@ void test_container_mt_stop_empty() {
c.auto_stop( false );
container_runner runner(c);
auto t = std::thread(runner);
-// Must ensure that thread is joined or detached
+// Must ensure that thread is joined
try {
ASSERT_EQUAL("start", th.wait());
c.stop();
t.join();
ASSERT_EQUAL("", th.error().name());
-} catch (...) {
-// We don't join as we don't know if we'll be stuck waiting
-if (t.joinable()) {
-t.detach();
-}
+} catch (const std::exception ) {
+std::cerr << FAIL_MSG(e.what()) << std::endl;
+// If join hangs, let the test die by timeout. We cannot
+// detach and continue: deleting the container while t is
+// still alive will put the process in an undefined state.
+t.join();
throw;
}
}
@@ -387,19 +388,20 @@ void test_container_mt_stop() {
c.auto_stop(false);
container_runner runner(c);
auto t = std::thread(runner);
-// Must ensure that thread is joined or detached
+// Must ensure that thread is joined
try {
test_listen_handler lh;
-c.listen("//:0", lh); // Also opens a connection
ASSERT_EQUAL("start", th.wait());
+c.listen("//:0", lh); // Also opens a connection
ASSERT_EQUAL("open", th.wait());
c.stop();
t.join();
-} catch (...) {
-// We don't join as we don't know if we'll be stuck waiting
-if (t.joinable()) {
-t.detach();
-}
+} catch (const std::exception& e) {
+std::cerr << FAIL_MSG(e.what()) << std::endl;
+// If join hangs, let the test die by timeout. We cannot
+// detach and continue: deleting the container while t is
+// still alive will put the process in an undefined state.
+t.join();
throw;
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
[6/6] qpid-proton git commit: PROTON-1929: [c] library prints directly to stderr/stdout
PROTON-1929: [c] library prints directly to stderr/stdout
Replace direct use of stdout with pn_log calls.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/6c765bc6
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/6c765bc6
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/6c765bc6
Branch: refs/heads/master
Commit: 6c765bc66461c17b02ce740804f4d9e171f0fe24
Parents: d3c1135
Author: Alan Conway
Authored: Mon Sep 10 15:54:57 2018 -0400
Committer: Alan Conway
Committed: Mon Sep 10 15:54:57 2018 -0400
--
c/src/core/log.c| 1 +
c/src/core/transport.c | 1 +
c/src/core/util.c | 22 +++
c/src/core/util.h | 3 +-
c/src/sasl/cyrus_sasl.c | 4 +-
c/src/ssl/openssl.c | 27 -
c/src/ssl/schannel.c| 133 +--
7 files changed, 87 insertions(+), 104 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/6c765bc6/c/src/core/log.c
--
diff --git a/c/src/core/log.c b/c/src/core/log.c
index 754eed3..ed04573 100644
--- a/c/src/core/log.c
+++ b/c/src/core/log.c
@@ -51,6 +51,7 @@ void pn_log_logger(pn_logger_t new_logger) {
void pni_vlogf_impl(const char *fmt, va_list ap) {
vfprintf(stderr, fmt, ap);
fprintf(stderr, "\n");
+fflush(stderr);
}
/**@internal
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/6c765bc6/c/src/core/transport.c
--
diff --git a/c/src/core/transport.c b/c/src/core/transport.c
index 106a2c8..7dee571 100644
--- a/c/src/core/transport.c
+++ b/c/src/core/transport.c
@@ -135,6 +135,7 @@ static void pni_delivery_map_clear(pn_delivery_map_t *dm)
static void pni_default_tracer(pn_transport_t *transport, const char *message)
{
fprintf(stderr, "[%p]:%s\n", (void *) transport, message);
+ fflush(stderr);
}
static ssize_t pn_io_layer_input_passthru(pn_transport_t *, unsigned int,
const char *, size_t );
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/6c765bc6/c/src/core/util.c
--
diff --git a/c/src/core/util.c b/c/src/core/util.c
index a676e9f..2597d60 100644
--- a/c/src/core/util.c
+++ b/c/src/core/util.c
@@ -19,9 +19,9 @@
*
*/
-#include "util.h"
-
#include "buffer.h"
+#include "log_private.h"
+#include "util.h"
#include
#include
@@ -82,27 +82,19 @@ int pn_quote(pn_string_t *dst, const char *src, size_t size)
}
}
-void pn_fprint_data(FILE *stream, const char *bytes, size_t size)
+void pn_log_data(const char *msg, const char *bytes, size_t size)
{
char buf[256];
ssize_t n = pn_quote_data(buf, 256, bytes, size);
if (n >= 0) {
-fputs(buf, stream);
+pn_logf("%s: %s", msg, buf);
+ } else if (n == PN_OVERFLOW) {
+pn_logf("%s: %s (truncated)", msg, buf);
} else {
-if (n == PN_OVERFLOW) {
- fputs(buf, stream);
- fputs("... (truncated)", stream);
-}
-else
- fprintf(stderr, "pn_quote_data: %s\n", pn_code(n));
+pn_logf("%s: cannot log data: %s", msg, pn_code(n));
}
}
-void pn_print_data(const char *bytes, size_t size)
-{
- pn_fprint_data(stdout, bytes, size);
-}
-
int pn_strcasecmp(const char *a, const char *b)
{
int diff;
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/6c765bc6/c/src/core/util.h
--
diff --git a/c/src/core/util.h b/c/src/core/util.h
index 78b1c4d..559fbaa 100644
--- a/c/src/core/util.h
+++ b/c/src/core/util.h
@@ -36,8 +36,7 @@
ssize_t pn_quote_data(char *dst, size_t capacity, const char *src, size_t
size);
int pn_quote(pn_string_t *dst, const char *src, size_t size);
-void pn_fprint_data(FILE *stream, const char *bytes, size_t size);
-void pn_print_data(const char *bytes, size_t size);
+void pn_log_data(const char *msg, const char *bytes, size_t size);
bool pn_env_bool(const char *name);
pn_timestamp_t pn_timestamp_min(pn_timestamp_t a, pn_timestamp_t b);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/6c765bc6/c/src/sasl/cyrus_sasl.c
--
diff --git a/c/src/sasl/cyrus_sasl.c b/c/src/sasl/cyrus_sasl.c
index a7387e6..30c5784 100644
--- a/c/src/sasl/cyrus_sasl.c
+++ b/c/src/sasl/cyrus_sasl.c
@@ -22,6 +22,8 @@
#define _GNU_SOURCE
#endif
+#include "core/log_private.h"
+
#include "proton/sasl.h"
#include "proton/sasl-plugin.h"
#include "proton/transport.h"
@@ -130,7 +132,7 @@ static void pni_cyrus_interact(pn_transport_t *transport,
sasl_interact_t *inter
break;
}
default:
- fprintf(stderr, "(%s): %s - %s\n", i->challenge, i->prompt,
i->defresult);
[3/6] qpid-proton git commit: PROTON-1798: [cpp] C++ broker example, remove unused shutdown code.
PROTON-1798: [cpp] C++ broker example, remove unused shutdown code.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/e5aac008
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/e5aac008
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/e5aac008
Branch: refs/heads/master
Commit: e5aac0083bb1cc2a6e96a361ff6031066e4dc449
Parents: c6db635
Author: Alan Conway
Authored: Wed Sep 5 13:56:46 2018 -0400
Committer: Alan Conway
Committed: Fri Sep 7 11:26:25 2018 -0400
--
c/examples/example_test.py | 4 ++--
cpp/examples/broker.cpp| 14 ++
2 files changed, 4 insertions(+), 14 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/e5aac008/c/examples/example_test.py
--
diff --git a/c/examples/example_test.py b/c/examples/example_test.py
index 5d6322f..1834989 100644
--- a/c/examples/example_test.py
+++ b/c/examples/example_test.py
@@ -117,8 +117,8 @@ class ExampleTest(unittest.TestCase):
self.assertIn("secure connection:", got)
self.assertIn(send_expect(), got)
self.assertMultiLineEqual(receive_expect(),
self.runex("receive", b.port))
-except subprocess.CalledProcessError as e:
-if e.output.startswith(b"error initializing SSL"):
+except TestProcessError as e:
+if e.output and e.output.startswith(b"error initializing SSL"):
print("Skipping %s: SSL not available" % self.id())
else:
raise
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/e5aac008/cpp/examples/broker.cpp
--
diff --git a/cpp/examples/broker.cpp b/cpp/examples/broker.cpp
index 479ec01..d45309e 100644
--- a/cpp/examples/broker.cpp
+++ b/cpp/examples/broker.cpp
@@ -337,18 +337,8 @@ public:
// A receiver receives messages from a publisher to a queue.
void on_receiver_open(proton::receiver ) OVERRIDE {
std::string qname = receiver.target().address();
-if (qname == "shutdown") {
-std::cout << "broker shutting down" << std::endl;
-// Sending to the special "shutdown" queue stops the broker.
-receiver.connection().container().stop(
-proton::error_condition("shutdown", "stop broker"));
-} else {
-if (qname.empty()) {
-DOUT(std::cerr << "ODD - trying to attach to a empty
address\n";);
-}
-Receiver* r = new Receiver(receiver);
-queue_manager_.add(make_work(::findQueueReceiver,
_manager_, r, qname));
-}
+Receiver* r = new Receiver(receiver);
+queue_manager_.add(make_work(::findQueueReceiver,
_manager_, r, qname));
}
void on_session_close(proton::session ) OVERRIDE {
-
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
qpid-dispatch git commit: DISPATCH-1118 - Code cleanup to remove valgrind errors.
Repository: qpid-dispatch Updated Branches: refs/heads/master 710b7059b -> 447937f37 DISPATCH-1118 - Code cleanup to remove valgrind errors. Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/447937f3 Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/447937f3 Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/447937f3 Branch: refs/heads/master Commit: 447937f374ec69802faf80fd4f0945f21076a287 Parents: 710b705 Author: Ted Ross Authored: Mon Sep 10 15:59:08 2018 -0400 Committer: Ted Ross Committed: Mon Sep 10 15:59:08 2018 -0400 -- src/router_core/route_control.c | 13 - 1 file changed, 4 insertions(+), 9 deletions(-) -- http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/447937f3/src/router_core/route_control.c -- diff --git a/src/router_core/route_control.c b/src/router_core/route_control.c index 6e5a16d..1e2d81b 100644 --- a/src/router_core/route_control.c +++ b/src/router_core/route_control.c @@ -418,17 +418,12 @@ void qdr_route_auto_link_detached_CT(qdr_core_t *core, qdr_link_t *link) if (!link->auto_link->retry_timer) link->auto_link->retry_timer = qdr_core_timer_CT(core, qdr_route_attempt_auto_link_CT, (void *)link->auto_link); -char *activation_failed = "Auto Link Activation Failed. "; -int error_length = 0; -if (link->auto_link->last_error) -error_length = strlen(link->auto_link->last_error); -int total_length = strlen(activation_failed) + 1; -if (error_length) -total_length += error_length; +static char *activation_failed = "Auto Link Activation Failed. "; +int error_length = link->auto_link->last_error ? strlen(link->auto_link->last_error) : 0; +int total_length = strlen(activation_failed) + error_length + 1; char error_msg[total_length]; -memset(error_msg, 0, error_length); -strcat(error_msg, activation_failed); +strcpy(error_msg, activation_failed); if (error_length) strcat(error_msg, link->auto_link->last_error); - To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org
qpid-dispatch git commit: DISPATCH-1096 - priority messaging support
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 03ba19e65 -> 710b7059b
DISPATCH-1096 - priority messaging support
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/710b7059
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/710b7059
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/710b7059
Branch: refs/heads/master
Commit: 710b7059b77a72337a25c79b1e4e01856ca484b7
Parents: 03ba19e
Author: Michael Goulish
Authored: Mon Sep 10 11:53:37 2018 -0400
Committer: Ted Ross
Committed: Mon Sep 10 14:19:25 2018 -0400
--
include/qpid/dispatch/iterator.h | 2 +
include/qpid/dispatch/message.h | 7 +
src/iterator.c| 8 ++
src/message.c | 90 +---
src/message_private.h | 11 ++
src/router_core/connections.c | 216 -
src/router_core/forwarder.c | 52 +--
src/router_core/route_tables.c| 7 +-
src/router_core/router_core_private.h | 14 +-
src/router_core/transfer.c| 15 +-
10 files changed, 287 insertions(+), 135 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/710b7059/include/qpid/dispatch/iterator.h
--
diff --git a/include/qpid/dispatch/iterator.h b/include/qpid/dispatch/iterator.h
index fa4267c..74d73ca 100644
--- a/include/qpid/dispatch/iterator.h
+++ b/include/qpid/dispatch/iterator.h
@@ -327,6 +327,8 @@ int qd_iterator_ncopy(qd_iterator_t *iter, unsigned char*
buffer, int n);
*/
unsigned char *qd_iterator_copy(qd_iterator_t *iter);
+uint8_t qd_iterator_uint8(qd_iterator_t *iter);
+
/**
* Return a new iterator that is a duplicate of the original iterator,
referring
* to the same base data. If the input iterator pointer is NULL, the duplicate
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/710b7059/include/qpid/dispatch/message.h
--
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index ca2ab47..ec0b901 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -411,6 +411,13 @@ bool qd_message_aborted(const qd_message_t *msg);
*/
void qd_message_set_aborted(const qd_message_t *msg, bool aborted);
+/**
+ * Return message priority
+ * @param msg A pointer to the message
+ */
+uint8_t qd_message_get_priority(qd_message_t *msg);
+
+
///@}
#endif
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/710b7059/src/iterator.c
--
diff --git a/src/iterator.c b/src/iterator.c
index 8889b17..910aa19 100644
--- a/src/iterator.c
+++ b/src/iterator.c
@@ -768,6 +768,14 @@ char* qd_iterator_strncpy(qd_iterator_t *iter, char*
buffer, int n)
}
+uint8_t qd_iterator_uint8(qd_iterator_t *iter ) {
+qd_iterator_reset(iter);
+if (qd_iterator_end(iter))
+return 0;
+return (uint8_t) qd_iterator_octet(iter);
+}
+
+
unsigned char *qd_iterator_copy(qd_iterator_t *iter)
{
if (!iter)
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/710b7059/src/message.c
--
diff --git a/src/message.c b/src/message.c
index fce3394..35363cf 100644
--- a/src/message.c
+++ b/src/message.c
@@ -724,22 +724,57 @@ static qd_field_location_t
*qd_message_properties_field(qd_message_t *msg, qd_me
// else 0)
static qd_field_location_t *qd_message_header_field(qd_message_t *msg,
qd_message_field_t field)
{
-qd_message_content_t *content = MSG_CONTENT(msg);
-
-if (!content->section_message_header.parsed) {
-if (!qd_message_check(msg, QD_DEPTH_HEADER) ||
!content->section_message_header.parsed)
-return 0;
-}
-
-switch (field) {
-case QD_FIELD_HEADER:
-return >section_message_properties;
-default:
-// TBD: add header fields as needed (see qd_message_properties_field()
-// as an example)
-assert(false);
+int first_header_field = QD_FIELD_DURABLE,
+last_header_field = QD_FIELD_DELIVERY_COUNT;
+static const intptr_t offsets[] = {
+// position of the fields' qd_field_location_t in the message content
object
+(intptr_t) &((qd_message_content_t *)0)->field_durable,
+(intptr_t) &((qd_message_content_t *)0)->field_priority,
+(intptr_t) &((qd_message_content_t *)0)->field_ttl,
+(intptr_t) &((qd_message_content_t *)0)->field_first_acquirer,
+(intptr_t) &((qd_message_content_t *)0)->field_delivery_count
+};
+if (!(first_header_field <= field && field <= last_header_field)) {
+
qpid-dispatch git commit: DISPATCH-1090 - Augmented drain test to reopen receiver and making sure the drained sender was re-issued credit this being able to send to the receiver
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 0716719b9 -> 03ba19e65
DISPATCH-1090 - Augmented drain test to reopen receiver and making sure the
drained sender was re-issued credit this being able to send to the receiver
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/03ba19e6
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/03ba19e6
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/03ba19e6
Branch: refs/heads/master
Commit: 03ba19e6534a7779970c59c4c9fb2f9ee6e3a2c8
Parents: 0716719
Author: Ganesh Murthy
Authored: Mon Sep 10 13:46:03 2018 -0400
Committer: Ganesh Murthy
Committed: Mon Sep 10 13:46:03 2018 -0400
--
tests/system_tests_drain.py | 47 +++-
1 file changed, 42 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/03ba19e6/tests/system_tests_drain.py
--
diff --git a/tests/system_tests_drain.py b/tests/system_tests_drain.py
index bdc56ed..da4f8e9 100644
--- a/tests/system_tests_drain.py
+++ b/tests/system_tests_drain.py
@@ -147,6 +147,10 @@ class ReceiverDropsOffSenderDrain(MessagingHandler):
self.receiver_closed = False
self.drained = 0
self.expected_drained = 249
+self.sender_drained = False
+
+# Second receiver link opened.
+self.sec_recv_link_opened = False
def timeout(self):
if not self.error:
@@ -175,8 +179,31 @@ class ReceiverDropsOffSenderDrain(MessagingHandler):
def on_message(self, event):
# As soon as the receiver receives the message, close the receiver
if event.receiver == self.receiver:
-self.receiver.close()
-self.receiver_closed = True
+if self.sec_recv_link_opened and self.sender_drained:
+
+# Make sure this is the same message body that was
+# sent by the newly created receiver
+if event.message.body[u'number'] == 3:
+self.receiver.close()
+self.error = None
+self.sender_conn.close()
+self.receiver_conn.close()
+else:
+self.receiver.close()
+self.receiver_closed = True
+
+def on_link_opened(self, event):
+if self.sender_drained:
+if event.receiver == self.receiver:
+self.sec_recv_link_opened = True
+
+if self.num_msgs < 3:
+# Send a message after the sender has been drained
+# and a new receiver has been created
+# and make sure that the message has reached the receiver
+self.num_msgs += 1
+msg = Message(body={'number': 3})
+self.sender.send(msg)
def on_link_closed(self, event):
if event.receiver == self.receiver:
@@ -185,6 +212,7 @@ class ReceiverDropsOffSenderDrain(MessagingHandler):
# this point. The router will receive this message and see that
# there are no receivers and it will send a drain to the sender
# This test will not work without the fix for DISPATCH-1090
+self.num_msgs += 1
msg = Message(body={'number': 2})
self.sender.send(msg)
@@ -193,9 +221,18 @@ class ReceiverDropsOffSenderDrain(MessagingHandler):
if event.sender:
self.drained = event.sender.drained()
if self.drained == self.expected_drained:
-self.error = None
-self.sender_conn.close()
-self.receiver_conn.close()
+# The sender has been drained. Now create another receiver
+# to the same address as the sender
+# and use the sender to send a message to see
+# if flow is re-issued by the router to the sender and if
+# the message reaches this newly created receiver
+self.sender_drained = True
+
+# Create a new receiver
+self.receiver = event.container.create_receiver(
+self.receiver_conn,
+self.dest,
+name="A")
def run(self):
Container(self).run()
-
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
qpid-dispatch git commit: NO-JIRA - Modified the version of Proton required from 0.23 to 0.25. This is because of some changes made to the Proton python binding which led to commit 6e03ab8a5dc5059bb46
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 0a8d43285 -> 0716719b9
NO-JIRA - Modified the version of Proton required from 0.23 to 0.25. This is
because of some changes made to the Proton python binding which led to commit
6e03ab8a5dc5059bb4645a0b608ded93a02ebc17
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/0716719b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/0716719b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/0716719b
Branch: refs/heads/master
Commit: 0716719b9b87cc56676a87047da6cc50275cb169
Parents: 0a8d432
Author: Ganesh Murthy
Authored: Mon Sep 10 10:12:25 2018 -0400
Committer: Ganesh Murthy
Committed: Mon Sep 10 10:12:25 2018 -0400
--
CMakeLists.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/0716719b/CMakeLists.txt
--
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 9dc7180..ecdabef 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -124,7 +124,7 @@ find_library(pthread_lib pthread)
find_library(dl_lib dl)
find_library(rt_lib rt)
-find_package(Proton 0.23 REQUIRED)
+find_package(Proton 0.25 REQUIRED)
message(STATUS "Found Proton: ${Proton_LIBRARIES} (found version
\"${Proton_VERSION}\")" )
-
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
qpid-dispatch git commit: DISPATCH-1090 - Drain excess credit if there are no receivers. This closes #371.
Repository: qpid-dispatch
Updated Branches:
refs/heads/master 06bfb5414 -> 0a8d43285
DISPATCH-1090 - Drain excess credit if there are no receivers. This closes #371.
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/0a8d4328
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/0a8d4328
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/0a8d4328
Branch: refs/heads/master
Commit: 0a8d4328580f805d490be32d7b8af7cbcc0e4d64
Parents: 06bfb54
Author: Ganesh Murthy
Authored: Tue Aug 28 15:09:16 2018 -0400
Committer: Ganesh Murthy
Committed: Mon Sep 10 09:48:55 2018 -0400
--
src/router_core/transfer.c | 8 +++-
tests/system_tests_drain.py | 94 +++-
2 files changed, 99 insertions(+), 3 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/0a8d4328/src/router_core/transfer.c
--
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index 92b320c..2031b0d 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -835,8 +835,13 @@ static void qdr_link_forward_CT(qdr_core_t *core,
qdr_link_t *link, qdr_delivery
} else {
if (more)
dlv->disposition = PN_RELEASED;
-else
+else {
qdr_delivery_release_CT(core, dlv);
+//
+// Drain credit on the link
+//
+qdr_link_issue_credit_CT(core, link, 0, true);
+}
}
if (qdr_is_addr_treatment_multicast(link->owning_addr))
@@ -949,7 +954,6 @@ static void qdr_link_forward_CT(qdr_core_t *core,
qdr_link_t *link, qdr_delivery
//
// Again, don't bother decrementing then incrementing the ref_count
//
-
DEQ_INSERT_TAIL(link->unsettled, dlv);
dlv->where = QDR_DELIVERY_IN_UNSETTLED;
qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer: dlv:%lx
qdr_link_forward_CT: action-list -> unsettled-list", (long) dlv);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/0a8d4328/tests/system_tests_drain.py
--
diff --git a/tests/system_tests_drain.py b/tests/system_tests_drain.py
index 1b38aba..bdc56ed 100644
--- a/tests/system_tests_drain.py
+++ b/tests/system_tests_drain.py
@@ -27,7 +27,9 @@ from system_test import TestCase, Qdrouterd, main_module
from system_tests_drain_support import DrainMessagesHandler,
DrainOneMessageHandler
from system_tests_drain_support import DrainNoMessagesHandler,
DrainNoMoreMessagesHandler
from system_tests_drain_support import DrainMessagesMoreHandler
-
+from proton.handlers import MessagingHandler
+from proton import Message
+from proton.reactor import Container
from time import sleep
class DrainSupportTest(TestCase):
@@ -109,5 +111,95 @@ class DrainSupportTest(TestCase):
self.assertEqual(drain_support.error, None)
+class ReceiverDropsOffDrainTest(TestCase):
+
+@classmethod
+def setUpClass(cls):
+super(ReceiverDropsOffDrainTest, cls).setUpClass()
+config = Qdrouterd.Config([
+('router', {'mode': 'standalone', 'id': 'Broker'}),
+('listener', {'role': 'normal',
+ 'port': cls.tester.get_port(),
+ 'saslMechanisms': 'ANONYMOUS'}),
+])
+
+cls.router = cls.tester.qdrouterd("A", config, wait=True)
+cls.address = cls.router.addresses[0]
+
+def test_receiver_drops_off_sender_receives_drain(self):
+test = ReceiverDropsOffSenderDrain(self.address, "examples")
+test.run()
+self.assertEqual(None, test.error)
+
+
+class ReceiverDropsOffSenderDrain(MessagingHandler):
+def __init__(self, address, dest):
+super(ReceiverDropsOffSenderDrain, self).__init__()
+self.sender_conn = None
+self.receiver_conn = None
+self.sender = None
+self.receiver = None
+self.error = None
+self.sender_drained = False
+self.address = address
+self.dest = dest
+self.num_msgs = 0
+self.receiver_closed = False
+self.drained = 0
+self.expected_drained = 249
+
+def timeout(self):
+if not self.error:
+self.error = "Timeout Expired: Sender was not drained. Expected " \
+ "drained=%s, actual drain=%s " % \
+ (self.expected_drained, self.drained)
+self.sender_conn.close()
+self.receiver_conn.close()
+
+def on_start(self, event):
+# Create sender and receiver in two separate connections
+self.sender_conn =
