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
apache · Nov 30, 2016 · a1a1268 · a1a1268
1 parent dc37144
commit a1a1268
Show file tree

Hide file tree

Showing 18 changed files with 585 additions and 58 deletions.
diff --git 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)

diff --git a/cmake/FindLibWebSockets.cmake 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()
diff --git 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'}
 	            });

diff --git a/console/stand-alone/plugin/html/qdrConnect.html b/console/stand-alone/plugin/html/qdrConnect.html
@@ -22,13 +22,8 @@
               <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>

diff --git 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'

diff --git 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 */
diff --git a/include/qpid/dispatch/server.h b/include/qpid/dispatch/server.h
@@ -253,6 +253,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.
      */

diff --git 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.",

diff --git 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}"
   )