You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2017/10/12 02:13:53 UTC
[1/5] qpid-proton git commit: PROTON-1618: c++ tests use test_port.h
for listen ports
Repository: qpid-proton
Updated Branches:
refs/heads/master 1bb1897fe -> 485cdbd3f
PROTON-1618: c++ tests use test_port.h for listen ports
POSIX: Use bind(0) with SO_REUSEADDR and hold the socket to acquire a port that
can safely be used for listen()
Windows: Use bind(0) to pick a port, but close the socket immediately. In theory
another process could steal the port between bind() and listen(), but in
practice this seems to be very unlikely.
The previous randomize-and-retry approach makes it hard to test the sequence of
events for listen(), since the random retry may cause multiple listen errors
even when it is finally successful.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/485cdbd3
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/485cdbd3
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/485cdbd3
Branch: refs/heads/master
Commit: 485cdbd3f680772e081ca05fbda97c0f271c676e
Parents: 1d2d791
Author: Alan Conway <ac...@redhat.com>
Authored: Wed Oct 11 09:47:33 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Oct 11 22:04:42 2017 -0400
----------------------------------------------------------------------
proton-c/bindings/cpp/src/container_test.cpp | 49 ++++----
proton-c/src/tests/proactor.c | 96 +--------------
proton-c/src/tests/test_port.h | 138 ++++++++++++++++++++++
3 files changed, 168 insertions(+), 115 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/485cdbd3/proton-c/bindings/cpp/src/container_test.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container_test.cpp b/proton-c/bindings/cpp/src/container_test.cpp
index 3b0c86f..c9657b9 100644
--- a/proton-c/bindings/cpp/src/container_test.cpp
+++ b/proton-c/bindings/cpp/src/container_test.cpp
@@ -19,6 +19,10 @@
#include "test_bits.hpp"
+extern "C" {
+#include "../../../../src/tests/test_port.h"
+}
+
#include "proton/connection.hpp"
#include "proton/connection_options.hpp"
#include "proton/container.hpp"
@@ -35,28 +39,23 @@
namespace {
-static std::string make_url(const std::string host, int port) {
+std::string make_url(const std::string& host, int port) {
std::ostringstream url;
url << "amqp://" << host << ":" << port;
return url.str();
}
-int listen_on_random_port(proton::container& c, proton::listener& l, proton::listen_handler* lh=0) {
- int port = 0;
- // I'm going to hell for this:
- std::srand((unsigned int)time(0));
- while (true) {
- port = 20000 + (std::rand() % 30000);
- std::string url = make_url("", port);
- try {
- l = lh ? c.listen(url, *lh) : c.listen(url);
- break;
- } catch (...) {
- // keep trying
- }
- }
- return port;
-}
+// C++ Wrapper for C test port.
+// Binds to a port with REUSEADDR set so that the port is protected from
+// other processes and can safely be used for listening.
+class listen_port {
+ ::test_port_t tp;
+ public:
+ listen_port() { tp = ::test_port(""); } // NOTE: assign tp, don't initialize - Windows.
+ ~listen_port() { ::test_port_close(&tp); }
+ int port() const { return tp.port; }
+ std::string url(const std::string& host="") const { return make_url(host, tp.port); }
+};
struct test_listen_handler : public proton::listen_handler {
bool on_open_, on_accept_, on_close_;
@@ -92,6 +91,7 @@ class test_handler : public proton::messaging_handler {
std::string peer_vhost;
std::string peer_container_id;
+ listen_port port;
proton::listener listener;
test_listen_handler listen_handler;
@@ -100,8 +100,8 @@ class test_handler : public proton::messaging_handler {
{}
void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
- int port = listen_on_random_port(c, listener, &listen_handler);
- proton::connection conn = c.connect(make_url(host, port), opts);
+ listener = c.listen(port.url(), listen_handler);
+ proton::connection conn = c.connect(port.url(host), opts);
}
void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
@@ -183,13 +183,14 @@ int test_container_bad_address() {
}
class stop_tester : public proton::messaging_handler {
+ listen_port port;
proton::listener listener;
// Set up a listener which would block forever
void on_container_start(proton::container& c) PN_CPP_OVERRIDE {
ASSERT(state==0);
- int port = listen_on_random_port(c, listener);
- c.connect(make_url("", port));
+ listener = c.listen(port.url());
+ c.connect(port.url());
c.auto_stop(false);
state = 1;
}
@@ -231,17 +232,17 @@ int test_container_stop() {
struct hang_tester : public proton::messaging_handler {
proton::listener listener;
- int port;
+ listen_port port;
bool done;
hang_tester() : done(false) {}
void connect(proton::container* c) {
- c->connect(make_url("", port));
+ c->connect(port.url());
}
void on_container_start(proton::container& c) PN_CPP_OVERRIDE {
- port = listen_on_random_port(c, listener);
+ listener = c.listen(port.url());
c.schedule(proton::duration(250), make_work(&hang_tester::connect, this, &c));
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/485cdbd3/proton-c/src/tests/proactor.c
----------------------------------------------------------------------
diff --git a/proton-c/src/tests/proactor.c b/proton-c/src/tests/proactor.c
index 5296802..5f84bfe 100644
--- a/proton-c/src/tests/proactor.c
+++ b/proton-c/src/tests/proactor.c
@@ -17,7 +17,7 @@
* under the License.
*/
-#include "../platform/platform.h"
+#include "test_port.h"
#include "test_tools.h"
#include "test_handler.h"
#include "test_config.h"
@@ -38,78 +38,6 @@
static const char *localhost = ""; /* host for connect/listen */
-/* Some very simple platform-secifics to acquire an unused socket */
-#if defined(_WIN32)
-
-#include <winsock2.h>
-#include <ws2tcpip.h>
-typedef SOCKET sock_t;
-void sock_close(sock_t sock) { closesocket(sock); }
-// pni_snprintf not exported. We can live with a simplified version
-// for this test's limited use. Abort if that assumption is wrong.
-#define pni_snprintf pnitst_snprintf
-static int pnitst_snprintf(char *buf, size_t count, const char *fmt, ...) {
- va_list ap;
- va_start(ap, fmt);
- int n = _vsnprintf(buf, count, fmt, ap);
- va_end(ap);
- if (count == 0 || n < 0) {
- perror("proton internal failure on Windows test snprintf");
- abort();
- }
- // Windows and C99 are in agreement.
- return n;
-}
-
-#else /* POSIX */
-# include <sys/types.h>
-# include <sys/socket.h>
-# include <netinet/in.h>
-# include <unistd.h>
-typedef int sock_t;
-void sock_close(sock_t sock) { close(sock); }
-#endif
-
-/* Combines a sock_t with the int and char* versions of the port for convenience */
-typedef struct test_port_t {
- sock_t sock;
- int port; /* port as integer */
- char str[PN_MAX_ADDR]; /* port as string */
- char host_port[PN_MAX_ADDR]; /* host:port string */
-} test_port_t;
-
-/* Modifies tp->host_port to use host, returns the new tp->host_port */
-const char *test_port_use_host(test_port_t *tp, const char *host) {
- pni_snprintf(tp->host_port, sizeof(tp->host_port), "%s:%d", host, tp->port);
- return tp->host_port;
-}
-
-/* Create a socket and bind(INADDR_LOOPBACK:0) to get a free port.
- Use SO_REUSEADDR so other processes can bind and listen on this port.
- Use host to create the host_port address string.
-*/
-test_port_t test_port(const char* host) {
- test_port_t tp = {0};
- tp.sock = socket(AF_INET, SOCK_STREAM, 0);
- TEST_ASSERT_ERRNO(tp.sock >= 0, errno);
- int on = 1;
- int err = setsockopt(tp.sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
- TEST_ASSERT_ERRNO(!err, errno);
- struct sockaddr_in addr = {0};
- addr.sin_family = AF_INET; /* set the type of connection to TCP/IP */
- addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
- addr.sin_port = 0; /* bind to port 0 */
- err = bind(tp.sock, (struct sockaddr*)&addr, sizeof(addr));
- TEST_ASSERT_ERRNO(!err, errno);
- socklen_t len = sizeof(addr);
- err = getsockname(tp.sock, (struct sockaddr*)&addr, &len); /* Get the bound port */
- TEST_ASSERT_ERRNO(!err, errno);
- tp.port = ntohs(addr.sin_port);
- pni_snprintf(tp.str, sizeof(tp.str), "%d", tp.port);
- test_port_use_host(&tp, host);
- return tp;
-}
-
#define ARRAYLEN(A) (sizeof(A)/sizeof((A)[0]))
/* Proactor and handler that take part in a test */
@@ -216,14 +144,9 @@ typedef struct test_listener_t {
test_listener_t test_listen(test_proactor_t *tp, const char *host) {
test_listener_t l = { test_port(host), pn_listener() };
-#if defined(_WIN32)
- sock_close(l.port.sock); // small chance another process will steal the port in Windows
-#endif
pn_proactor_listen(tp->proactor, l.listener, l.port.host_port, 4);
TEST_ETYPE_EQUAL(tp->handler.t, PN_LISTENER_OPEN, test_proactors_run(tp, 1));
-#if !defined(_WIN32)
- sock_close(l.port.sock);
-#endif
+ test_port_close(&l.port);
return l;
}
@@ -620,7 +543,7 @@ static void test_errors(test_t *t) {
TEST_ETYPE_EQUAL(t, PN_PROACTOR_INACTIVE, TEST_PROACTORS_RUN(tps));
}
- sock_close(port.sock);
+ test_port_close(&port);
TEST_PROACTORS_DESTROY(tps);
}
@@ -932,15 +855,6 @@ static void test_parse_addr(test_t *t) {
/* Test pn_proactor_addr funtions */
-/* Windows will need winsock2.h etc.
- These headers are *only* needed for test_netaddr and only for the getnameinfo part.
- This is the only non-portable part of the proactor test suite.
- */
-#if !defined(_WIN32)
-#include <sys/socket.h> /* For socket_storage */
-#include <netdb.h> /* For NI_MAXHOST/NI_MAXSERV */
-#endif
-
static void test_netaddr(test_t *t) {
test_proactor_t tps[] ={ test_proactor(t, open_wake_handler), test_proactor(t, listen_handler) };
pn_proactor_t *client = tps[0].proactor;
@@ -975,8 +889,8 @@ static void test_netaddr(test_t *t) {
const pn_netaddr_t *na = pn_netaddr_remote(ct);
const struct sockaddr *sa = pn_netaddr_sockaddr(na);
TEST_CHECK(t, AF_INET == sa->sa_family);
- char host[NI_MAXHOST] = "";
- char serv[NI_MAXSERV] = "";
+ char host[TEST_PORT_MAX_STR] = "";
+ char serv[TEST_PORT_MAX_STR] = "";
int err = getnameinfo(sa, pn_netaddr_socklen(na),
host, sizeof(host), serv, sizeof(serv),
NI_NUMERICHOST | NI_NUMERICSERV);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/485cdbd3/proton-c/src/tests/test_port.h
----------------------------------------------------------------------
diff --git a/proton-c/src/tests/test_port.h b/proton-c/src/tests/test_port.h
new file mode 100644
index 0000000..85a3fd7
--- /dev/null
+++ b/proton-c/src/tests/test_port.h
@@ -0,0 +1,138 @@
+#ifndef TESTS_TEST_PORT_H
+#define TESTS_TEST_PORT_H
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/* Some simple platform-secifics to acquire an unused socket */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#if defined(_WIN32)
+
+# include <winsock2.h>
+# include <ws2tcpip.h>
+
+typedef SOCKET sock_t;
+
+static void test_snprintf(char *buf, size_t count, const char *fmt, ...) {
+ va_list ap;
+ va_start(ap, fmt);
+ _vsnprintf(buf, count, fmt, ap);
+ buf[count-1] = '\0'; /* _vsnprintf doesn't null-terminate on overflow */
+}
+
+void check_err(int ret, const char *what) {
+ if (ret) {
+ char buf[512];
+ FormatMessage(
+ FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
+ NULL, WSAGetLastError(), NULL, buf, sizeof(buf), NULL);
+ fprintf(stderr, "%s: %s\n", what, buf);
+ abort();
+ }
+}
+
+#else /* POSIX */
+
+# include <sys/types.h>
+# include <sys/socket.h>
+# include <netinet/in.h>
+# include <unistd.h>
+# include <netdb.h>
+# define test_snprintf snprintf
+
+typedef int sock_t;
+
+void check_err(int ret, const char *what) {
+ if (ret) {
+ perror(what); abort();
+ }
+}
+
+#endif
+
+#define TEST_PORT_MAX_STR 1060
+
+/* Combines a sock_t with the int and char* versions of the port for convenience */
+typedef struct test_port_t {
+ sock_t sock;
+ int port; /* port as integer */
+ char str[TEST_PORT_MAX_STR]; /* port as string */
+ char host_port[TEST_PORT_MAX_STR]; /* host:port string */
+} test_port_t;
+
+/* Modifies tp->host_port to use host, returns the new tp->host_port */
+const char *test_port_use_host(test_port_t *tp, const char *host) {
+ test_snprintf(tp->host_port, sizeof(tp->host_port), "%s:%d", host, tp->port);
+ return tp->host_port;
+}
+
+/* Create a socket and bind(INADDR_LOOPBACK:0) to get a free port.
+ Set socket options so the port can be bound and used for listen() within this process,
+ even though it is bound to the test_port socket.
+ Use host to create the host_port address string.
+*/
+test_port_t test_port(const char* host) {
+#ifdef _WIN32
+ static int wsa_started = 0;
+ if (!wsa_started) {
+ WORD wsa_ver = MAKEWORD(2, 2);
+ WSADATA unused;
+ check_err(WSAStartup(wsa_ver, &unused), "WSAStartup");
+ }
+#endif
+ int err = 0;
+ test_port_t tp = {0};
+ tp.sock = socket(AF_INET, SOCK_STREAM, 0);
+ check_err(tp.sock < 0, "socket");
+ int on = 1;
+#ifndef _WIN32
+ check_err(setsockopt(tp.sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on)), "setsockopt");
+#endif
+ struct sockaddr_in addr = {0};
+ addr.sin_family = AF_INET; /* set the type of connection to TCP/IP */
+ addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ addr.sin_port = 0; /* bind to port 0 */
+ err = bind(tp.sock, (struct sockaddr*)&addr, sizeof(addr));
+ check_err(err, "bind");
+ socklen_t len = sizeof(addr);
+ err = getsockname(tp.sock, (struct sockaddr*)&addr, &len); /* Get the bound port */
+ check_err(err, "getsockname");
+ tp.port = ntohs(addr.sin_port);
+ test_snprintf(tp.str, sizeof(tp.str), "%d", tp.port);
+ test_port_use_host(&tp, host);
+#ifdef _WIN32 /* Windows doesn't support the twice-open socket trick */
+ closesocket(tp.sock);
+#endif
+ return tp;
+}
+
+void test_port_close(test_port_t *tp) {
+#ifdef _WIN32
+ WSACleanup();
+#else
+ close(tp->sock);
+#endif
+}
+
+
+#endif // TESTS_TEST_PORT_H
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[4/5] qpid-proton git commit: PROTON-1517: C++ consistent linkage for
listen_handler
Posted by ac...@apache.org.
PROTON-1517: C++ consistent linkage for listen_handler
Make it all non-inline, consistent with message_handler.
Fixes ASAN runtime warnings caused by duplicate vtables.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/1d2d7919
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/1d2d7919
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/1d2d7919
Branch: refs/heads/master
Commit: 1d2d791921bea582b27778bf0b5ebb1853c7e95a
Parents: 31d5ba0
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Oct 10 18:42:01 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Oct 11 22:04:42 2017 -0400
----------------------------------------------------------------------
.../bindings/cpp/include/proton/listen_handler.hpp | 14 ++++++++------
.../bindings/cpp/include/proton/messaging_handler.hpp | 2 +-
proton-c/bindings/cpp/src/listener.cpp | 9 ++++++++-
3 files changed, 17 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1d2d7919/proton-c/bindings/cpp/include/proton/listen_handler.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/listen_handler.hpp b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
index 3e18475..d19be3f 100644
--- a/proton-c/bindings/cpp/include/proton/listen_handler.hpp
+++ b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
@@ -23,6 +23,8 @@
*/
#include "./fwd.hpp"
+#include "./internal/export.hpp"
+#include <string>
/// @file
/// @copybrief proton::listen_handler
@@ -34,12 +36,12 @@ namespace proton {
///
/// Implement this interface and pass to proton::container::listen()
/// to be notified of new connections.
-class listen_handler {
+class PN_CPP_CLASS_EXTERN listen_handler {
public:
- virtual ~listen_handler() {}
+ PN_CPP_EXTERN virtual ~listen_handler();
/// Called when the listener is opened successfully.
- virtual void on_open(listener&) {}
+ PN_CPP_EXTERN virtual void on_open(listener&);
/// Called for each accepted connection.
///
@@ -47,14 +49,14 @@ class listen_handler {
/// the connection. messaging_handler::on_connection_open() will be called with
/// the proton::connection, it can call connection::open() to accept or
/// connection::close() to reject the connection.
- virtual connection_options on_accept(listener&)= 0;
+ PN_CPP_EXTERN virtual connection_options on_accept(listener&)= 0;
/// Called if there is a listening error, with an error message.
/// close() will also be called.
- virtual void on_error(listener&, const std::string&) {}
+ PN_CPP_EXTERN virtual void on_error(listener&, const std::string&);
/// Called when this listen_handler is no longer needed, and can be deleted.
- virtual void on_close(listener&) {}
+ PN_CPP_EXTERN virtual void on_close(listener&);
};
} // proton
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1d2d7919/proton-c/bindings/cpp/include/proton/messaging_handler.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/messaging_handler.hpp b/proton-c/bindings/cpp/include/proton/messaging_handler.hpp
index e777d7d..938d3ff 100644
--- a/proton-c/bindings/cpp/include/proton/messaging_handler.hpp
+++ b/proton-c/bindings/cpp/include/proton/messaging_handler.hpp
@@ -80,7 +80,7 @@ PN_CPP_CLASS_EXTERN messaging_handler {
/// The underlying network transport is open
PN_CPP_EXTERN virtual void on_transport_open(transport &t);
-
+
/// The underlying network transport has closed.
PN_CPP_EXTERN virtual void on_transport_close(transport &t);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/1d2d7919/proton-c/bindings/cpp/src/listener.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/listener.cpp b/proton-c/bindings/cpp/src/listener.cpp
index 2e38076..646cfe1 100644
--- a/proton-c/bindings/cpp/src/listener.cpp
+++ b/proton-c/bindings/cpp/src/listener.cpp
@@ -17,7 +17,9 @@
* under the License.
*/
+#include "proton/connection_options.hpp"
#include "proton/listener.hpp"
+#include "proton/listen_handler.hpp"
#include <proton/listener.h>
@@ -32,7 +34,12 @@ listener::listener(const listener& l) : listener_(l.listener_) {}
listener::~listener() {}
listener& listener::operator=(const listener& l) { listener_ = l.listener_; return *this; }
-// FIXME aconway 2017-10-06: should be a no-op if already closed
+// FIXME aconway 2017-10-06: should be a no-op if already closed - there is a race here.
void listener::stop() { if (listener_) pn_listener_close(listener_); }
+listen_handler::~listen_handler() {}
+void listen_handler::on_open(listener&) {}
+connection_options listen_handler::on_accept(listener&) { return connection_options(); }
+void listen_handler::on_error(listener&, const std::string&) {}
+void listen_handler::on_close(listener&) {}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/5] qpid-proton git commit: PROTON-1618: c++ container: unambiguous
listen success/fail indicator
Posted by ac...@apache.org.
PROTON-1618: c++ container: unambiguous listen success/fail indicator
Added listener_handler::on_open() to indicate a successful listen.
After a call to container::listen():
- on success, call listener_handler::on_open() before any call to listener_handler::on_accept()
- on failure, call listener_handler::on_error() followed by listener_handler::on_close()
An application can tell from the first event received (on_open() vs. on_close()) if the
listen call succeeded or failed.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/31d5ba09
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/31d5ba09
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/31d5ba09
Branch: refs/heads/master
Commit: 31d5ba096aa850e49b0f2475a3c1ef9f171f9a3c
Parents: d0b641b
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Oct 10 16:24:28 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Oct 11 22:04:42 2017 -0400
----------------------------------------------------------------------
.../bindings/cpp/include/proton/container.hpp | 12 ++-
.../cpp/include/proton/listen_handler.hpp | 3 +
proton-c/bindings/cpp/src/container_test.cpp | 89 ++++++++++++--------
.../cpp/src/proactor_container_impl.cpp | 10 ++-
4 files changed, 74 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/31d5ba09/proton-c/bindings/cpp/include/proton/container.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/container.hpp b/proton-c/bindings/cpp/include/proton/container.hpp
index d30e45a..fe3857d 100644
--- a/proton-c/bindings/cpp/include/proton/container.hpp
+++ b/proton-c/bindings/cpp/include/proton/container.hpp
@@ -120,12 +120,16 @@ class PN_CPP_CLASS_EXTERN container {
/// Listen for new connections on `listen_url`.
///
- /// listen_handler::on_accept is called for each incoming connection to determine
+ /// If the listener opens successfully, listen_handler::on_open() is called.
+ /// If it fails to open, listen_handler::on_error() then listen_handler::close()
+ /// are called.
+ ///
+ /// listen_handler::on_accept() is called for each incoming connection to determine
/// the @ref connection_options to use, including the @ref messaging_handler.
///
- /// **Thread safety** - Calls to `listen_handler` methods
- /// are serialized for this listener, but handlers attached to
- /// separate listeners can be safely called concurrently.
+ /// **Thread safety** - Calls to `listen_handler` methods are serialized for
+ /// this listener, but handlers attached to separate listeners may be called
+ /// concurrently.
PN_CPP_EXTERN listener listen(const std::string& listen_url,
listen_handler& handler);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/31d5ba09/proton-c/bindings/cpp/include/proton/listen_handler.hpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/include/proton/listen_handler.hpp b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
index 4ce20e9..3e18475 100644
--- a/proton-c/bindings/cpp/include/proton/listen_handler.hpp
+++ b/proton-c/bindings/cpp/include/proton/listen_handler.hpp
@@ -38,6 +38,9 @@ class listen_handler {
public:
virtual ~listen_handler() {}
+ /// Called when the listener is opened successfully.
+ virtual void on_open(listener&) {}
+
/// Called for each accepted connection.
///
/// Returns connection_options to apply, including a proton::messaging_handler for
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/31d5ba09/proton-c/bindings/cpp/src/container_test.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/container_test.cpp b/proton-c/bindings/cpp/src/container_test.cpp
index c2b1609..3b0c86f 100644
--- a/proton-c/bindings/cpp/src/container_test.cpp
+++ b/proton-c/bindings/cpp/src/container_test.cpp
@@ -35,20 +35,21 @@
namespace {
-static std::string int2string(int n) {
- std::ostringstream strm;
- strm << n;
- return strm.str();
+static std::string make_url(const std::string host, int port) {
+ std::ostringstream url;
+ url << "amqp://" << host << ":" << port;
+ return url.str();
}
-int listen_on_random_port(proton::container& c, proton::listener& l) {
- int port;
+int listen_on_random_port(proton::container& c, proton::listener& l, proton::listen_handler* lh=0) {
+ int port = 0;
// I'm going to hell for this:
std::srand((unsigned int)time(0));
while (true) {
port = 20000 + (std::rand() % 30000);
+ std::string url = make_url("", port);
try {
- l = c.listen("0.0.0.0:" + int2string(port));
+ l = lh ? c.listen(url, *lh) : c.listen(url);
break;
} catch (...) {
// keep trying
@@ -57,6 +58,31 @@ int listen_on_random_port(proton::container& c, proton::listener& l) {
return port;
}
+struct test_listen_handler : public proton::listen_handler {
+ bool on_open_, on_accept_, on_close_;
+ std::string on_error_;
+ test_listen_handler() : on_open_(false), on_accept_(false), on_close_(false) {}
+ proton::connection_options on_accept(proton::listener&) PN_CPP_OVERRIDE {
+ on_accept_ = true;
+ return proton::connection_options();
+ }
+ void on_open(proton::listener&) PN_CPP_OVERRIDE {
+ on_open_ = true;
+ ASSERT(!on_accept_);
+ ASSERT(on_error_.empty());
+ ASSERT(!on_close_);
+ }
+ void on_close(proton::listener&) PN_CPP_OVERRIDE {
+ on_close_ = true;
+ ASSERT(on_open_ || on_error_.size());
+ }
+
+ void on_error(proton::listener&, const std::string& e) PN_CPP_OVERRIDE {
+ on_error_ = e;
+ ASSERT(!on_close_);
+ }
+};
+
class test_handler : public proton::messaging_handler {
public:
const std::string host;
@@ -67,17 +93,21 @@ class test_handler : public proton::messaging_handler {
std::string peer_vhost;
std::string peer_container_id;
proton::listener listener;
+ test_listen_handler listen_handler;
test_handler(const std::string h, const proton::connection_options& c_opts)
: host(h), opts(c_opts), closing(false), done(false)
{}
void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
- int port = listen_on_random_port(c, listener);
- proton::connection conn = c.connect(host + ":" + int2string(port), opts);
+ int port = listen_on_random_port(c, listener, &listen_handler);
+ proton::connection conn = c.connect(make_url(host, port), opts);
}
void on_connection_open(proton::connection &c) PN_CPP_OVERRIDE {
+ ASSERT(listen_handler.on_open_);
+ ASSERT(!listen_handler.on_close_);
+ ASSERT(listen_handler.on_error_.empty());
if (peer_vhost.empty() && !c.virtual_host().empty())
peer_vhost = c.virtual_host();
if (peer_container_id.empty() && !c.container_id().empty())
@@ -94,26 +124,28 @@ class test_handler : public proton::messaging_handler {
int test_container_default_container_id() {
proton::connection_options opts;
- test_handler th(std::string("127.0.0.1"), opts);
+ test_handler th("", opts);
proton::container(th).run();
ASSERT(!th.peer_container_id.empty());
+ ASSERT(th.listen_handler.on_error_.empty());
+ ASSERT(th.listen_handler.on_close_);
return 0;
}
int test_container_vhost() {
proton::connection_options opts;
- opts.virtual_host(std::string("a.b.c"));
- test_handler th(std::string("127.0.0.1"), opts);
+ opts.virtual_host("a.b.c");
+ test_handler th("", opts);
proton::container(th).run();
- ASSERT_EQUAL(th.peer_vhost, std::string("a.b.c"));
+ ASSERT_EQUAL(th.peer_vhost, "a.b.c");
return 0;
}
int test_container_default_vhost() {
proton::connection_options opts;
- test_handler th(std::string("127.0.0.1"), opts);
+ test_handler th("127.0.0.1", opts);
proton::container(th).run();
- ASSERT_EQUAL(th.peer_vhost, std::string("127.0.0.1"));
+ ASSERT_EQUAL(th.peer_vhost, "127.0.0.1");
return 0;
}
@@ -123,25 +155,13 @@ int test_container_no_vhost() {
// Sadly whether or not a 'hostname' field was received cannot be
// determined from here, so just exercise the code
proton::connection_options opts;
- opts.virtual_host(std::string(""));
- test_handler th(std::string("127.0.0.1"), opts);
+ opts.virtual_host("");
+ test_handler th("127.0.0.1", opts);
proton::container(th).run();
- ASSERT_EQUAL(th.peer_vhost, std::string(""));
+ ASSERT_EQUAL(th.peer_vhost, "");
return 0;
}
-struct test_listener : public proton::listen_handler {
- bool on_accept_, on_close_;
- std::string on_error_;
- test_listener() : on_accept_(false), on_close_(false) {}
- proton::connection_options on_accept(proton::listener&) PN_CPP_OVERRIDE {
- on_accept_ = true;
- return proton::connection_options();
- }
- void on_close(proton::listener&) PN_CPP_OVERRIDE { on_close_ = true; }
- void on_error(proton::listener&, const std::string& e) PN_CPP_OVERRIDE { on_error_ = e; }
-};
-
int test_container_bad_address() {
// Listen on a bad address, check for leaks
// Regression test for https://issues.apache.org/jira/browse/PROTON-1217
@@ -151,10 +171,11 @@ int test_container_bad_address() {
try { c.listen("999.666.999.666:0"); } catch (const proton::error&) {}
c.run();
// Dummy listener.
- test_listener l;
- test_handler h2(std::string("999.999.999.666"), proton::connection_options());
+ test_listen_handler l;
+ test_handler h2("999.999.999.666", proton::connection_options());
try { c.listen("999.666.999.666:0", l); } catch (const proton::error&) {}
c.run();
+ ASSERT(!l.on_open_);
ASSERT(!l.on_accept_);
ASSERT(l.on_close_);
ASSERT(!l.on_error_.empty());
@@ -168,7 +189,7 @@ class stop_tester : public proton::messaging_handler {
void on_container_start(proton::container& c) PN_CPP_OVERRIDE {
ASSERT(state==0);
int port = listen_on_random_port(c, listener);
- c.connect("127.0.0.1:" + int2string(port));
+ c.connect(make_url("", port));
c.auto_stop(false);
state = 1;
}
@@ -216,7 +237,7 @@ struct hang_tester : public proton::messaging_handler {
hang_tester() : done(false) {}
void connect(proton::container* c) {
- c->connect("localhost:"+ int2string(port));
+ c->connect(make_url("", port));
}
void on_container_start(proton::container& c) PN_CPP_OVERRIDE {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/31d5ba09/proton-c/bindings/cpp/src/proactor_container_impl.cpp
----------------------------------------------------------------------
diff --git a/proton-c/bindings/cpp/src/proactor_container_impl.cpp b/proton-c/bindings/cpp/src/proactor_container_impl.cpp
index e83cc66..ec617bd 100644
--- a/proton-c/bindings/cpp/src/proactor_container_impl.cpp
+++ b/proton-c/bindings/cpp/src/proactor_container_impl.cpp
@@ -498,9 +498,15 @@ bool container::impl::handle(pn_event_t* event) {
}
return false;
}
- case PN_LISTENER_OPEN:
+ case PN_LISTENER_OPEN: {
+ pn_listener_t* l = pn_event_listener(event);
+ listener_context &lc(listener_context::get(l));
+ if (lc.listen_handler_) {
+ listener lstnr(l);
+ lc.listen_handler_->on_open(lstnr);
+ }
return false;
-
+ }
case PN_LISTENER_ACCEPT: {
pn_listener_t* l = pn_event_listener(event);
pn_connection_t* c = pn_connection();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/5] qpid-proton git commit: PROTON-1592: [python] tx_recv.py raises
exception.
Posted by ac...@apache.org.
PROTON-1592: [python] tx_recv.py raises exception.
Fixed the example: starting a transactional internally opens a sending link for
transactional commands. The example was coded on the assumption that the only
link being opened was its own receiver link, which is incorrect.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/198664fc
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/198664fc
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/198664fc
Branch: refs/heads/master
Commit: 198664fc8e0273d4eab9f6d6f73df20114080a8c
Parents: 1bb1897
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Oct 10 13:07:20 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Oct 11 22:04:42 2017 -0400
----------------------------------------------------------------------
examples/python/tx_recv.py | 6 ++++++
1 file changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/198664fc/examples/python/tx_recv.py
----------------------------------------------------------------------
diff --git a/examples/python/tx_recv.py b/examples/python/tx_recv.py
index 4baddcf..7c0b7cc 100755
--- a/examples/python/tx_recv.py
+++ b/examples/python/tx_recv.py
@@ -40,6 +40,12 @@ class TxRecv(MessagingHandler, TransactionHandler):
self.container.declare_transaction(self.conn, handler=self)
self.transaction = None
+ def on_link_opened(self, event):
+ # NOTE: a transactional client opens an internal sender link for transaction commands,
+ # which we want to ignore.
+ if event.receiver:
+ event.receiver.drain_mode = True
+
def on_message(self, event):
print(event.message.body)
self.transaction.accept(event.delivery)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[5/5] qpid-proton git commit: PROTON-1618: c proactor: unambiguous
listen success/fail indicator
Posted by ac...@apache.org.
PROTON-1618: c proactor: unambiguous listen success/fail indicator
Changed the use of events in all 3 proactor implementations as follows:
After a call to pn_proactor_listen():
- on success, dispatch PN_LISTENER_OPEN before any PN_PROACTOR_ACCEPT
- on failure, set the pn_listener_condition() and dispatch PN_LISTENER_CLOSE
An application can tell from the first event received (OPEN vs. CLOSE) if the OS
listen call succeeded or failed.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/d0b641b7
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/d0b641b7
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/d0b641b7
Branch: refs/heads/master
Commit: d0b641b75eb235d9959bab2792c0264522396321
Parents: 198664f
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Oct 10 14:19:53 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Oct 11 22:04:42 2017 -0400
----------------------------------------------------------------------
proton-c/include/proton/proactor.h | 7 ++++---
proton-c/src/proactor/epoll.c | 4 ++--
proton-c/src/proactor/libuv.c | 4 ++--
proton-c/src/proactor/win_iocp.c | 5 ++---
proton-c/src/tests/proactor.c | 3 ---
5 files changed, 10 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d0b641b7/proton-c/include/proton/proactor.h
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/proactor.h b/proton-c/include/proton/proactor.h
index 414883a..52d3891 100644
--- a/proton-c/include/proton/proactor.h
+++ b/proton-c/include/proton/proactor.h
@@ -122,11 +122,12 @@ PNP_EXTERN void pn_proactor_connect(pn_proactor_t *proactor, pn_connection_t *co
* Start listening for incoming connections.
*
* pn_proactor_wait() will return a @ref PN_LISTENER_OPEN event when the
- * listener is ready to accept connections, or if the listen operation fails.
- * If the listen operation failed, then pn_listener_condition() will be set.
+ * listener is ready to accept connections, or a PN_LISTENER_CLOSE if the listen
+ * operation fails. If the listen failed, pn_listener_condition() will be set.
*
* When the listener is closed by pn_listener_close(), or because of an error, a
- * PN_LISTENER_CLOSE event will be returned and pn_listener_condition() will be set.
+ * PN_LISTENER_CLOSE event will be returned and pn_listener_condition() will be set
+ * for an error.
*
* @note Thread-safe
*
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d0b641b7/proton-c/src/proactor/epoll.c
----------------------------------------------------------------------
diff --git a/proton-c/src/proactor/epoll.c b/proton-c/src/proactor/epoll.c
index d739ab1..3cec04c 100644
--- a/proton-c/src/proactor/epoll.c
+++ b/proton-c/src/proactor/epoll.c
@@ -1401,8 +1401,6 @@ void pn_proactor_listen(pn_proactor_t *p, pn_listener_t *l, const char *addr, in
if (addrinfo) {
freeaddrinfo(addrinfo);
}
- /* Always put an OPEN event for symmetry, even if we immediately close with err */
- pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
bool notify = wake(&l->context);
if (l->psockets_size == 0) { /* All failed, create dummy socket with an error */
@@ -1414,6 +1412,8 @@ void pn_proactor_listen(pn_proactor_t *p, pn_listener_t *l, const char *addr, in
} else {
psocket_error(l->psockets, errno, "listen on");
}
+ } else {
+ pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
}
proactor_add(&l->context);
unlock(&l->context.mutex);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d0b641b7/proton-c/src/proactor/libuv.c
----------------------------------------------------------------------
diff --git a/proton-c/src/proactor/libuv.c b/proton-c/src/proactor/libuv.c
index 5effbfe..4088b07 100644
--- a/proton-c/src/proactor/libuv.c
+++ b/proton-c/src/proactor/libuv.c
@@ -641,9 +641,9 @@ static void leader_listen_lh(pn_listener_t *l) {
}
if (err) {
listener_error_lh(l, err, "listening on");
+ } else {
+ pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
}
- /* Always put an OPEN event for symmetry, even if we have an error. */
- pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
}
void pn_listener_free(pn_listener_t *l) {
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d0b641b7/proton-c/src/proactor/win_iocp.c
----------------------------------------------------------------------
diff --git a/proton-c/src/proactor/win_iocp.c b/proton-c/src/proactor/win_iocp.c
index 9a27ae6..09d39cf 100644
--- a/proton-c/src/proactor/win_iocp.c
+++ b/proton-c/src/proactor/win_iocp.c
@@ -2810,9 +2810,6 @@ void pn_proactor_listen(pn_proactor_t *p, pn_listener_t *l, const char *addr, in
if (addrinfo) {
freeaddrinfo(addrinfo);
}
- /* Always put an OPEN event for symmetry, even if we immediately close with err */
- pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
-
if (l->psockets_size == 0) { /* All failed, create dummy socket with an error */
l->psockets = (psocket_t*)calloc(sizeof(psocket_t), 1);
psocket_init(l->psockets, l, false, addr);
@@ -2821,6 +2818,8 @@ void pn_proactor_listen(pn_proactor_t *p, pn_listener_t *l, const char *addr, in
} else {
psocket_error(l->psockets, wsa_err, "listen on");
}
+ } else {
+ pn_collector_put(l->collector, pn_listener__class(), l, PN_LISTENER_OPEN);
}
wakeup(l->psockets);
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d0b641b7/proton-c/src/tests/proactor.c
----------------------------------------------------------------------
diff --git a/proton-c/src/tests/proactor.c b/proton-c/src/tests/proactor.c
index 2a71f6f..5296802 100644
--- a/proton-c/src/tests/proactor.c
+++ b/proton-c/src/tests/proactor.c
@@ -584,7 +584,6 @@ static void test_errors(test_t *t) {
TEST_ETYPE_EQUAL(t, PN_PROACTOR_INACTIVE, TEST_PROACTORS_RUN(tps));
pn_proactor_listen(server, pn_listener(), "127.0.0.1:xxx", 1);
- TEST_ETYPE_EQUAL(t, PN_LISTENER_OPEN, TEST_PROACTORS_RUN(tps));
TEST_ETYPE_EQUAL(t, PN_LISTENER_CLOSE, TEST_PROACTORS_RUN(tps));
TEST_COND_DESC(t, "xxx", last_condition);
TEST_ETYPE_EQUAL(t, PN_PROACTOR_INACTIVE, TEST_PROACTORS_RUN(tps));
@@ -597,7 +596,6 @@ static void test_errors(test_t *t) {
TEST_ETYPE_EQUAL(t, PN_PROACTOR_INACTIVE, TEST_PROACTORS_RUN(tps));
pn_proactor_listen(server, pn_listener(), "nosuch.example.com:", 1);
- TEST_ETYPE_EQUAL(t, PN_LISTENER_OPEN, TEST_PROACTORS_RUN(tps));
TEST_ETYPE_EQUAL(t, PN_LISTENER_CLOSE, TEST_PROACTORS_RUN(tps));
TEST_COND_DESC(t, "nosuch", last_condition);
TEST_ETYPE_EQUAL(t, PN_PROACTOR_INACTIVE, TEST_PROACTORS_RUN(tps));
@@ -608,7 +606,6 @@ static void test_errors(test_t *t) {
pn_proactor_listen(server, l, port.host_port, 1);
TEST_ETYPE_EQUAL(t, PN_LISTENER_OPEN, TEST_PROACTORS_RUN(tps));
pn_proactor_listen(server, pn_listener(), port.host_port, 1); /* Busy */
- TEST_ETYPE_EQUAL(t, PN_LISTENER_OPEN, TEST_PROACTORS_RUN(tps));
TEST_ETYPE_EQUAL(t, PN_LISTENER_CLOSE, TEST_PROACTORS_RUN(tps));
TEST_COND_NAME(t, "proton:io", last_condition);
pn_listener_close(l);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org