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 2016/11/30 23:13:02 UTC
qpid-dispatch git commit: DISPATCH-103: WebSocket Listeners - first
draft.
Repository: qpid-dispatch
Updated Branches:
refs/heads/master dc3714421 -> a1a1268ff
DISPATCH-103: WebSocket Listeners - first draft.
WebSocket support for dispatch router.
Requires libwebsockets (packaged on fedora, dnf install libwebsockets-delve)
Create a listener with configuration http=true, point a web console at that port to test.
Known issues:
- Occasional problems (segfault, unexpected poll events) when a console disconnects/reconnects.
Limitations to be addressed:
- No SSL support for HTTP ports
- Does not serve the console files via the port, need to set up a separate web server.
- No protocol detection - only HTTP allowed on listeners with http=true
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/a1a1268f
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/a1a1268f
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/a1a1268f
Branch: refs/heads/master
Commit: a1a1268ffbd18eb5c4605fe5e503d70efa7be689
Parents: dc37144
Author: Alan Conway <ac...@redhat.com>
Authored: Mon Nov 28 09:46:24 2016 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Nov 30 18:09:36 2016 -0500
----------------------------------------------------------------------
CMakeLists.txt | 18 +-
cmake/FindLibWebSockets.cmake | 51 ++++
.../dashboard/dispatch/dispatch.comService.js | 4 +-
console/stand-alone/plugin/html/qdrConnect.html | 7 +-
console/stand-alone/plugin/js/qdrService.js | 6 +-
include/qpid/dispatch/driver.h | 36 ++-
include/qpid/dispatch/server.h | 5 +
python/qpid_dispatch/management/qdrouter.json | 40 +--
src/CMakeLists.txt | 8 +-
src/connection_manager.c | 4 +-
src/dispatch.c | 1 +
src/http-libwebsockets.c | 295 +++++++++++++++++++
src/http-none.c | 41 +++
src/http.h | 32 ++
src/posix/driver.c | 71 ++++-
src/server.c | 18 +-
src/server_private.h | 2 +
tests/system_tests_protocol_settings.py | 4 +-
18 files changed, 585 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5b2069d..b0c1f45 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -26,6 +26,8 @@ endif (CMAKE_BUILD_TYPE MATCHES "Deb")
project(qpid-dispatch C)
+set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake")
+
# Build time switch to turn off memory pooling.
option(USE_MEMORY_POOL "Use per-thread memory pools" ON)
option(QD_MEMORY_STATS "Track memory pool usage statistics" ON)
@@ -81,6 +83,12 @@ endif()
set(QPID_DISPATCH_CONFDIR ${SYSCONF_INSTALL_DIR}/qpid-dispatch)
+if (NOT COMMAND add_compile_options)
+ macro (add_compile_options option)
+ set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${option}")
+ endmacro (add_compile_options)
+endif (NOT COMMAND add_compile_options)
+
##
## Find dependencies
##
@@ -89,6 +97,10 @@ find_library(dl_lib dl)
find_library(rt_lib rt)
find_package(Proton 0.13 REQUIRED)
+## Optional dependencies
+include(FindLibWebSockets)
+option(USE_LIBWEBSOCKETS "Use libwebsockets for WebSocket support" ${LibWebSockets_FOUND})
+
##
## Find Valgrind
##
@@ -107,12 +119,6 @@ include_directories(
${PYTHON_INCLUDE_PATH}
)
-if (NOT COMMAND add_compile_options)
- macro (add_compile_options option)
- SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${option}")
- endmacro (add_compile_options)
-endif (NOT COMMAND add_compile_options)
-
add_compile_options(-pthread)
add_compile_options(-Wall)
if (NOT CMAKE_SYSTEM_NAME STREQUAL SunOS)
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/cmake/FindLibWebSockets.cmake
----------------------------------------------------------------------
diff --git a/cmake/FindLibWebSockets.cmake b/cmake/FindLibWebSockets.cmake
new file mode 100644
index 0000000..2e6bb9b
--- /dev/null
+++ b/cmake/FindLibWebSockets.cmake
@@ -0,0 +1,51 @@
+#
+# 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.
+#
+
+# Find libwebsockets include dirs and libraries.
+#
+# Sets the following variables:
+#
+# LibWebSockets_FOUND - True if headers and requested libraries were found
+# LibWebSockets_INCLUDE_DIRS - LibWebSockets include directories
+# LibWebSockets_LIBRARIES - Link these to use libwebsockets.
+#
+# This module reads hints about search locations from variables::
+# LIBWEBSOCKETS_LIBRARYDIR - Preferred library directory e.g. <prefix>/lib
+# LIBWEBSOCKETS_ROOT - Preferred installation prefix
+# CMAKE_INSTALL_PREFIX - Install location for the current project.
+# LIBWEBSOCKETS_INCLUDEDIR - Preferred include directory e.g. <prefix>/include
+
+find_library(LibWebSockets_LIBRARIES
+ NAMES websockets libwebsockets
+ HINTS ${LIBWEBSOCKETS_LIBRARYDIR} ${LIBWEBSOCKETS_ROOT} ${CMAKE_INSTALL_PREFIX}
+ )
+
+find_path(LibWebSockets_INCLUDE_DIRS
+ NAMES libwebsockets.h
+ HINTS ${LIBWEBSOCKETS_INCLUDEDIR} ${LIBWEBSOCKETS_ROOT}/include ${CMAKE_INSTALL_PREFIX}/include
+ PATHS /usr/include
+ )
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(LibWebSockets DEFAULT_MSG LibWebSockets_LIBRARIES LibWebSockets_INCLUDE_DIRS)
+
+if(NOT LibWebSockets_FOUND)
+ set(LibWebSockets_LIBRARIES "")
+ set(LibWebSockets_INCLUDE_DIRS "")
+endif()
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/console/dispatch-dashboard/dispatch/static/dashboard/dispatch/dispatch.comService.js
----------------------------------------------------------------------
diff --git a/console/dispatch-dashboard/dispatch/static/dashboard/dispatch/dispatch.comService.js b/console/dispatch-dashboard/dispatch/static/dashboard/dispatch/dispatch.comService.js
index ace792a..41814e4 100644
--- a/console/dispatch-dashboard/dispatch/static/dashboard/dispatch/dispatch.comService.js
+++ b/console/dispatch-dashboard/dispatch/static/dashboard/dispatch/dispatch.comService.js
@@ -875,7 +875,9 @@ var QDR = (function(QDR) {
try {
QDR.log.debug("trying to connect to ws://" + baseAddress)
connection = self.rhea.connect({
- connection_details:ws('ws://' + baseAddress, ["binary", "base64", "AMQWSB10"]),
+ // FIXME aconway 2016-11-29: "binary" for wsproxy,
+ // should also include "amqp" - waiting on libwebsocket fix.
+ connection_details:ws('ws://' + baseAddress, ["binary"]),
reconnect:true,
properties: {console_identifier: 'Dispatch console'}
});
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/console/stand-alone/plugin/html/qdrConnect.html
----------------------------------------------------------------------
diff --git a/console/stand-alone/plugin/html/qdrConnect.html b/console/stand-alone/plugin/html/qdrConnect.html
index 270a4be..1ca10b2 100644
--- a/console/stand-alone/plugin/html/qdrConnect.html
+++ b/console/stand-alone/plugin/html/qdrConnect.html
@@ -22,13 +22,8 @@ under the License.
<div class="connect-column">
<div class="alert alert-success">
<p>
- Enter the address and port of a <strong><a href="http://qpid.apache.org/components/dispatch-router/" target="_blank">Qpid Dispatch Router</a></strong> to connect..
+ Enter the address and a HTML-enabled port of a <strong><a href="http://qpid.apache.org/components/dispatch-router/" target="_blank">Qpid Dispatch Router</a></strong> to connect..
</p>
-
- <p>
- The port should be a websockets <==> tcp proxy.
- </p>
-
<p>
When Autostart is checked, you will be automatically logged in to the router the next time you start the console.
</p>
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/console/stand-alone/plugin/js/qdrService.js
----------------------------------------------------------------------
diff --git a/console/stand-alone/plugin/js/qdrService.js b/console/stand-alone/plugin/js/qdrService.js
index 8691b75..6b08b2e 100644
--- a/console/stand-alone/plugin/js/qdrService.js
+++ b/console/stand-alone/plugin/js/qdrService.js
@@ -971,8 +971,10 @@ console.dump(e)
QDR.log.debug("****** calling rhea.connect ********")
var connection;
try {
- connection = self.rhea.connect({
- connection_details: ws('ws://' + baseAddress, ["binary", "base64", "AMQWSB10"]),
+ connection = self.rhea.connect({
+ // FIXME aconway 2016-11-29: "binary" for wsproxy,
+ // should also include "amqp" - waiting on libwebsocket fix.
+ connection_details: ws('ws://' + baseAddress, ["binary"]),
reconnect: true,
properties: {
console_identifier: 'Dispatch console'
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/include/qpid/dispatch/driver.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/driver.h b/include/qpid/dispatch/driver.h
index 52cbbed..28552f7 100644
--- a/include/qpid/dispatch/driver.h
+++ b/include/qpid/dispatch/driver.h
@@ -30,6 +30,8 @@
#include <proton/types.h>
typedef struct qd_log_source_t qd_log_source_t;
+typedef struct qd_http_t qd_http_t;
+typedef struct qd_http_connector_t qd_http_connector_t;
/** @file
* API for the Driver Layer.
@@ -138,6 +140,7 @@ void qdpn_driver_free(qdpn_driver_t *driver);
* @param[in] host local host address to listen on
* @param[in] port local port to listen on
* @param[in] protocol family to use (IPv4 or IPv6 or 0). If 0 (zero) is passed in the protocol family will be automatically determined from the address
+ * @param[in] http points to qd_http_t if HTTP is enabled.
* @param[in] context application-supplied, can be accessed via
* qdpn_listener_context()
* @return a new listener on the given host:port, NULL if error
@@ -146,8 +149,11 @@ qdpn_listener_t *qdpn_listener(qdpn_driver_t *driver,
const char *host,
const char *port,
const char *protocol_family,
+ qd_http_t *http,
void* context);
+qd_http_t *qdpn_listener_http(qdpn_listener_t *l);
+
/** Access the head listener for a driver.
*
* @param[in] driver the driver whose head listener will be returned
@@ -330,6 +336,19 @@ pn_transport_t *qdpn_connector_transport(qdpn_connector_t *connector);
*/
void qdpn_connector_close(qdpn_connector_t *connector);
+/** Call when the socket is already closed, an the connector needs updating.
+ *
+ * @param[in] connector the connector whose socket will be closed
+ */
+void qdpn_connector_after_close(qdpn_connector_t *connector);
+
+
+/** Socket has been closed externally, mark it closed.
+ *
+ * @param[in] connector the connector whose socket will be closed
+ */
+void qdpn_connector_mark_closed(qdpn_connector_t *connector);
+
/** Determine if the connector is closed.
*
* @return True if closed, otherwise false
@@ -382,11 +401,13 @@ bool qdpn_connector_activated(qdpn_connector_t *connector, qdpn_activate_criteri
*
* @param[in] driver driver that will 'own' this listener
* @param[in] fd existing socket for listener to listen on
+ * @param[in] http if non-NULL enable as a HTTP listener
* @param[in] context application-supplied, can be accessed via
* qdpn_listener_context()
* @return a new listener on the given host:port, NULL if error
*/
-qdpn_listener_t *qdpn_listener_fd(qdpn_driver_t *driver, pn_socket_t fd, void *context);
+qdpn_listener_t *qdpn_listener_fd(qdpn_driver_t *driver, pn_socket_t fd,
+ qd_http_t *http, void *context);
pn_socket_t qdpn_listener_get_fd(qdpn_listener_t *listener);
@@ -400,7 +421,18 @@ pn_socket_t qdpn_listener_get_fd(qdpn_listener_t *listener);
*/
qdpn_connector_t *qdpn_connector_fd(qdpn_driver_t *driver, pn_socket_t fd, void *context);
-pn_socket_t qdpn_connector_get_fd(qdpn_connector_t *connector);
+/** Get the file descriptor for this connector */
+int qdpn_connector_get_fd(qdpn_connector_t *connector);
+
+/** Get the HTTP per-connector state for this connector, NULL if not enabled. */
+qd_http_connector_t *qdpn_connector_http(qdpn_connector_t* c);
+
+/** Set the wakeup time on the connector */
+void qdpn_connector_wakeup(qdpn_connector_t* c, pn_timestamp_t t);
+
+/** Current time according */
+pn_timestamp_t qdpn_now();
+/**@}*/
#endif /* driver.h */
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/include/qpid/dispatch/server.h
----------------------------------------------------------------------
diff --git a/include/qpid/dispatch/server.h b/include/qpid/dispatch/server.h
index de20385..aebe36e 100644
--- a/include/qpid/dispatch/server.h
+++ b/include/qpid/dispatch/server.h
@@ -254,6 +254,11 @@ typedef struct qd_server_config_t {
char *protocol_family;
/**
+ * Accept HTTP connections, allow WebSocket "amqp" protocol upgrades.
+ */
+ bool http;
+
+ /**
* Connection name, used as a reference from other parts of the configuration.
*/
char *name;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/python/qpid_dispatch/management/qdrouter.json
----------------------------------------------------------------------
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index 64c9131..e666455 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -497,10 +497,10 @@
"deprecated": true,
"description": "(DEPRECATED) This value is no longer used in the router.",
"create": true
- }
+ }
}
},
-
+
"sslProfile": {
"description":"Attributes for setting TLS/SSL configuration for connections.",
"referential": true,
@@ -548,7 +548,7 @@
}
}
},
-
+
"listener": {
"description": "Listens for incoming connections to the router.",
"extends": "configurationEntity",
@@ -573,6 +573,12 @@
"description": "['IPv4', 'IPv6'] IPv4: Internet Protocol version 4; IPv6: Internet Protocol version 6. If not specified, the protocol family will be automatically determined from the address.",
"create": true
},
+ "http": {
+ "type": "boolean",
+ "default": false,
+ "description": "Accept HTTP connections that can upgrade to AMQP over WebSocket",
+ "create": true
+ },
"role": {
"type": [
"normal",
@@ -590,13 +596,13 @@
"required": false,
"create": true,
"description": "For the 'inter-router' role only. This value assigns a cost metric to the inter-router connection. The default (and minimum) value is one. Higher values represent higher costs. The cost is used to influence the routing algorithm as it attempts to use the path with the lowest total cost from ingress to egress."
- },
+ },
"sslProfile": {
"type": "string",
"required": false,
"description": "Name of the sslProfile.",
"create": true
- },
+ },
"saslMechanisms": {
"type": "string",
"required": false,
@@ -687,7 +693,7 @@
"create": true,
"deprecated": true,
"description": "(DEPRECATED) This attribute is now controlled by the requireEncryption attribute."
- }
+ }
}
},
@@ -715,7 +721,7 @@
"description": "['IPv4', 'IPv6'] IPv4: Internet Protocol version 4; IPv6: Internet Protocol version 6. If not specified, the protocol family will be automatically determined from the address.",
"create": true
},
-
+
"role": {
"type": [
"normal",
@@ -814,7 +820,7 @@
"type": "string",
"default": "127.0.0.1",
"create": true
- }
+ }
}
},
@@ -1321,12 +1327,12 @@
"singleton": true,
"attributes": {
"listener": {
- "type": "string",
+ "type": "string",
"description": "The name of the listener to send the proxied tcp traffic to."
},
"wsport": {
"type": "integer",
- "description": "port on which to listen for websocket traffic",
+ "description": "port on which to listen for websocket traffic",
"default": 5673
},
"proxy": {
@@ -1462,7 +1468,7 @@
"receiverDenied": {"type": "integer", "graph": true}
}
},
-
+
"container": {
"description":"(DEPRECATED)Attributes related to the AMQP container. This entity has been deprecated. Use the router entity instead.",
"extends": "configurationEntity",
@@ -1502,7 +1508,7 @@
"create": true
}
}
- },
+ },
"waypoint": {
"description":"(DEPRECATED) A remote node that messages for an address pass through. This entity has been deprecated. Use autoLink instead",
@@ -1535,8 +1541,8 @@
"create": true
}
}
- },
-
+ },
+
"fixedAddress": {
"description":"(DEPRECATED) Establishes treatment for addresses starting with a prefix. This entity has been deprecated. Use address instead",
"extends": "configurationEntity",
@@ -1573,8 +1579,8 @@
"create": true
}
}
- },
-
+ },
+
"linkRoutePattern": {
"description":"(DEPRECATED) An address pattern to match against link sources and targets to cause the router to link-route the attach across the network to a remote node. This entity has been deprecated. Use linkRoute instead",
"deprecated": true,
@@ -1602,7 +1608,7 @@
"create": true
}
}
- },
+ },
"dummy": {
"description": "Dummy entity for test purposes.",
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 72f4d07..8e37e56 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -90,6 +90,12 @@ set(qpid_dispatch_SOURCES
trace_mask.c
)
+if(USE_LIBWEBSOCKETS)
+ set(qpid_dispatch_SOURCES ${qpid_dispatch_SOURCES} http-libwebsockets.c)
+else(USE_LIBWEBSOCKETS)
+ set(qpid_dispatch_SOURCES ${qpid_dispatch_SOURCES} http-none.c)
+endif(USE_LIBWEBSOCKETS)
+
if(USE_MEMORY_POOL)
list(APPEND qpid_dispatch_SOURCES alloc_pool.c)
endif()
@@ -100,7 +106,7 @@ set_property(
)
add_library(qpid-dispatch SHARED ${qpid_dispatch_SOURCES})
-target_link_libraries(qpid-dispatch ${Proton_LIBRARIES} ${pthread_lib} ${rt_lib} ${dl_lib} ${PYTHON_LIBRARIES})
+target_link_libraries(qpid-dispatch ${Proton_LIBRARIES} ${pthread_lib} ${rt_lib} ${dl_lib} ${PYTHON_LIBRARIES} ${LibWebSockets_LIBRARIES})
set_target_properties(qpid-dispatch PROPERTIES
LINK_FLAGS "${CATCH_UNDEFINED}"
)
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/connection_manager.c
----------------------------------------------------------------------
diff --git a/src/connection_manager.c b/src/connection_manager.c
index f737b73..63e2d45 100644
--- a/src/connection_manager.c
+++ b/src/connection_manager.c
@@ -193,6 +193,7 @@ static qd_error_t load_server_config(qd_dispatch_t *qd, qd_server_config_t *conf
config->role = qd_entity_get_string(entity, "role"); CHECK();
config->inter_router_cost = qd_entity_opt_long(entity, "cost", 1); CHECK();
config->protocol_family = qd_entity_opt_string(entity, "protocolFamily", 0); CHECK();
+ config->http = qd_entity_opt_bool(entity, "http", false); CHECK();
config->max_frame_size = qd_entity_get_long(entity, "maxFrameSize"); CHECK();
config->max_sessions = qd_entity_get_long(entity, "maxSessions"); CHECK();
uint64_t ssn_frames = qd_entity_get_long(entity, "maxSessionFrames"); CHECK();
@@ -329,10 +330,11 @@ qd_config_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_entit
DEQ_ITEM_INIT(cl);
DEQ_INSERT_TAIL(cm->config_listeners, cl);
- qd_log(cm->log_source, QD_LOG_INFO, "Configured Listener: %s:%s proto=%s, role=%s%s%s",
+ qd_log(cm->log_source, QD_LOG_INFO, "Configured Listener: %s:%s proto=%s, role=%s%s%s%s",
cl->configuration.host, cl->configuration.port,
cl->configuration.protocol_family ? cl->configuration.protocol_family : "any",
cl->configuration.role,
+ cl->configuration.http ? ", http" : "",
cl->ssl_profile ? ", sslProfile=":"",
cl->ssl_profile ? cl->ssl_profile->name:"");
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/dispatch.c
----------------------------------------------------------------------
diff --git a/src/dispatch.c b/src/dispatch.c
index 15f9ca8..3a8c412 100644
--- a/src/dispatch.c
+++ b/src/dispatch.c
@@ -27,6 +27,7 @@
#include "config.h"
#include "dispatch_private.h"
#include "alloc.h"
+#include "http.h"
#include "log_private.h"
#include "router_private.h"
#include "message_private.h"
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/http-libwebsockets.c
----------------------------------------------------------------------
diff --git a/src/http-libwebsockets.c b/src/http-libwebsockets.c
new file mode 100644
index 0000000..bbc3b15
--- /dev/null
+++ b/src/http-libwebsockets.c
@@ -0,0 +1,295 @@
+/*
+ * 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.
+ */
+
+#include <qpid/dispatch/atomic.h>
+#include <qpid/dispatch/amqp.h>
+#include <qpid/dispatch/driver.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/threading.h>
+#include <qpid/dispatch/timer.h>
+
+#include <libwebsockets.h>
+
+#include <assert.h>
+#include <errno.h>
+
+#include "http.h"
+
+typedef struct qd_http_t {
+ sys_mutex_t *lock;
+ qd_dispatch_t *dispatch;
+ qd_log_source_t *log;
+ struct lws_context *context;
+ qd_timer_t *timer;
+} qd_http_t;
+
+/* TODO aconway 2016-11-29: First cut serializes all access to libwebsockets.
+ * LWS does have multi-thread facilities but it segregates file descriptors into
+ * "serialization groups" which does not match well with dispatches current
+ * and planned future threading strategies. Review when we refactor dispatch
+ * to use the pn_proactor. At least 2 possibilities:
+ *
+ * - treat LWS as single-threaded IO code in the 'leader follower' model,
+ * analogous to how we handle libuv.
+ * - work with LWS upstream to abstract out IO code so each LWS WSI can operate
+ * as a thread-independent unit, like the proton connection_driver.
+ */
+
+static __thread struct {
+ qdpn_connector_t *connector; /* Set before each lws_service call */
+} per_thread = { NULL };
+
+typedef struct buffer_t { void *start; size_t size; size_t cap; } buffer_t;
+
+/* Extra buffering per connection, stored in the lws_wsi_user() space. */
+typedef struct buffers_t {
+ buffer_t wtmp; /* Temp buffer with pre-data header space required by LWS */
+ buffer_t over; /* Can't control LWS read size, buffer the overflow */
+} buffers_t;
+
+static void resize(buffer_t *b, size_t size) {
+ /* FIXME aconway 2016-11-30: handle alloc failure */
+ if (b->start == NULL || b->cap < size) {
+ b->start = realloc(b->start, size);
+ b->size = b->cap = size;
+ }
+ b->size = size;
+}
+
+/* Push as much as possible into the transport, store overflow in over. */
+static void transport_push_max(pn_transport_t *t, pn_bytes_t buf, buffer_t *over) {
+ ssize_t cap;
+ while (buf.size > 0 && (cap = pn_transport_capacity(t)) > 0) {
+ if (buf.size > cap) {
+ pn_transport_push(t, buf.start, cap);
+ buf.start += cap;
+ buf.size -= cap;
+ } else {
+ pn_transport_push(t, buf.start, buf.size);
+ buf.size = 0;
+ }
+ }
+ if (buf.size > 0) {
+ if (buf.size > over->cap) {
+ resize(over, buf.size);
+ }
+ memmove(over->start, buf.start, buf.size);
+ }
+ over->size = buf.size;
+}
+
+static qd_http_t *qd_http_from_wsi(struct lws *wsi) {
+ return (qd_http_t *)lws_context_user(lws_get_context(wsi));
+}
+
+static int callback_amqpws(struct lws *wsi, enum lws_callback_reasons reason,
+ void *user, void *in, size_t len)
+{
+ buffers_t *b = (buffers_t*)user;
+ qd_http_t *h = qd_http_from_wsi(wsi);
+ qdpn_connector_t *c = per_thread.connector;
+ pn_transport_t *t = qdpn_connector_transport(c);
+ const char *name = qdpn_connector_name(c);
+
+ switch (reason) {
+
+ case LWS_CALLBACK_ESTABLISHED: {
+ qd_log(h->log, QD_LOG_DEBUG, "HTTP from %s upgraded to AMQP/WebSocket", name);
+ memset(b, 0, sizeof(*b));
+ break;
+ }
+
+ case LWS_CALLBACK_SERVER_WRITEABLE: {
+ ssize_t size = pn_transport_pending(t);
+ if (size < 0) {
+ return -1;
+ }
+ if (size > 0) {
+ pn_bytes_t wbuf = { size, pn_transport_head(t) };
+ /* lws_write() demands LWS_PRE bytes of free space before the data */
+ resize(&b->wtmp, wbuf.size + LWS_PRE);
+ unsigned char *start = (unsigned char*)b->wtmp.start + LWS_PRE;
+ memcpy(start, wbuf.start, wbuf.size);
+ ssize_t wrote = lws_write(wsi, start, wbuf.size, LWS_WRITE_BINARY);
+ if (wrote < 0) {
+ pn_transport_close_head(t);
+ return -1;
+ } else {
+ pn_transport_pop(t, (size_t)wrote);
+ }
+ }
+ break;
+ }
+
+ case LWS_CALLBACK_RECEIVE: {
+ if (pn_transport_capacity(t) < 0) {
+ return -1;
+ }
+ assert(b->over.size == 0);
+ transport_push_max(t, pn_bytes(len, in), &b->over);
+ if (b->over.size > 0) {
+ qd_log(h->log, QD_LOG_TRACE, "amqp/ws read buffered %z bytes on %s", name);
+ }
+ break;
+ }
+
+ case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE: {
+ qd_log(h->log, QD_LOG_DEBUG, "AMQP/WebSocket peer close from %s", name);
+ pn_transport_close_tail(t);
+ break;
+ }
+
+ case LWS_CALLBACK_CLOSED: {
+ qd_log(h->log, QD_LOG_DEBUG, "AMQP/WebSocket from %s closed", name);
+ qdpn_connector_after_close(c);
+ break;
+ }
+
+ default:
+ break;
+ }
+ return 0;
+}
+
+static int callback_http(struct lws *wsi, enum lws_callback_reasons reason, void *user,
+ void *in, size_t len)
+{
+ qdpn_connector_t *c = per_thread.connector;
+ buffers_t *b = (buffers_t*)user;
+
+ switch (reason) {
+ case LWS_CALLBACK_ESTABLISHED: {
+ memset(b, 0, sizeof(*b));
+ break;
+ }
+ case LWS_CALLBACK_CLOSED: {
+ qdpn_connector_after_close(c);
+ }
+ default:
+ break;
+ }
+ return 0;
+}
+
+static void set_timer_lh(qd_http_t *h);
+
+static void fire_timer(void *void_http) {
+ qd_http_t *h = (qd_http_t*)void_http;
+ sys_mutex_lock(h->lock);
+ lws_service_fd(h->context, NULL);
+ set_timer_lh(h);
+ sys_mutex_unlock(h->lock);
+}
+
+static void set_timer_lh(qd_http_t *h) {
+ if (!h->timer) {
+ h->timer = qd_timer(h->dispatch, fire_timer, h);
+ }
+ qd_timer_schedule(h->timer, 1000);
+}
+
+void qd_http_connector_process(qdpn_connector_t *c) {
+ per_thread.connector = c; /* Pass to lws via thread-local storage */
+
+ struct lws *wsi = (struct lws*)qdpn_connector_http(c);
+ buffers_t *b = (buffers_t*)lws_wsi_user(wsi);
+ qd_http_t * h = qd_http_from_wsi(wsi);
+ pn_transport_t *t = qdpn_connector_transport(c);
+
+ int flags =
+ (qdpn_connector_activated(c, QDPN_CONNECTOR_READABLE) ? POLLIN : 0) |
+ (qdpn_connector_activated(c, QDPN_CONNECTOR_WRITABLE) ? POLLOUT : 0);
+
+ if (b && b->over.size) { /* Consume last over-buffered read */
+ transport_push_max(t, pn_bytes(b->over.size, b->over.start), &b->over);
+ if (b->over.size) { /* Don't let LIBWS read if we still are over */
+ flags &= ~POLLIN;
+ }
+ }
+
+ sys_mutex_lock(h->lock);
+ struct lws_pollfd pfd = { qdpn_connector_get_fd(c), flags, flags };
+ lws_service_fd(h->context, &pfd);
+ set_timer_lh(h);
+ sys_mutex_unlock(h->lock);
+
+ if (pn_transport_capacity(t) > 0)
+ qdpn_connector_activate(c, QDPN_CONNECTOR_READABLE);
+ if (pn_transport_pending(t) > 0)
+ qdpn_connector_activate(c, QDPN_CONNECTOR_WRITABLE);
+
+ pn_timestamp_t now = qdpn_now(NULL);
+ pn_timestamp_t next = pn_transport_tick(t, now);
+ /* If we have overflow, re-process immediately after dispatch, otherwise at
+ * next proton tick.
+ */
+ qdpn_connector_wakeup(c, (b && b->over.size) ? now : next);
+}
+
+qd_http_connector_t *qd_http_connector(qd_http_t *h, qdpn_connector_t *c) {
+ struct lws* wsi = lws_adopt_socket(h->context, qdpn_connector_get_fd(c));
+ return (qd_http_connector_t*)wsi;
+}
+
+static struct lws_protocols protocols[] = {
+ /* first protocol must always be HTTP handler */
+ {
+ "http-only", /* name */
+ callback_http, /* callback */
+ sizeof(buffers_t), /* user data size */
+ },
+ /* "amqp" is the official oasis AMQP over WebSocket protocol name */
+ {
+ "amqp",
+ callback_amqpws,
+ sizeof(buffers_t),
+ },
+ /* "binary" is an alias for "amqp", for compatibility with clients designed
+ * to work with a WebSocket proxy
+ */
+ {
+ "binary",
+ callback_amqpws,
+ sizeof(buffers_t),
+ },
+
+};
+
+qd_http_t *qd_http(qd_dispatch_t *d, qd_log_source_t *log) {
+ qd_http_t *h = calloc(1, sizeof(qd_http_t));
+ h->lock = sys_mutex();
+ h->dispatch = d;
+ h->log = log;
+ lws_set_log_level(0, NULL);
+ struct lws_context_creation_info info = {0};
+ info.port = CONTEXT_PORT_NO_LISTEN;
+ info.protocols = protocols;
+ info.gid = info.uid = -1;
+ info.user = h;
+ h->context = lws_create_context(&info);
+ h->timer = NULL; /* Initialized later. */
+ return h;
+}
+
+void qd_http_free(qd_http_t *h) {
+ sys_mutex_free(h->lock);
+ if (h->timer) qd_timer_free(h->timer);
+ lws_context_destroy(h->context);
+ free(h);
+}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/http-none.c
----------------------------------------------------------------------
diff --git a/src/http-none.c b/src/http-none.c
new file mode 100644
index 0000000..36a25fb
--- /dev/null
+++ b/src/http-none.c
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/driver.h>
+#include "http.h"
+
+/* No HTTP implementation available. */
+
+static qd_log_source_t *log = NULL;
+
+void qd_http_connector_process(qdpn_connector_t *c) {}
+
+qd_http_connector_t *qd_http_connector(qd_http_t *h, qdpn_connector_t *c) {
+
+ return NULL;
+}
+
+qd_http_t *qd_http(qd_dispatch_t *d, qd_log_source_t *l) {
+ log = l;
+ qd_log(log, QD_LOG_WARNING, "HTTP support is not available");
+ return NULL;
+}
+
+void qd_http_free(qd_http_t *h) {}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/http.h
----------------------------------------------------------------------
diff --git a/src/http.h b/src/http.h
new file mode 100644
index 0000000..d2ef982
--- /dev/null
+++ b/src/http.h
@@ -0,0 +1,32 @@
+#ifndef QD_HTTP_H
+#define QD_HTTP_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.
+ */
+
+typedef struct qd_http_t qd_http_t;
+typedef struct qd_dispatch_t qd_dispatch_t;
+typedef struct qd_http_connector_t qd_http_connector_t;
+
+qd_http_t *qd_http(qd_dispatch_t *d, qd_log_source_t *log);
+void qd_http_free(qd_http_t *http);
+qd_http_connector_t *qd_http_connector(qd_http_t *h, qdpn_connector_t *c);
+void qd_http_connector_process(qdpn_connector_t *c);
+
+#endif // QD_HTTP_H
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/posix/driver.c
----------------------------------------------------------------------
diff --git a/src/posix/driver.c b/src/posix/driver.c
index c0f4f6f..c4381b5 100644
--- a/src/posix/driver.c
+++ b/src/posix/driver.c
@@ -46,6 +46,7 @@
#include <qpid/dispatch/error.h>
#include <qpid/dispatch/threading.h>
#include "alloc.h"
+#include "http.h"
#include <proton/error.h>
#include <proton/ssl.h>
#include <proton/object.h>
@@ -70,6 +71,9 @@ const char *protocol_family_ipv6 = "IPv6";
const char *AF_INET6_STR = "AF_INET6";
const char *AF_INET_STR = "AF_INET";
+/* Connector processing function, direct or HTTP. */
+void (*process_fn)(qdpn_connector_t *c);
+
static inline void ignore_result(int unused_result) {
(void) unused_result;
}
@@ -101,10 +105,11 @@ struct qdpn_listener_t {
DEQ_LINKS(qdpn_listener_t);
qdpn_driver_t *driver;
void *context;
+ qd_http_t *http;
int idx;
int fd;
- bool pending;
- bool closed;
+ bool pending:1;
+ bool closed:1;
};
#define PN_NAME_MAX (256)
@@ -119,14 +124,16 @@ struct qdpn_connector_t {
pn_transport_t *transport;
qdpn_listener_t *listener;
void *context;
+ void (*process)(qdpn_connector_t *c);
+ qd_http_connector_t *http;
int idx;
int fd;
int status;
- bool pending_tick;
- bool pending_read;
- bool pending_write;
- bool socket_error;
- bool closed;
+ bool pending_tick:1;
+ bool pending_read:1;
+ bool pending_write:1;
+ bool socket_error:1;
+ bool closed:1;
};
ALLOC_DECLARE(qdpn_listener_t);
@@ -160,6 +167,8 @@ pn_timestamp_t pn_i_now(void)
return ((pn_timestamp_t)now.tv_sec) * 1000 + (now.tv_nsec / 1000000);
}
+pn_timestamp_t qdpn_now() { return pn_i_now(); }
+
#define pn_min(X,Y) ((X) > (Y) ? (Y) : (X))
#define pn_max(X,Y) ((X) < (Y) ? (Y) : (X))
@@ -247,6 +256,7 @@ qdpn_listener_t *qdpn_listener(qdpn_driver_t *driver,
const char *host,
const char *port,
const char *protocol_family,
+ qd_http_t *http,
void* context)
{
if (!driver) return NULL;
@@ -292,11 +302,12 @@ qdpn_listener_t *qdpn_listener(qdpn_driver_t *driver,
return 0;
}
- qdpn_listener_t *l = qdpn_listener_fd(driver, sock, context);
+ qdpn_listener_t *l = qdpn_listener_fd(driver, sock, http, context);
return l;
}
-qdpn_listener_t *qdpn_listener_fd(qdpn_driver_t *driver, int fd, void *context)
+qdpn_listener_t *qdpn_listener_fd(qdpn_driver_t *driver, int fd,
+ qd_http_t *http, void *context)
{
if (!driver) return NULL;
@@ -309,6 +320,7 @@ qdpn_listener_t *qdpn_listener_fd(qdpn_driver_t *driver, int fd, void *context)
l->fd = fd;
l->closed = false;
l->context = context;
+ l->http = http;
qdpn_driver_add_listener(driver, l);
return l;
@@ -397,6 +409,10 @@ qdpn_connector_t *qdpn_listener_accept(qdpn_listener_t *l,
snprintf(c->name, PN_NAME_MAX, "%s", name);
snprintf(c->hostip, PN_NAME_MAX, "%s", hostip);
c->listener = l;
+ if (l->http) {
+ c->http = qd_http_connector(l->http, c);
+ c->process = qd_http_connector_process;
+ }
return c;
}
@@ -490,6 +506,9 @@ qdpn_connector_t *qdpn_connector(qdpn_driver_t *driver,
return c;
}
+
+static void connector_process(qdpn_connector_t *c);
+
qdpn_connector_t *qdpn_connector_fd(qdpn_driver_t *driver, int fd, void *context)
{
if (!driver) return NULL;
@@ -512,6 +531,7 @@ qdpn_connector_t *qdpn_connector_fd(qdpn_driver_t *driver, int fd, void *context
c->transport = pn_transport();
c->context = context;
c->listener = NULL;
+ c->process = connector_process;
qdpn_driver_add_connector(driver, c);
return c;
}
@@ -595,14 +615,11 @@ qdpn_listener_t *qdpn_connector_listener(qdpn_connector_t *ctor)
return ctor ? ctor->listener : NULL;
}
-void qdpn_connector_close(qdpn_connector_t *ctor)
+/* FD is already closed, update the connector state */
+void qdpn_connector_after_close(qdpn_connector_t *ctor)
{
- // XXX: should probably signal engine and callback here
if (!ctor) return;
-
ctor->status = 0;
- if (close(ctor->fd) == -1)
- qdpn_log_errno(ctor->driver, "close");
if (!ctor->closed) {
sys_mutex_lock(ctor->driver->lock);
ctor->closed = true;
@@ -611,6 +628,15 @@ void qdpn_connector_close(qdpn_connector_t *ctor)
}
}
+void qdpn_connector_close(qdpn_connector_t *ctor)
+{
+ if (ctor && !ctor->closed) {
+ if (close(ctor->fd) == -1)
+ qdpn_log_errno(ctor->driver, "close");
+ qdpn_connector_after_close(ctor);
+ }
+}
+
bool qdpn_connector_closed(qdpn_connector_t *ctor)
{
return ctor ? ctor->closed : true;
@@ -690,8 +716,14 @@ static pn_timestamp_t qdpn_connector_tick(qdpn_connector_t *ctor, pn_timestamp_t
void qdpn_connector_process(qdpn_connector_t *c)
{
if (!c || c->closed) return;
- pn_transport_t *transport = c->transport;
+ c->process(c);
+}
+
+static void connector_process(qdpn_connector_t *c)
+{
+ if(c->closed) return;
+ pn_transport_t *transport = c->transport;
///
/// Socket read
///
@@ -743,7 +775,7 @@ void qdpn_connector_process(qdpn_connector_t *c)
}
c->status = 0;
- if (pn_transport_closed(transport)) {
+ if (pn_transport_closed(c->transport)) {
qd_log(c->driver->log, QD_LOG_TRACE, "Closed %s", c->name);
qdpn_connector_close(c);
} else {
@@ -999,3 +1031,10 @@ qdpn_connector_t *qdpn_driver_connector(qdpn_driver_t *d)
return NULL;
}
+qd_http_connector_t *qdpn_connector_http(qdpn_connector_t* c) { return c->http; }
+
+void qdpn_connector_wakeup(qdpn_connector_t* c, pn_timestamp_t t) {
+ c->wakeup = t;
+}
+
+qd_http_t *qdpn_listener_http(qdpn_listener_t* l) { return l->http; }
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/server.c
----------------------------------------------------------------------
diff --git a/src/server.c b/src/server.c
index ee400b1..557c0b4 100644
--- a/src/server.c
+++ b/src/server.c
@@ -631,7 +631,7 @@ static void thread_process_listeners_LH(qd_server_t *qd_server)
continue;
char logbuf[qd_log_max_len()];
-
+
ctx = qd_connection_allocate();
ctx->server = qd_server;
ctx->owner_thread = CONTEXT_UNSPECIFIED_OWNER;
@@ -1388,6 +1388,7 @@ qd_server_t *qd_server(qd_dispatch_t *qd, int thread_count, const char *containe
qd_server->heartbeat_timer = 0;
qd_server->next_connection_id = 1;
qd_server->py_displayname_obj = 0;
+ qd_server->http = qd_http(qd, qd_server->log_source);
qd_log(qd_server->log_source, QD_LOG_INFO, "Container Name: %s", qd_server->container_name);
@@ -1398,6 +1399,7 @@ qd_server_t *qd_server(qd_dispatch_t *qd, int thread_count, const char *containe
void qd_server_free(qd_server_t *qd_server)
{
if (!qd_server) return;
+ qd_http_free(qd_server->http);
for (int i = 0; i < qd_server->thread_count; i++)
thread_free(qd_server->threads[i]);
qd_timer_finalize();
@@ -1688,13 +1690,23 @@ qd_listener_t *qd_server_listen(qd_dispatch_t *qd, const qd_server_config_t *con
li->server = qd_server;
li->config = config;
li->context = context;
- li->pn_listener = qdpn_listener(qd_server->driver, config->host, config->port, config->protocol_family, (void*) li);
+ qd_http_t *http = NULL;
+ if (config->http) {
+ http = qd->server->http;
+ if (!http) {
+ qd_log(qd_server->log_source, QD_LOG_CRITICAL, "HTTP support not available for %s:%s",
+ config->host, config->port);
+ }
+ }
+ li->pn_listener = qdpn_listener(
+ qd_server->driver, config->host, config->port, config->protocol_family, http, li);
if (!li->pn_listener) {
free_qd_listener_t(li);
return 0;
}
- qd_log(qd_server->log_source, QD_LOG_TRACE, "Listening on %s:%s", config->host, config->port);
+ qd_log(qd_server->log_source, QD_LOG_TRACE, "Listening on %s:%s%s", config->host, config->port,
+ config->http ? "(http)":"");
return li;
}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/src/server_private.h
----------------------------------------------------------------------
diff --git a/src/server_private.h b/src/server_private.h
index 642f89d..eb23fa1 100644
--- a/src/server_private.h
+++ b/src/server_private.h
@@ -31,6 +31,7 @@
#include "dispatch_private.h"
#include "timer_private.h"
+#include "http.h"
void qd_server_timer_pending_LH(qd_timer_t *timer);
void qd_server_timer_cancel_LH(qd_timer_t *timer);
@@ -181,6 +182,7 @@ struct qd_server_t {
qd_timer_t *heartbeat_timer;
uint64_t next_connection_id;
void *py_displayname_obj;
+ qd_http_t *http;
};
ALLOC_DECLARE(qd_work_item_t);
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/a1a1268f/tests/system_tests_protocol_settings.py
----------------------------------------------------------------------
diff --git a/tests/system_tests_protocol_settings.py b/tests/system_tests_protocol_settings.py
index 233b9f3..7a9b73d 100644
--- a/tests/system_tests_protocol_settings.py
+++ b/tests/system_tests_protocol_settings.py
@@ -271,10 +271,8 @@ class MaxFrameMaxSessionFramesTooBigTest(TestCase):
begin_lines = [s for s in log_lines if "-> @begin" in s]
# incoming-window is truncated
self.assertTrue(" incoming-window=2147," in begin_lines[0])
- warning_lines = [s for s in log_lines if "(warning)" in s]
+ warning_lines = [s for s in log_lines if "requested maxSessionFrames truncated from 5000000 to 2147" in s]
self.assertTrue(len(warning_lines) == 1)
- self.assertTrue("requested maxSessionFrames truncated from 5000000 to 2147" in warning_lines[0])
-
class MaxFrameMaxSessionFramesZeroTest(TestCase):
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org