You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@qpid.apache.org by GitBox <gi...@apache.org> on 2020/06/05 14:52:36 UTC
[GitHub] [qpid-dispatch] kgiusti commented on a change in pull request #744: For review - Server code clean up
kgiusti commented on a change in pull request #744:
URL: https://github.com/apache/qpid-dispatch/pull/744#discussion_r435955308
##########
File path: src/connection_manager.c
##########
@@ -157,6 +160,7 @@ static qd_config_sasl_plugin_t *qd_find_sasl_plugin(qd_connection_manager_t *cm,
return 0;
}
+// XXX Should this move to server.c or its own server_config.c?
Review comment:
IIRC the qd_server_config_t structure is public - connection_manager and server (perhaps others) have direct access to its fields. I prefer breaking these out to their own .h and .c files.
##########
File path: src/server.c
##########
@@ -17,1657 +17,609 @@
* under the License.
*/
-#include "python_private.h" // must be first!
-#include "dispatch_private.h"
-#include <qpid/dispatch/python_embedded.h>
-
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/log.h>
-#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/server.h>
-#include <qpid/dispatch/failoverlist.h>
-#include <qpid/dispatch/alloc.h>
-#include <qpid/dispatch/platform.h>
+// This must be first to work around a Python bug
+//
+// clang-format off
+#include "python_private.h"
+// clang-format on
+#include <errno.h>
+#include <inttypes.h>
+#include <proton/condition.h>
#include <proton/event.h>
#include <proton/listener.h>
#include <proton/netaddr.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
+#include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/amqp.h>
+#include <qpid/dispatch/ctools.h>
+#include <qpid/dispatch/failoverlist.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/platform.h>
+#include <qpid/dispatch/python_embedded.h>
+#include <qpid/dispatch/server.h>
+#include <qpid/dispatch/threading.h>
+#include <stdio.h>
+#include <string.h>
-
+#include "config.h"
+#include "connection.h"
+#include "connector.h"
+#include "dispatch_private.h"
#include "entity.h"
#include "entity_cache.h"
-#include "dispatch_private.h"
+#include "listener.h"
#include "policy.h"
+#include "remote_sasl.h"
#include "server_private.h"
#include "timer_private.h"
-#include "config.h"
-#include "remote_sasl.h"
-#include <stdio.h>
-#include <string.h>
-#include <errno.h>
-#include <inttypes.h>
-
-struct qd_server_t {
- qd_dispatch_t *qd;
- const int thread_count; /* Immutable */
- const char *container_name;
- const char *sasl_config_path;
- const char *sasl_config_name;
- pn_proactor_t *proactor;
- qd_container_t *container;
- qd_log_source_t *log_source;
- qd_log_source_t *protocol_log_source; // Log source for the PROTOCOL module
- void *start_context;
- sys_cond_t *cond;
- sys_mutex_t *lock;
- qd_connection_list_t conn_list;
- int pause_requests;
- int threads_paused;
- int pause_next_sequence;
- int pause_now_serving;
- uint64_t next_connection_id;
- void *py_displayname_obj;
- qd_http_server_t *http;
- sys_mutex_t *conn_activation_lock;
-};
#define HEARTBEAT_INTERVAL 1000
-ALLOC_DEFINE(qd_listener_t);
-ALLOC_DEFINE(qd_connector_t);
ALLOC_DEFINE(qd_deferred_call_t);
-ALLOC_DEFINE(qd_connection_t);
-
-const char *MECH_EXTERNAL = "EXTERNAL";
-
-//Allowed uidFormat fields.
-const char CERT_COUNTRY_CODE = 'c';
-const char CERT_STATE = 's';
-const char CERT_CITY_LOCALITY = 'l';
-const char CERT_ORGANIZATION_NAME = 'o';
-const char CERT_ORGANIZATION_UNIT = 'u';
-const char CERT_COMMON_NAME = 'n';
-const char CERT_FINGERPRINT_SHA1 = '1';
-const char CERT_FINGERPRINT_SHA256 = '2';
-const char CERT_FINGERPRINT_SHA512 = '5';
-char *COMPONENT_SEPARATOR = ";";
-
-static const int BACKLOG = 50; /* Listening backlog */
-
-static void setup_ssl_sasl_and_open(qd_connection_t *ctx);
-static qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct);
-
-/**
- * This function is set as the pn_transport->tracer and is invoked when proton tries to write the log message to pn_transport->tracer
- */
-void transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // The PROTOCOL module is used exclusively for logging protocol related tracing. The protocol could be AMQP, HTTP, TCP etc.
- qd_log(ctx->server->protocol_log_source, QD_LOG_TRACE, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-
-void connection_transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // Unconditionally write the log at TRACE level to the log file.
- qd_log_impl_v1(ctx->server->protocol_log_source, QD_LOG_TRACE, __FILE__, __LINE__, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-/**
- * Save displayNameService object instance and ImportModule address
- * Called with qd_python_lock held
- */
-qd_error_t qd_register_display_name_service(qd_dispatch_t *qd, void *displaynameservice)
-{
+// Save displayNameService object instance and ImportModule address.
+// Called with qd_python_lock held.
+//
+// XXX But it doesn't use the _lh convention?
+qd_error_t qd_register_display_name_service(qd_dispatch_t* qd, void* displaynameservice) {
Review comment:
We've been pretty lax about the format of function documentation. While I like the way the doxygen format reads in the code, I've never actually generated (or used) documentation from the source.
Way back in the beginning we designed a big chunk of what is now the router as a library intended to be public. We no longer have that goal, but it may explain why some functions have more formal documentation than others.
##########
File path: src/connector.c
##########
@@ -0,0 +1,111 @@
+/*
+ * 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 "connector.h"
+
+#include <proton/connection.h>
+
+#include "connection.h"
+#include "server_private.h"
+
+ALLOC_DEFINE(qd_connector_t);
+
+const char* qd_connector_policy_vhost(qd_connector_t* connector) {
Review comment:
Our code convention puts the opening brace for functions on its own line directly under the function. All other C constructs - structs, enums, control statements - put the opening brace on the same line.
Closing braces always appear on their own line.
##########
File path: src/server.c
##########
@@ -17,1657 +17,609 @@
* under the License.
*/
-#include "python_private.h" // must be first!
-#include "dispatch_private.h"
-#include <qpid/dispatch/python_embedded.h>
-
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/log.h>
-#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/server.h>
-#include <qpid/dispatch/failoverlist.h>
-#include <qpid/dispatch/alloc.h>
-#include <qpid/dispatch/platform.h>
+// This must be first to work around a Python bug
+//
+// clang-format off
+#include "python_private.h"
+// clang-format on
+#include <errno.h>
+#include <inttypes.h>
+#include <proton/condition.h>
#include <proton/event.h>
#include <proton/listener.h>
#include <proton/netaddr.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
+#include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/amqp.h>
+#include <qpid/dispatch/ctools.h>
+#include <qpid/dispatch/failoverlist.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/platform.h>
+#include <qpid/dispatch/python_embedded.h>
+#include <qpid/dispatch/server.h>
+#include <qpid/dispatch/threading.h>
+#include <stdio.h>
+#include <string.h>
-
+#include "config.h"
+#include "connection.h"
+#include "connector.h"
+#include "dispatch_private.h"
#include "entity.h"
#include "entity_cache.h"
-#include "dispatch_private.h"
+#include "listener.h"
#include "policy.h"
+#include "remote_sasl.h"
#include "server_private.h"
#include "timer_private.h"
-#include "config.h"
-#include "remote_sasl.h"
-#include <stdio.h>
-#include <string.h>
-#include <errno.h>
-#include <inttypes.h>
-
-struct qd_server_t {
- qd_dispatch_t *qd;
- const int thread_count; /* Immutable */
- const char *container_name;
- const char *sasl_config_path;
- const char *sasl_config_name;
- pn_proactor_t *proactor;
- qd_container_t *container;
- qd_log_source_t *log_source;
- qd_log_source_t *protocol_log_source; // Log source for the PROTOCOL module
- void *start_context;
- sys_cond_t *cond;
- sys_mutex_t *lock;
- qd_connection_list_t conn_list;
- int pause_requests;
- int threads_paused;
- int pause_next_sequence;
- int pause_now_serving;
- uint64_t next_connection_id;
- void *py_displayname_obj;
- qd_http_server_t *http;
- sys_mutex_t *conn_activation_lock;
-};
#define HEARTBEAT_INTERVAL 1000
-ALLOC_DEFINE(qd_listener_t);
-ALLOC_DEFINE(qd_connector_t);
ALLOC_DEFINE(qd_deferred_call_t);
-ALLOC_DEFINE(qd_connection_t);
-
-const char *MECH_EXTERNAL = "EXTERNAL";
-
-//Allowed uidFormat fields.
-const char CERT_COUNTRY_CODE = 'c';
-const char CERT_STATE = 's';
-const char CERT_CITY_LOCALITY = 'l';
-const char CERT_ORGANIZATION_NAME = 'o';
-const char CERT_ORGANIZATION_UNIT = 'u';
-const char CERT_COMMON_NAME = 'n';
-const char CERT_FINGERPRINT_SHA1 = '1';
-const char CERT_FINGERPRINT_SHA256 = '2';
-const char CERT_FINGERPRINT_SHA512 = '5';
-char *COMPONENT_SEPARATOR = ";";
-
-static const int BACKLOG = 50; /* Listening backlog */
-
-static void setup_ssl_sasl_and_open(qd_connection_t *ctx);
-static qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct);
-
-/**
- * This function is set as the pn_transport->tracer and is invoked when proton tries to write the log message to pn_transport->tracer
- */
-void transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // The PROTOCOL module is used exclusively for logging protocol related tracing. The protocol could be AMQP, HTTP, TCP etc.
- qd_log(ctx->server->protocol_log_source, QD_LOG_TRACE, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-
-void connection_transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // Unconditionally write the log at TRACE level to the log file.
- qd_log_impl_v1(ctx->server->protocol_log_source, QD_LOG_TRACE, __FILE__, __LINE__, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-/**
- * Save displayNameService object instance and ImportModule address
- * Called with qd_python_lock held
- */
-qd_error_t qd_register_display_name_service(qd_dispatch_t *qd, void *displaynameservice)
-{
+// Save displayNameService object instance and ImportModule address.
+// Called with qd_python_lock held.
+//
+// XXX But it doesn't use the _lh convention?
+qd_error_t qd_register_display_name_service(qd_dispatch_t* qd, void* displaynameservice) {
if (displaynameservice) {
qd->server->py_displayname_obj = displaynameservice;
- Py_XINCREF((PyObject *)qd->server->py_displayname_obj);
+ Py_XINCREF((PyObject*) qd->server->py_displayname_obj);
Review comment:
IIUC the (unspoken) convention is to put a space between <type> and * in casts.
A space between the closing paran on the cast and the casted object should be part of the convention going forward.
##########
File path: src/server.c
##########
@@ -17,1657 +17,609 @@
* under the License.
*/
-#include "python_private.h" // must be first!
-#include "dispatch_private.h"
-#include <qpid/dispatch/python_embedded.h>
-
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/log.h>
-#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/server.h>
-#include <qpid/dispatch/failoverlist.h>
-#include <qpid/dispatch/alloc.h>
-#include <qpid/dispatch/platform.h>
+// This must be first to work around a Python bug
+//
+// clang-format off
+#include "python_private.h"
+// clang-format on
+#include <errno.h>
+#include <inttypes.h>
+#include <proton/condition.h>
#include <proton/event.h>
#include <proton/listener.h>
#include <proton/netaddr.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
+#include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/amqp.h>
+#include <qpid/dispatch/ctools.h>
+#include <qpid/dispatch/failoverlist.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/platform.h>
+#include <qpid/dispatch/python_embedded.h>
+#include <qpid/dispatch/server.h>
+#include <qpid/dispatch/threading.h>
+#include <stdio.h>
+#include <string.h>
-
+#include "config.h"
+#include "connection.h"
+#include "connector.h"
+#include "dispatch_private.h"
#include "entity.h"
#include "entity_cache.h"
-#include "dispatch_private.h"
+#include "listener.h"
#include "policy.h"
+#include "remote_sasl.h"
#include "server_private.h"
#include "timer_private.h"
-#include "config.h"
-#include "remote_sasl.h"
-#include <stdio.h>
-#include <string.h>
-#include <errno.h>
-#include <inttypes.h>
-
-struct qd_server_t {
- qd_dispatch_t *qd;
- const int thread_count; /* Immutable */
- const char *container_name;
- const char *sasl_config_path;
- const char *sasl_config_name;
- pn_proactor_t *proactor;
- qd_container_t *container;
- qd_log_source_t *log_source;
- qd_log_source_t *protocol_log_source; // Log source for the PROTOCOL module
- void *start_context;
- sys_cond_t *cond;
- sys_mutex_t *lock;
- qd_connection_list_t conn_list;
- int pause_requests;
- int threads_paused;
- int pause_next_sequence;
- int pause_now_serving;
- uint64_t next_connection_id;
- void *py_displayname_obj;
- qd_http_server_t *http;
- sys_mutex_t *conn_activation_lock;
-};
#define HEARTBEAT_INTERVAL 1000
-ALLOC_DEFINE(qd_listener_t);
-ALLOC_DEFINE(qd_connector_t);
ALLOC_DEFINE(qd_deferred_call_t);
-ALLOC_DEFINE(qd_connection_t);
-
-const char *MECH_EXTERNAL = "EXTERNAL";
-
-//Allowed uidFormat fields.
-const char CERT_COUNTRY_CODE = 'c';
-const char CERT_STATE = 's';
-const char CERT_CITY_LOCALITY = 'l';
-const char CERT_ORGANIZATION_NAME = 'o';
-const char CERT_ORGANIZATION_UNIT = 'u';
-const char CERT_COMMON_NAME = 'n';
-const char CERT_FINGERPRINT_SHA1 = '1';
-const char CERT_FINGERPRINT_SHA256 = '2';
-const char CERT_FINGERPRINT_SHA512 = '5';
-char *COMPONENT_SEPARATOR = ";";
-
-static const int BACKLOG = 50; /* Listening backlog */
-
-static void setup_ssl_sasl_and_open(qd_connection_t *ctx);
-static qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct);
-
-/**
- * This function is set as the pn_transport->tracer and is invoked when proton tries to write the log message to pn_transport->tracer
- */
-void transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // The PROTOCOL module is used exclusively for logging protocol related tracing. The protocol could be AMQP, HTTP, TCP etc.
- qd_log(ctx->server->protocol_log_source, QD_LOG_TRACE, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-
-void connection_transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // Unconditionally write the log at TRACE level to the log file.
- qd_log_impl_v1(ctx->server->protocol_log_source, QD_LOG_TRACE, __FILE__, __LINE__, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-/**
- * Save displayNameService object instance and ImportModule address
- * Called with qd_python_lock held
- */
-qd_error_t qd_register_display_name_service(qd_dispatch_t *qd, void *displaynameservice)
-{
+// Save displayNameService object instance and ImportModule address.
+// Called with qd_python_lock held.
+//
+// XXX But it doesn't use the _lh convention?
+qd_error_t qd_register_display_name_service(qd_dispatch_t* qd, void* displaynameservice) {
if (displaynameservice) {
qd->server->py_displayname_obj = displaynameservice;
- Py_XINCREF((PyObject *)qd->server->py_displayname_obj);
+ Py_XINCREF((PyObject*) qd->server->py_displayname_obj);
return QD_ERROR_NONE;
- }
- else {
+ } else {
return qd_error(QD_ERROR_VALUE, "displaynameservice is not set");
}
}
-
-/**
- * Returns a char pointer to a user id which is constructed from components specified in the config->ssl_uid_format.
- * Parses through each component and builds a semi-colon delimited string which is returned as the user id.
- */
-static const char *transport_get_user(qd_connection_t *conn, pn_transport_t *tport)
-{
- const qd_server_config_t *config =
- conn->connector ? &conn->connector->config : &conn->listener->config;
-
- if (config->ssl_uid_format) {
- // The ssl_uid_format length cannot be greater that 7
- assert(strlen(config->ssl_uid_format) < 8);
-
- //
- // The tokens in the uidFormat strings are delimited by comma. Load the individual components of the uidFormat
- // into the components[] array. The maximum number of components that are allowed are 7 namely, c, s, l, o, u, n, (1 or 2 or 5)
- //
- char components[8];
-
- //The strcpy() function copies the string pointed to by src, including the terminating null byte ('\0'), to the buffer pointed to by dest.
- strncpy(components, config->ssl_uid_format, 7);
-
- const char *country_code = 0;
- const char *state = 0;
- const char *locality_city = 0;
- const char *organization = 0;
- const char *org_unit = 0;
- const char *common_name = 0;
- //
- // SHA1 is 20 octets (40 hex characters); SHA256 is 32 octets (64 hex characters).
- // SHA512 is 64 octets (128 hex characters)
- //
- char fingerprint[129] = "\0";
-
- int uid_length = 0;
- int semi_colon_count = -1;
-
- int component_count = strlen(components);
-
- for (int x = 0; x < component_count ; x++) {
- // accumulate the length into uid_length on each pass so we definitively know the number of octets to malloc.
- if (components[x] == CERT_COUNTRY_CODE) {
- country_code = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_COUNTRY_NAME);
- if (country_code) {
- uid_length += strlen((const char *)country_code);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_STATE) {
- state = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_STATE_OR_PROVINCE);
- if (state) {
- uid_length += strlen((const char *)state);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_CITY_LOCALITY) {
- locality_city = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_CITY_OR_LOCALITY);
- if (locality_city) {
- uid_length += strlen((const char *)locality_city);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_ORGANIZATION_NAME) {
- organization = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_ORGANIZATION_NAME);
- if(organization) {
- uid_length += strlen((const char *)organization);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_ORGANIZATION_UNIT) {
- org_unit = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_ORGANIZATION_UNIT);
- if(org_unit) {
- uid_length += strlen((const char *)org_unit);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_COMMON_NAME) {
- common_name = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_COMMON_NAME);
- if(common_name) {
- uid_length += strlen((const char *)common_name);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_FINGERPRINT_SHA1 || components[x] == CERT_FINGERPRINT_SHA256 || components[x] == CERT_FINGERPRINT_SHA512) {
- // Allocate the memory for message digest
- int out = 0;
-
- int fingerprint_length = 0;
- if(components[x] == CERT_FINGERPRINT_SHA1) {
- fingerprint_length = 40;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA1);
- }
- else if (components[x] == CERT_FINGERPRINT_SHA256) {
- fingerprint_length = 64;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA256);
- }
- else if (components[x] == CERT_FINGERPRINT_SHA512) {
- fingerprint_length = 128;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA512);
- }
-
- (void) out; // avoid 'out unused' compiler warnings if NDEBUG undef'ed
- assert (out != PN_ERR);
-
- uid_length += fingerprint_length;
- semi_colon_count++;
- }
- else {
- // This is an unrecognized component. log a critical error
- qd_log(conn->server->log_source, QD_LOG_CRITICAL, "Unrecognized component '%c' in uidFormat ", components[x]);
- return 0;
- }
- }
-
- if(uid_length > 0) {
- char *user_id = malloc((uid_length + semi_colon_count + 1) * sizeof(char)); // the +1 is for the '\0' character
- //
- // We have allocated memory for user_id. We are responsible for freeing this memory. Set conn->free_user_id
- // to true so that we know that we have to free the user_id
- //
- conn->free_user_id = true;
- memset(user_id, 0, uid_length + semi_colon_count + 1);
-
- // The components in the user id string must appear in the same order as it appears in the component string. that is
- // why we have this loop
- for (int x=0; x < component_count ; x++) {
- if (components[x] == CERT_COUNTRY_CODE) {
- if (country_code) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) country_code);
- }
- }
- else if (components[x] == CERT_STATE) {
- if (state) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) state);
- }
- }
- else if (components[x] == CERT_CITY_LOCALITY) {
- if (locality_city) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) locality_city);
- }
- }
- else if (components[x] == CERT_ORGANIZATION_NAME) {
- if (organization) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) organization);
- }
- }
- else if (components[x] == CERT_ORGANIZATION_UNIT) {
- if (org_unit) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) org_unit);
- }
- }
- else if (components[x] == CERT_COMMON_NAME) {
- if (common_name) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) common_name);
- }
- }
- else if (components[x] == CERT_FINGERPRINT_SHA1 || components[x] == CERT_FINGERPRINT_SHA256 || components[x] == CERT_FINGERPRINT_SHA512) {
- if (strlen((char *) fingerprint) > 0) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) fingerprint);
- }
- }
- }
- if (config->ssl_uid_name_mapping_file) {
- // Translate extracted id into display name
- qd_python_lock_state_t lock_state = qd_python_lock();
- PyObject *result = PyObject_CallMethod((PyObject *)conn->server->py_displayname_obj, "query", "(ss)", config->ssl_profile, user_id );
- if (result) {
- free(user_id);
- user_id = py_string_2_c(result);
- Py_XDECREF(result);
- } else {
- qd_log(conn->server->log_source, QD_LOG_DEBUG, "Internal: failed to read displaynameservice query result");
- }
- qd_python_unlock(lock_state);
- }
- qd_log(conn->server->log_source, QD_LOG_DEBUG, "User id is '%s' ", user_id);
- return user_id;
- }
- }
- else //config->ssl_uid_format not specified, just return the username provided by the proton transport.
- return pn_transport_get_user(tport);
-
- return 0;
-}
-
-
-void qd_connection_set_user(qd_connection_t *conn)
-{
- pn_transport_t *tport = pn_connection_transport(conn->pn_conn);
- pn_sasl_t *sasl = pn_sasl(tport);
- if (sasl) {
- const char *mech = pn_sasl_get_mech(sasl);
- conn->user_id = pn_transport_get_user(tport);
- // We want to set the user name only if it is not already set and the selected sasl mechanism is EXTERNAL
- if (mech && strcmp(mech, MECH_EXTERNAL) == 0) {
- const char *user_id = transport_get_user(conn, tport);
- if (user_id)
- conn->user_id = user_id;
- }
- }
-}
-
-
-qd_error_t qd_entity_refresh_sslProfile(qd_entity_t* entity, void *impl)
-{
+qd_error_t qd_entity_refresh_sslProfile(qd_entity_t* entity, void* impl) {
return QD_ERROR_NONE;
}
-qd_error_t qd_entity_refresh_authServicePlugin(qd_entity_t* entity, void *impl)
-{
+qd_error_t qd_entity_refresh_authServicePlugin(qd_entity_t* entity, void* impl) {
return QD_ERROR_NONE;
}
+qd_server_t* qd_server(qd_dispatch_t* qd, int thread_count, const char* container_name, const char* sasl_config_path,
+ const char* sasl_config_name) {
+ // Initialize const members. Zero initialize all others.
-static qd_error_t listener_setup_ssl(qd_connection_t *ctx, const qd_server_config_t *config, pn_transport_t *tport)
-{
- pn_ssl_domain_t *domain = pn_ssl_domain(PN_SSL_MODE_SERVER);
- if (!domain) return qd_error(QD_ERROR_RUNTIME, "No SSL support");
+ qd_server_t tmp = {.thread_count = thread_count};
+ // XXX This type contructor is different from what I see elsewhere.
+ qd_server_t* server = NEW(qd_server_t);
- // setup my identifying cert:
- if (pn_ssl_domain_set_credentials(domain,
- config->ssl_certificate_file,
- config->ssl_private_key_file,
- config->ssl_password)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set SSL credentials");
+ if (!server) {
+ return NULL;
}
- // for peer authentication:
- if (config->ssl_trusted_certificate_db) {
- if (pn_ssl_domain_set_trusted_ca_db(domain, config->ssl_trusted_certificate_db)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set trusted SSL CA" );
- }
- }
+ memcpy(server, &tmp, sizeof(tmp));
- if (config->ssl_ciphers) {
- if (pn_ssl_domain_set_ciphers(domain, config->ssl_ciphers)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set ciphers. The ciphers string might be invalid. Use openssl ciphers -v <ciphers> to validate");
- }
- }
+ server->qd = qd;
+ server->log_source = qd_log_source("SERVER");
+ server->protocol_log_source = qd_log_source("PROTOCOL");
+ server->container_name = container_name;
+ server->sasl_config_path = sasl_config_path;
+ server->sasl_config_name = sasl_config_name;
+ server->proactor = pn_proactor();
+ server->container = NULL;
+ server->start_context = 0; // XXX Should be NULL?
Review comment:
In general the code avoids using NULL in favor of 0. Not arguing for or against - that's Just the Way It Is.
Probably should be consistent.
##########
File path: src/server.c
##########
@@ -17,1657 +17,609 @@
* under the License.
*/
-#include "python_private.h" // must be first!
-#include "dispatch_private.h"
-#include <qpid/dispatch/python_embedded.h>
-
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/log.h>
-#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/server.h>
-#include <qpid/dispatch/failoverlist.h>
-#include <qpid/dispatch/alloc.h>
-#include <qpid/dispatch/platform.h>
+// This must be first to work around a Python bug
+//
+// clang-format off
+#include "python_private.h"
+// clang-format on
+#include <errno.h>
+#include <inttypes.h>
+#include <proton/condition.h>
#include <proton/event.h>
#include <proton/listener.h>
#include <proton/netaddr.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
+#include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/amqp.h>
+#include <qpid/dispatch/ctools.h>
+#include <qpid/dispatch/failoverlist.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/platform.h>
+#include <qpid/dispatch/python_embedded.h>
+#include <qpid/dispatch/server.h>
+#include <qpid/dispatch/threading.h>
+#include <stdio.h>
+#include <string.h>
-
+#include "config.h"
+#include "connection.h"
+#include "connector.h"
+#include "dispatch_private.h"
#include "entity.h"
#include "entity_cache.h"
-#include "dispatch_private.h"
+#include "listener.h"
#include "policy.h"
+#include "remote_sasl.h"
#include "server_private.h"
#include "timer_private.h"
-#include "config.h"
-#include "remote_sasl.h"
-#include <stdio.h>
-#include <string.h>
-#include <errno.h>
-#include <inttypes.h>
-
-struct qd_server_t {
- qd_dispatch_t *qd;
- const int thread_count; /* Immutable */
- const char *container_name;
- const char *sasl_config_path;
- const char *sasl_config_name;
- pn_proactor_t *proactor;
- qd_container_t *container;
- qd_log_source_t *log_source;
- qd_log_source_t *protocol_log_source; // Log source for the PROTOCOL module
- void *start_context;
- sys_cond_t *cond;
- sys_mutex_t *lock;
- qd_connection_list_t conn_list;
- int pause_requests;
- int threads_paused;
- int pause_next_sequence;
- int pause_now_serving;
- uint64_t next_connection_id;
- void *py_displayname_obj;
- qd_http_server_t *http;
- sys_mutex_t *conn_activation_lock;
-};
#define HEARTBEAT_INTERVAL 1000
-ALLOC_DEFINE(qd_listener_t);
-ALLOC_DEFINE(qd_connector_t);
ALLOC_DEFINE(qd_deferred_call_t);
-ALLOC_DEFINE(qd_connection_t);
-
-const char *MECH_EXTERNAL = "EXTERNAL";
-
-//Allowed uidFormat fields.
-const char CERT_COUNTRY_CODE = 'c';
-const char CERT_STATE = 's';
-const char CERT_CITY_LOCALITY = 'l';
-const char CERT_ORGANIZATION_NAME = 'o';
-const char CERT_ORGANIZATION_UNIT = 'u';
-const char CERT_COMMON_NAME = 'n';
-const char CERT_FINGERPRINT_SHA1 = '1';
-const char CERT_FINGERPRINT_SHA256 = '2';
-const char CERT_FINGERPRINT_SHA512 = '5';
-char *COMPONENT_SEPARATOR = ";";
-
-static const int BACKLOG = 50; /* Listening backlog */
-
-static void setup_ssl_sasl_and_open(qd_connection_t *ctx);
-static qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct);
-
-/**
- * This function is set as the pn_transport->tracer and is invoked when proton tries to write the log message to pn_transport->tracer
- */
-void transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // The PROTOCOL module is used exclusively for logging protocol related tracing. The protocol could be AMQP, HTTP, TCP etc.
- qd_log(ctx->server->protocol_log_source, QD_LOG_TRACE, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-
-void connection_transport_tracer(pn_transport_t *transport, const char *message)
-{
- qd_connection_t *ctx = (qd_connection_t*) pn_transport_get_context(transport);
- if (ctx) {
- // Unconditionally write the log at TRACE level to the log file.
- qd_log_impl_v1(ctx->server->protocol_log_source, QD_LOG_TRACE, __FILE__, __LINE__, "[%"PRIu64"]:%s", ctx->connection_id, message);
- }
-}
-/**
- * Save displayNameService object instance and ImportModule address
- * Called with qd_python_lock held
- */
-qd_error_t qd_register_display_name_service(qd_dispatch_t *qd, void *displaynameservice)
-{
+// Save displayNameService object instance and ImportModule address.
+// Called with qd_python_lock held.
+//
+// XXX But it doesn't use the _lh convention?
+qd_error_t qd_register_display_name_service(qd_dispatch_t* qd, void* displaynameservice) {
if (displaynameservice) {
qd->server->py_displayname_obj = displaynameservice;
- Py_XINCREF((PyObject *)qd->server->py_displayname_obj);
+ Py_XINCREF((PyObject*) qd->server->py_displayname_obj);
return QD_ERROR_NONE;
- }
- else {
+ } else {
return qd_error(QD_ERROR_VALUE, "displaynameservice is not set");
}
}
-
-/**
- * Returns a char pointer to a user id which is constructed from components specified in the config->ssl_uid_format.
- * Parses through each component and builds a semi-colon delimited string which is returned as the user id.
- */
-static const char *transport_get_user(qd_connection_t *conn, pn_transport_t *tport)
-{
- const qd_server_config_t *config =
- conn->connector ? &conn->connector->config : &conn->listener->config;
-
- if (config->ssl_uid_format) {
- // The ssl_uid_format length cannot be greater that 7
- assert(strlen(config->ssl_uid_format) < 8);
-
- //
- // The tokens in the uidFormat strings are delimited by comma. Load the individual components of the uidFormat
- // into the components[] array. The maximum number of components that are allowed are 7 namely, c, s, l, o, u, n, (1 or 2 or 5)
- //
- char components[8];
-
- //The strcpy() function copies the string pointed to by src, including the terminating null byte ('\0'), to the buffer pointed to by dest.
- strncpy(components, config->ssl_uid_format, 7);
-
- const char *country_code = 0;
- const char *state = 0;
- const char *locality_city = 0;
- const char *organization = 0;
- const char *org_unit = 0;
- const char *common_name = 0;
- //
- // SHA1 is 20 octets (40 hex characters); SHA256 is 32 octets (64 hex characters).
- // SHA512 is 64 octets (128 hex characters)
- //
- char fingerprint[129] = "\0";
-
- int uid_length = 0;
- int semi_colon_count = -1;
-
- int component_count = strlen(components);
-
- for (int x = 0; x < component_count ; x++) {
- // accumulate the length into uid_length on each pass so we definitively know the number of octets to malloc.
- if (components[x] == CERT_COUNTRY_CODE) {
- country_code = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_COUNTRY_NAME);
- if (country_code) {
- uid_length += strlen((const char *)country_code);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_STATE) {
- state = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_STATE_OR_PROVINCE);
- if (state) {
- uid_length += strlen((const char *)state);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_CITY_LOCALITY) {
- locality_city = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_CITY_OR_LOCALITY);
- if (locality_city) {
- uid_length += strlen((const char *)locality_city);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_ORGANIZATION_NAME) {
- organization = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_ORGANIZATION_NAME);
- if(organization) {
- uid_length += strlen((const char *)organization);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_ORGANIZATION_UNIT) {
- org_unit = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_ORGANIZATION_UNIT);
- if(org_unit) {
- uid_length += strlen((const char *)org_unit);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_COMMON_NAME) {
- common_name = pn_ssl_get_remote_subject_subfield(pn_ssl(tport), PN_SSL_CERT_SUBJECT_COMMON_NAME);
- if(common_name) {
- uid_length += strlen((const char *)common_name);
- semi_colon_count++;
- }
- }
- else if (components[x] == CERT_FINGERPRINT_SHA1 || components[x] == CERT_FINGERPRINT_SHA256 || components[x] == CERT_FINGERPRINT_SHA512) {
- // Allocate the memory for message digest
- int out = 0;
-
- int fingerprint_length = 0;
- if(components[x] == CERT_FINGERPRINT_SHA1) {
- fingerprint_length = 40;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA1);
- }
- else if (components[x] == CERT_FINGERPRINT_SHA256) {
- fingerprint_length = 64;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA256);
- }
- else if (components[x] == CERT_FINGERPRINT_SHA512) {
- fingerprint_length = 128;
- out = pn_ssl_get_cert_fingerprint(pn_ssl(tport), fingerprint, fingerprint_length + 1, PN_SSL_SHA512);
- }
-
- (void) out; // avoid 'out unused' compiler warnings if NDEBUG undef'ed
- assert (out != PN_ERR);
-
- uid_length += fingerprint_length;
- semi_colon_count++;
- }
- else {
- // This is an unrecognized component. log a critical error
- qd_log(conn->server->log_source, QD_LOG_CRITICAL, "Unrecognized component '%c' in uidFormat ", components[x]);
- return 0;
- }
- }
-
- if(uid_length > 0) {
- char *user_id = malloc((uid_length + semi_colon_count + 1) * sizeof(char)); // the +1 is for the '\0' character
- //
- // We have allocated memory for user_id. We are responsible for freeing this memory. Set conn->free_user_id
- // to true so that we know that we have to free the user_id
- //
- conn->free_user_id = true;
- memset(user_id, 0, uid_length + semi_colon_count + 1);
-
- // The components in the user id string must appear in the same order as it appears in the component string. that is
- // why we have this loop
- for (int x=0; x < component_count ; x++) {
- if (components[x] == CERT_COUNTRY_CODE) {
- if (country_code) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) country_code);
- }
- }
- else if (components[x] == CERT_STATE) {
- if (state) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) state);
- }
- }
- else if (components[x] == CERT_CITY_LOCALITY) {
- if (locality_city) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) locality_city);
- }
- }
- else if (components[x] == CERT_ORGANIZATION_NAME) {
- if (organization) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) organization);
- }
- }
- else if (components[x] == CERT_ORGANIZATION_UNIT) {
- if (org_unit) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) org_unit);
- }
- }
- else if (components[x] == CERT_COMMON_NAME) {
- if (common_name) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) common_name);
- }
- }
- else if (components[x] == CERT_FINGERPRINT_SHA1 || components[x] == CERT_FINGERPRINT_SHA256 || components[x] == CERT_FINGERPRINT_SHA512) {
- if (strlen((char *) fingerprint) > 0) {
- if(*user_id != '\0')
- strcat(user_id, COMPONENT_SEPARATOR);
- strcat(user_id, (char *) fingerprint);
- }
- }
- }
- if (config->ssl_uid_name_mapping_file) {
- // Translate extracted id into display name
- qd_python_lock_state_t lock_state = qd_python_lock();
- PyObject *result = PyObject_CallMethod((PyObject *)conn->server->py_displayname_obj, "query", "(ss)", config->ssl_profile, user_id );
- if (result) {
- free(user_id);
- user_id = py_string_2_c(result);
- Py_XDECREF(result);
- } else {
- qd_log(conn->server->log_source, QD_LOG_DEBUG, "Internal: failed to read displaynameservice query result");
- }
- qd_python_unlock(lock_state);
- }
- qd_log(conn->server->log_source, QD_LOG_DEBUG, "User id is '%s' ", user_id);
- return user_id;
- }
- }
- else //config->ssl_uid_format not specified, just return the username provided by the proton transport.
- return pn_transport_get_user(tport);
-
- return 0;
-}
-
-
-void qd_connection_set_user(qd_connection_t *conn)
-{
- pn_transport_t *tport = pn_connection_transport(conn->pn_conn);
- pn_sasl_t *sasl = pn_sasl(tport);
- if (sasl) {
- const char *mech = pn_sasl_get_mech(sasl);
- conn->user_id = pn_transport_get_user(tport);
- // We want to set the user name only if it is not already set and the selected sasl mechanism is EXTERNAL
- if (mech && strcmp(mech, MECH_EXTERNAL) == 0) {
- const char *user_id = transport_get_user(conn, tport);
- if (user_id)
- conn->user_id = user_id;
- }
- }
-}
-
-
-qd_error_t qd_entity_refresh_sslProfile(qd_entity_t* entity, void *impl)
-{
+qd_error_t qd_entity_refresh_sslProfile(qd_entity_t* entity, void* impl) {
return QD_ERROR_NONE;
}
-qd_error_t qd_entity_refresh_authServicePlugin(qd_entity_t* entity, void *impl)
-{
+qd_error_t qd_entity_refresh_authServicePlugin(qd_entity_t* entity, void* impl) {
return QD_ERROR_NONE;
}
+qd_server_t* qd_server(qd_dispatch_t* qd, int thread_count, const char* container_name, const char* sasl_config_path,
+ const char* sasl_config_name) {
+ // Initialize const members. Zero initialize all others.
-static qd_error_t listener_setup_ssl(qd_connection_t *ctx, const qd_server_config_t *config, pn_transport_t *tport)
-{
- pn_ssl_domain_t *domain = pn_ssl_domain(PN_SSL_MODE_SERVER);
- if (!domain) return qd_error(QD_ERROR_RUNTIME, "No SSL support");
+ qd_server_t tmp = {.thread_count = thread_count};
+ // XXX This type contructor is different from what I see elsewhere.
+ qd_server_t* server = NEW(qd_server_t);
- // setup my identifying cert:
- if (pn_ssl_domain_set_credentials(domain,
- config->ssl_certificate_file,
- config->ssl_private_key_file,
- config->ssl_password)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set SSL credentials");
+ if (!server) {
+ return NULL;
}
- // for peer authentication:
- if (config->ssl_trusted_certificate_db) {
- if (pn_ssl_domain_set_trusted_ca_db(domain, config->ssl_trusted_certificate_db)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set trusted SSL CA" );
- }
- }
+ memcpy(server, &tmp, sizeof(tmp));
- if (config->ssl_ciphers) {
- if (pn_ssl_domain_set_ciphers(domain, config->ssl_ciphers)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set ciphers. The ciphers string might be invalid. Use openssl ciphers -v <ciphers> to validate");
- }
- }
+ server->qd = qd;
+ server->log_source = qd_log_source("SERVER");
+ server->protocol_log_source = qd_log_source("PROTOCOL");
+ server->container_name = container_name;
+ server->sasl_config_path = sasl_config_path;
+ server->sasl_config_name = sasl_config_name;
+ server->proactor = pn_proactor();
+ server->container = NULL;
+ server->start_context = 0; // XXX Should be NULL?
+ server->lock = sys_mutex();
+ server->conn_activation_lock = sys_mutex();
+ server->cond = sys_cond();
- if (config->ssl_protocols) {
- if (pn_ssl_domain_set_protocols(domain, config->ssl_protocols)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set protocols. The protocols string might be invalid. This list is a space separated string of the allowed TLS protocols (TLSv1 TLSv1.1 TLSv1.2)");
- }
- }
+ DEQ_INIT(server->conn_list);
+ qd_timer_initialize(server->lock);
- const char *trusted = config->ssl_trusted_certificate_db;
- if (config->ssl_trusted_certificates)
- trusted = config->ssl_trusted_certificates;
+ server->pause_requests = 0;
+ server->threads_paused = 0;
+ server->pause_next_sequence = 0;
+ server->pause_now_serving = 0;
+ server->next_connection_id = 1;
+ server->py_displayname_obj = 0;
- // do we force the peer to send a cert?
- if (config->ssl_require_peer_authentication) {
- if (!trusted || pn_ssl_domain_set_peer_authentication(domain, PN_SSL_VERIFY_PEER, trusted)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot set peer authentication");
- }
- }
+ server->http = qd_http_server(server, server->log_source);
- ctx->ssl = pn_ssl(tport);
- if (!ctx->ssl || pn_ssl_init(ctx->ssl, domain, 0)) {
- pn_ssl_domain_free(domain);
- return qd_error(QD_ERROR_RUNTIME, "Cannot initialize SSL");
- }
+ qd_log(server->log_source, QD_LOG_INFO, "Container Name: %s", server->container_name);
- // By default adding ssl to a transport forces encryption to be required, so if it's not set that here
- if (!config->ssl_required) {
- pn_transport_require_encryption(tport, false);
- }
-
- pn_ssl_domain_free(domain);
- return QD_ERROR_NONE;
+ return server;
}
+void qd_server_free(qd_server_t* server) {
+ if (!server) {
+ return;
+ }
-static void decorate_connection(qd_server_t *qd_server, pn_connection_t *conn, const qd_server_config_t *config)
-{
- size_t clen = strlen(QD_CAPABILITY_ANONYMOUS_RELAY);
-
- //
- // Set the container name
- //
- pn_connection_set_container(conn, qd_server->container_name);
+ qd_connection_t* conn = DEQ_HEAD(server->conn_list);
- //
- // Offer ANONYMOUS_RELAY capability
- //
- pn_data_put_symbol(pn_connection_offered_capabilities(conn), pn_bytes(clen, (char*) QD_CAPABILITY_ANONYMOUS_RELAY));
+ while (conn) {
+ qd_log(server->log_source, QD_LOG_INFO, "[C%" PRIu64 "] Closing connection on shutdown", conn->connection_id);
- //
- // Create the connection properties map
- //
- pn_data_put_map(pn_connection_properties(conn));
- pn_data_enter(pn_connection_properties(conn));
+ DEQ_REMOVE_HEAD(server->conn_list);
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_KEY), QD_CONNECTION_PROPERTY_PRODUCT_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_PRODUCT_VALUE), QD_CONNECTION_PROPERTY_PRODUCT_VALUE));
+ if (conn->pn_conn) {
+ pn_transport_t* transport = pn_connection_transport(conn->pn_conn);
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_VERSION_KEY), QD_CONNECTION_PROPERTY_VERSION_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(QPID_DISPATCH_VERSION), QPID_DISPATCH_VERSION));
+ if (transport) {
+ // For the transport tracer
+ pn_transport_set_context(transport, 0);
+ }
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_CONN_ID), QD_CONNECTION_PROPERTY_CONN_ID));
- qd_connection_t *qd_conn = pn_connection_get_context(conn);
- pn_data_put_int(pn_connection_properties(conn), qd_conn->connection_id);
+ qd_session_cleanup(conn);
+ pn_connection_set_context(conn->pn_conn, 0);
+ }
- if (config && config->inter_router_cost > 1) {
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_COST_KEY), QD_CONNECTION_PROPERTY_COST_KEY));
- pn_data_put_int(pn_connection_properties(conn), config->inter_router_cost);
- }
+ if (conn->free_user_id) {
+ free((char*) conn->user_id);
+ }
- if (config) {
- qd_failover_list_t *fol = config->failover_list;
- if (fol) {
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY), QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY));
- pn_data_put_list(pn_connection_properties(conn));
- pn_data_enter(pn_connection_properties(conn));
- int fol_count = qd_failover_list_size(fol);
- for (int i = 0; i < fol_count; i++) {
- pn_data_put_map(pn_connection_properties(conn));
- pn_data_enter(pn_connection_properties(conn));
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY), QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(qd_failover_list_host(fol, i)), qd_failover_list_host(fol, i)));
-
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY), QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(qd_failover_list_port(fol, i)), qd_failover_list_port(fol, i)));
+ sys_mutex_free(conn->deferred_call_lock);
- if (qd_failover_list_scheme(fol, i)) {
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY), QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(qd_failover_list_scheme(fol, i)), qd_failover_list_scheme(fol, i)));
- }
+ free(conn->name);
+ free(conn->role);
- if (qd_failover_list_hostname(fol, i)) {
- pn_data_put_symbol(pn_connection_properties(conn),
- pn_bytes(strlen(QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY), QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY));
- pn_data_put_string(pn_connection_properties(conn),
- pn_bytes(strlen(qd_failover_list_hostname(fol, i)), qd_failover_list_hostname(fol, i)));
- }
- pn_data_exit(pn_connection_properties(conn));
- }
- pn_data_exit(pn_connection_properties(conn));
+ if (conn->policy_settings) {
+ free_qd_policy_settings_t(conn->policy_settings);
}
- }
-
- pn_data_exit(pn_connection_properties(conn));
-}
-/* Wake function for proactor-manaed connections */
-static void connection_wake(qd_connection_t *ctx) {
- if (ctx->pn_conn) pn_connection_wake(ctx->pn_conn);
-}
+ free_qd_connection_t(conn);
-/* Construct a new qd_connection. Thread safe. */
-qd_connection_t *qd_server_connection(qd_server_t *server, qd_server_config_t *config)
-{
- qd_connection_t *ctx = new_qd_connection_t();
- if (!ctx) return NULL;
- ZERO(ctx);
- ctx->pn_conn = pn_connection();
- ctx->deferred_call_lock = sys_mutex();
- ctx->role = strdup(config->role);
- if (!ctx->pn_conn || !ctx->deferred_call_lock || !ctx->role) {
- if (ctx->pn_conn) pn_connection_free(ctx->pn_conn);
- if (ctx->deferred_call_lock) sys_mutex_free(ctx->deferred_call_lock);
- free(ctx->role);
- free(ctx);
- return NULL;
+ conn = DEQ_HEAD(server->conn_list);
}
- ctx->server = server;
- ctx->wake = connection_wake; /* Default, over-ridden for HTTP connections */
- pn_connection_set_context(ctx->pn_conn, ctx);
- DEQ_ITEM_INIT(ctx);
- DEQ_INIT(ctx->deferred_calls);
- DEQ_INIT(ctx->free_link_session_list);
- sys_mutex_lock(server->lock);
- ctx->connection_id = server->next_connection_id++;
- DEQ_INSERT_TAIL(server->conn_list, ctx);
- sys_mutex_unlock(server->lock);
- decorate_connection(ctx->server, ctx->pn_conn, config);
- return ctx;
-}
+ pn_proactor_free(server->proactor);
+ qd_timer_finalize();
+ sys_mutex_free(server->lock);
+ sys_mutex_free(server->conn_activation_lock);
+ sys_cond_free(server->cond);
+ Py_XDECREF((PyObject*) server->py_displayname_obj);
-static void on_accept(pn_event_t *e)
-{
- assert(pn_event_type(e) == PN_LISTENER_ACCEPT);
- pn_listener_t *pn_listener = pn_event_listener(e);
- qd_listener_t *listener = pn_listener_get_context(pn_listener);
- qd_connection_t *ctx = qd_server_connection(listener->server, &listener->config);
- if (!ctx) {
- qd_log(listener->server->log_source, QD_LOG_CRITICAL,
- "Allocation failure during accept to %s", listener->config.host_port);
- return;
- }
- ctx->listener = listener;
- qd_log(listener->server->log_source, QD_LOG_TRACE,
- "[%"PRIu64"]: Accepting incoming connection to '%s'",
- ctx->connection_id, ctx->listener->config.host_port);
- /* Asynchronous accept, configure the transport on PN_CONNECTION_BOUND */
- pn_listener_accept(pn_listener, ctx->pn_conn);
- }
-
-
-/* Log the description, set the transport condition (name, description) close the transport tail. */
-void connect_fail(qd_connection_t *ctx, const char *name, const char *description, ...)
-{
- va_list ap;
- va_start(ap, description);
- qd_verror(QD_ERROR_RUNTIME, description, ap);
- va_end(ap);
- if (ctx->pn_conn) {
- pn_transport_t *t = pn_connection_transport(ctx->pn_conn);
- /* Normally this is closing the transport but if not bound close the connection. */
- pn_condition_t *cond = t ? pn_transport_condition(t) : pn_connection_condition(ctx->pn_conn);
- if (cond && !pn_condition_is_set(cond)) {
- va_start(ap, description);
- pn_condition_vformat(cond, name, description, ap);
- va_end(ap);
- }
- if (t) {
- pn_transport_close_tail(t);
- } else {
- pn_connection_close(ctx->pn_conn);
- }
- }
+ free(server);
}
-
-/* Get the host IP address for the remote end */
-static void set_rhost_port(qd_connection_t *ctx) {
- pn_transport_t *tport = pn_connection_transport(ctx->pn_conn);
- const struct sockaddr* sa = pn_netaddr_sockaddr(pn_transport_remote_addr(tport));
- size_t salen = pn_netaddr_socklen(pn_transport_remote_addr(tport));
- if (sa && salen) {
- char rport[NI_MAXSERV] = "";
- int err = getnameinfo(sa, salen,
- ctx->rhost, sizeof(ctx->rhost), rport, sizeof(rport),
- NI_NUMERICHOST | NI_NUMERICSERV);
- if (!err) {
- snprintf(ctx->rhost_port, sizeof(ctx->rhost_port), "%s:%s", ctx->rhost, rport);
- }
- }
+void qd_server_set_container(qd_dispatch_t* qd, qd_container_t* container) {
+ qd->server->container = container;
}
+void qd_server_trace_all_connections() {
+ qd_dispatch_t* qd = qd_dispatch_get_dispatch();
-/* Configure the transport once it is bound to the connection */
-static void on_connection_bound(qd_server_t *server, pn_event_t *e) {
- pn_connection_t *pn_conn = pn_event_connection(e);
- qd_connection_t *ctx = pn_connection_get_context(pn_conn);
- pn_transport_t *tport = pn_connection_transport(pn_conn);
- pn_transport_set_context(tport, ctx); /* for transport_tracer */
-
- //
- // Proton pushes out its trace to transport_tracer() which in turn writes a trace
- // message to the qdrouter log
- // If trace level logging is enabled on the PROTOCOL module, set PN_TRACE_FRM as the transport trace
- // and also set the transport tracer callback.
- // Note here that if trace level logging is enabled on the DEFAULT module, all modules are logging at trace level too.
- //
- if (qd_log_enabled(ctx->server->protocol_log_source, QD_LOG_TRACE)) {
- pn_transport_trace(tport, PN_TRACE_FRM);
- pn_transport_set_tracer(tport, transport_tracer);
- }
+ if (qd->server) {
+ sys_mutex_lock(qd->server->lock);
- const qd_server_config_t *config = NULL;
- if (ctx->listener) { /* Accepting an incoming connection */
- config = &ctx->listener->config;
- const char *name = config->host_port;
- pn_transport_set_server(tport);
- set_rhost_port(ctx);
-
- sys_mutex_lock(server->lock); /* Policy check is not thread safe */
- ctx->policy_counted = qd_policy_socket_accept(server->qd->policy, ctx->rhost);
- sys_mutex_unlock(server->lock);
- if (!ctx->policy_counted) {
- pn_transport_close_tail(tport);
- pn_transport_close_head(tport);
- return;
- }
+ qd_connection_list_t conn_list = qd->server->conn_list;
+ qd_connection_t* conn = DEQ_HEAD(conn_list);
- // Set up SSL
- if (config->ssl_profile) {
- qd_log(ctx->server->log_source, QD_LOG_TRACE, "Configuring SSL on %s", name);
- if (listener_setup_ssl(ctx, config, tport) != QD_ERROR_NONE) {
- connect_fail(ctx, QD_AMQP_COND_INTERNAL_ERROR, "%s on %s", qd_error_message(), name);
- return;
- }
- }
- //
- // Set up SASL
- //
- sys_mutex_lock(ctx->server->lock);
- pn_sasl_t *sasl = pn_sasl(tport);
- if (ctx->server->sasl_config_path)
- pn_sasl_config_path(sasl, ctx->server->sasl_config_path);
- pn_sasl_config_name(sasl, ctx->server->sasl_config_name);
- if (config->sasl_mechanisms)
- pn_sasl_allowed_mechs(sasl, config->sasl_mechanisms);
- if (config->sasl_plugin_config.auth_service) {
- qd_log(server->log_source, QD_LOG_INFO, "enabling remote authentication service %s", config->sasl_plugin_config.auth_service);
- pn_ssl_domain_t* plugin_ssl_domain = NULL;
- if (config->sasl_plugin_config.use_ssl) {
- plugin_ssl_domain = pn_ssl_domain(PN_SSL_MODE_CLIENT);
-
- if (config->sasl_plugin_config.ssl_certificate_file) {
- if (pn_ssl_domain_set_credentials(plugin_ssl_domain,
- config->sasl_plugin_config.ssl_certificate_file,
- config->sasl_plugin_config.ssl_private_key_file,
- config->sasl_plugin_config.ssl_password)) {
- qd_log(server->log_source, QD_LOG_ERROR, "Cannot set SSL credentials for authentication service");
- }
- }
- if (config->sasl_plugin_config.ssl_trusted_certificate_db) {
- if (pn_ssl_domain_set_trusted_ca_db(plugin_ssl_domain, config->sasl_plugin_config.ssl_trusted_certificate_db)) {
- qd_log(server->log_source, QD_LOG_ERROR, "Cannot set trusted SSL certificate db for authentication service" );
- } else {
- if (pn_ssl_domain_set_peer_authentication(plugin_ssl_domain, PN_SSL_VERIFY_PEER, config->sasl_plugin_config.ssl_trusted_certificate_db)) {
- qd_log(server->log_source, QD_LOG_ERROR, "Cannot set SSL peer verification for authentication service");
- }
- }
- }
- if (config->sasl_plugin_config.ssl_ciphers) {
- if (pn_ssl_domain_set_ciphers(plugin_ssl_domain, config->sasl_plugin_config.ssl_ciphers)) {
- qd_log(server->log_source, QD_LOG_ERROR, "Cannot set ciphers for authentication service");
- }
- }
- if (config->sasl_plugin_config.ssl_protocols) {
- if (pn_ssl_domain_set_protocols(plugin_ssl_domain, config->sasl_plugin_config.ssl_protocols)) {
- qd_log(server->log_source, QD_LOG_ERROR, "Cannot set protocols for authentication service");
- }
- }
- }
- qdr_use_remote_authentication_service(tport, config->sasl_plugin_config.auth_service, config->sasl_plugin_config.sasl_init_hostname, plugin_ssl_domain, server->proactor);
- }
- pn_transport_require_auth(tport, config->requireAuthentication);
- pn_transport_require_encryption(tport, config->requireEncryption);
- pn_sasl_set_allow_insecure_mechs(sasl, config->allowInsecureAuthentication);
- sys_mutex_unlock(ctx->server->lock);
-
- qd_log(ctx->server->log_source, QD_LOG_INFO, "[C%"PRIu64"] Accepted connection to %s from %s",
- ctx->connection_id, name, ctx->rhost_port);
- } else if (ctx->connector) { /* Establishing an outgoing connection */
- config = &ctx->connector->config;
- setup_ssl_sasl_and_open(ctx);
-
- } else { /* No connector and no listener */
- connect_fail(ctx, QD_AMQP_COND_INTERNAL_ERROR, "unknown Connection");
- return;
- }
-
- //
- // Common transport configuration.
- //
- pn_transport_set_max_frame(tport, config->max_frame_size);
- pn_transport_set_channel_max(tport, config->max_sessions - 1);
- pn_transport_set_idle_timeout(tport, config->idle_timeout_seconds * 1000);
-}
+ while (conn) {
+ // If there is already a tracer on the transport, nothing
+ // to do, move on to the next connection.
+ pn_transport_t* transport = pn_connection_transport(conn->pn_conn);
+ if (!pn_transport_get_tracer(transport)) {
+ pn_transport_trace(transport, PN_TRACE_FRM);
+ pn_transport_set_tracer(transport, transport_tracer);
+ }
-static void invoke_deferred_calls(qd_connection_t *conn, bool discard)
-{
- if (!conn)
- return;
+ conn = DEQ_NEXT(conn);
+ }
- // Lock access to deferred_calls, other threads may concurrently add to it. Invoke
- // the calls outside of the critical section.
- //
- sys_mutex_lock(conn->deferred_call_lock);
- qd_deferred_call_t *dc;
- while ((dc = DEQ_HEAD(conn->deferred_calls))) {
- DEQ_REMOVE_HEAD(conn->deferred_calls);
- sys_mutex_unlock(conn->deferred_call_lock);
- dc->call(dc->context, discard);
- free_qd_deferred_call_t(dc);
- sys_mutex_lock(conn->deferred_call_lock);
+ sys_mutex_unlock(qd->server->lock);
}
- sys_mutex_unlock(conn->deferred_call_lock);
}
-void qd_container_handle_event(qd_container_t *container, pn_event_t *event, pn_connection_t *pn_conn, qd_connection_t *qd_conn);
-void qd_conn_event_batch_complete(qd_container_t *container, qd_connection_t *qd_conn, bool conn_closed);
+static double normalize_memory_size(const uint64_t bytes, const char** suffix);
+static void* server_thread_run(void* arg);
-static void handle_listener(pn_event_t *e, qd_server_t *qd_server) {
- qd_log_source_t *log = qd_server->log_source;
+void qd_server_run(qd_dispatch_t* qd) {
+ qd_server_t* server = qd->server;
+ int i;
- qd_listener_t *li = (qd_listener_t*) pn_listener_get_context(pn_event_listener(e));
- const char *host_port = li->config.host_port;
- const char *port = li->config.port;
+ assert(server);
+ // Server can't run without a container
+ assert(server->container);
- switch (pn_event_type(e)) {
+ qd_log(server->log_source, QD_LOG_NOTICE, "Operational, %d Threads Running (process ID %ld)", server->thread_count,
+ (long) getpid());
- case PN_LISTENER_OPEN: {
-
- if (strcmp(port, "0") == 0) {
- // If a 0 (zero) is specified for a port, get the actual listening port from the listener.
- pn_listener_t *l = pn_event_listener(e);
- const pn_netaddr_t *na = pn_listener_addr(l);
- char str[PN_MAX_ADDR] = "";
- pn_netaddr_str(na, str, sizeof(str));
- // "str" contains the host and port on which this listener is listening.
- if (li->config.name)
- qd_log(log, QD_LOG_NOTICE, "Listening on %s (%s)", str, li->config.name);
- else
- qd_log(log, QD_LOG_NOTICE, "Listening on %s", str);
- }
- else {
- qd_log(log, QD_LOG_NOTICE, "Listening on %s", host_port);
- }
-
- break;
- }
+ const uintmax_t ram_size = qd_platform_memory_size();
+ const uint64_t vm_size = qd_router_memory_usage();
- case PN_LISTENER_ACCEPT:
- qd_log(log, QD_LOG_TRACE, "Accepting connection on %s", host_port);
- on_accept(e);
- break;
-
- case PN_LISTENER_CLOSE:
- if (li->pn_listener) {
- pn_condition_t *cond = pn_listener_condition(li->pn_listener);
- if (pn_condition_is_set(cond)) {
- qd_log(log, QD_LOG_ERROR, "Listener error on %s: %s (%s)", host_port,
- pn_condition_get_description(cond),
- pn_condition_get_name(cond));
- if (li->exit_on_error) {
- qd_log(log, QD_LOG_CRITICAL, "Shutting down, required listener failed %s",
- host_port);
- exit(1);
- }
- } else {
- qd_log(log, QD_LOG_TRACE, "Listener closed on %s", host_port);
- }
- pn_listener_set_context(li->pn_listener, 0);
- li->pn_listener = 0;
- qd_listener_decref(li);
- }
- break;
+ if (ram_size && vm_size) {
+ const char* suffix_vm = 0;
+ const char* suffix_ram = 0;
+ double vm = normalize_memory_size(vm_size, &suffix_vm);
+ double ram = normalize_memory_size(ram_size, &suffix_ram);
- default:
- break;
+ qd_log(server->log_source, QD_LOG_NOTICE, "Process VmSize %.2f %s (%.2f %s available memory)", vm, suffix_vm,
+ ram, suffix_ram);
}
-}
-
-
-bool qd_connector_has_failover_info(qd_connector_t* ct)
-{
- if (ct && DEQ_SIZE(ct->conn_info_list) > 1)
- return true;
- return false;
-}
-
-
-static void qd_connection_free(qd_connection_t *ctx)
-{
- qd_server_t *qd_server = ctx->server;
- // If this is a dispatch connector, schedule the re-connect timer
- if (ctx->connector) {
- long delay = ctx->connector->delay;
- sys_mutex_lock(ctx->connector->lock);
- ctx->connector->ctx = 0;
- // Increment the connection index by so that we can try connecting to the failover url (if any).
- bool has_failover = qd_connector_has_failover_info(ctx->connector);
+#ifndef NDEBUG
+ qd_log(server->log_source, QD_LOG_INFO, "Running in DEBUG Mode");
+#endif
- if (has_failover) {
- // Go thru the failover list round robin.
- // IMPORTANT: Note here that we set the re-try timer to 1 second.
- // We want to quickly keep cycling thru the failover urls every second.
- delay = 1000;
- }
+ // Start count - 1 threads and use the current thread
- ctx->connector->state = CXTR_STATE_CONNECTING;
- sys_mutex_unlock(ctx->connector->lock);
+ int n = server->thread_count - 1;
+ sys_thread_t** threads = (sys_thread_t**) calloc(n, sizeof(sys_thread_t*));
- //
- // Increment the ref-count to account for the timer's reference to the connector.
- //
- sys_atomic_inc(&ctx->connector->ref_count);
- qd_timer_schedule(ctx->connector->timer, delay);
+ for (i = 0; i < n; i++) {
+ threads[i] = sys_thread(server_thread_run, server);
}
- sys_mutex_lock(qd_server->lock);
- DEQ_REMOVE(qd_server->conn_list, ctx);
+ // Use the current thread
+ server_thread_run(server);
- // If counted for policy enforcement, notify it has closed
- if (ctx->policy_counted) {
- qd_policy_socket_close(qd_server->qd->policy, ctx);
+ for (i = 0; i < n; i++) {
+ sys_thread_join(threads[i]);
+ sys_thread_free(threads[i]);
}
- sys_mutex_unlock(qd_server->lock);
-
- invoke_deferred_calls(ctx, true); // Discard any pending deferred calls
- sys_mutex_free(ctx->deferred_call_lock);
- qd_policy_settings_free(ctx->policy_settings);
- if (ctx->free_user_id) free((char*)ctx->user_id);
- if (ctx->timer) qd_timer_free(ctx->timer);
- free(ctx->name);
- free(ctx->role);
- sys_mutex_lock(qd_server->conn_activation_lock);
- free_qd_connection_t(ctx);
- sys_mutex_unlock(qd_server->conn_activation_lock);
-
- /* Note: pn_conn is freed by the proactor */
-}
+ free(threads);
-static void timeout_on_handhsake(void *context, bool discard)
-{
- if (discard)
- return;
+ // Stop HTTP threads immediately
+ qd_http_server_stop(server->http);
+ qd_http_server_free(server->http);
- qd_connection_t *ctx = (qd_connection_t*) context;
- pn_transport_t *tport = pn_connection_transport(ctx->pn_conn);
- pn_transport_close_head(tport);
- connect_fail(ctx, QD_AMQP_COND_NOT_ALLOWED, "Timeout waiting for initial handshake");
+ qd_log(server->log_source, QD_LOG_NOTICE, "Shut Down");
}
-
-static void startup_timer_handler(void *context)
-{
- //
- // This timer fires for a connection if it has not had a REMOTE_OPEN
- // event in a time interval from the CONNECTION_INIT event. Close
- // down the transport in an IO thread reserved for that connection.
- //
- qd_connection_t *ctx = (qd_connection_t*) context;
- qd_timer_free(ctx->timer);
- ctx->timer = 0;
- qd_connection_invoke_deferred(ctx, timeout_on_handhsake, context);
+void qd_server_stop(qd_dispatch_t* qd) {
+ // Interrupt the proactor. Async-signal safe.
+ pn_proactor_interrupt(qd->server->proactor);
}
-static void qd_increment_conn_index(qd_connection_t *ctx)
-{
- if (ctx->connector) {
- qd_failover_item_t *item = qd_connector_get_conn_info(ctx->connector);
-
- if (item->retries == 1) {
- ctx->connector->conn_index += 1;
- if (ctx->connector->conn_index > DEQ_SIZE(ctx->connector->conn_info_list))
- ctx->connector->conn_index = 1;
- item->retries = 0;
- }
- else
- item->retries += 1;
+void qd_server_activate(qd_connection_t* conn) {
+ if (conn) {
+ conn->wake(conn);
}
-
}
-
-/* Events involving a connection or listener are serialized by the proactor so
- * only one event per connection / listener will be processed at a time.
- */
-static bool handle(qd_server_t *qd_server, pn_event_t *e, pn_connection_t *pn_conn, qd_connection_t *ctx)
-{
- if (pn_conn && qdr_is_authentication_service_connection(pn_conn)) {
- qdr_handle_authentication_service_connection_event(e);
- return true;
- }
-
- switch (pn_event_type(e)) {
-
- case PN_PROACTOR_INTERRUPT:
- /* Interrupt the next thread */
- pn_proactor_interrupt(qd_server->proactor);
- /* Stop the current thread */
- return false;
-
- case PN_PROACTOR_TIMEOUT:
- qd_timer_visit();
- break;
-
- case PN_LISTENER_OPEN:
- case PN_LISTENER_ACCEPT:
- case PN_LISTENER_CLOSE:
- handle_listener(e, qd_server);
- break;
-
- case PN_CONNECTION_INIT: {
- const qd_server_config_t *config = ctx && ctx->listener ? &ctx->listener->config : 0;
- if (config && config->initial_handshake_timeout_seconds > 0) {
- ctx->timer = qd_timer(qd_server->qd, startup_timer_handler, ctx);
- qd_timer_schedule(ctx->timer, config->initial_handshake_timeout_seconds * 1000);
- }
- break;
- }
-
- case PN_CONNECTION_BOUND:
- on_connection_bound(qd_server, e);
- break;
-
- case PN_CONNECTION_REMOTE_OPEN:
- // If we are transitioning to the open state, notify the client via callback.
- if (ctx && ctx->timer) {
- qd_timer_free(ctx->timer);
- ctx->timer = 0;
- }
- if (ctx && !ctx->opened) {
- ctx->opened = true;
- if (ctx->connector) {
- ctx->connector->delay = 2000; // Delay re-connect in case there is a recurring error
- qd_failover_item_t *item = qd_connector_get_conn_info(ctx->connector);
- if (item)
- item->retries = 0;
- }
- }
- break;
-
- case PN_CONNECTION_WAKE:
- invoke_deferred_calls(ctx, false);
- break;
-
- case PN_TRANSPORT_ERROR:
- {
- pn_transport_t *transport = pn_event_transport(e);
- pn_condition_t* condition = transport ? pn_transport_condition(transport) : NULL;
- if (ctx && ctx->connector) { /* Outgoing connection */
- qd_increment_conn_index(ctx);
- const qd_server_config_t *config = &ctx->connector->config;
- ctx->connector->state = CXTR_STATE_FAILED;
- char conn_msg[300];
- if (condition && pn_condition_is_set(condition)) {
- qd_format_string(conn_msg, 300, "[C%"PRIu64"] Connection to %s failed: %s %s", ctx->connection_id, config->host_port,
- pn_condition_get_name(condition), pn_condition_get_description(condition));
- strcpy(ctx->connector->conn_msg, conn_msg);
-
- qd_log(qd_server->log_source, QD_LOG_INFO, conn_msg);
- } else {
- qd_format_string(conn_msg, 300, "[C%"PRIu64"] Connection to %s failed", ctx->connection_id, config->host_port);
- strcpy(ctx->connector->conn_msg, conn_msg);
- qd_log(qd_server->log_source, QD_LOG_INFO, conn_msg);
- }
- } else if (ctx && ctx->listener) { /* Incoming connection */
- if (condition && pn_condition_is_set(condition)) {
- qd_log(ctx->server->log_source, QD_LOG_INFO, "[C%"PRIu64"] Connection from %s (to %s) failed: %s %s",
- ctx->connection_id, ctx->rhost_port, ctx->listener->config.host_port, pn_condition_get_name(condition),
- pn_condition_get_description(condition));
- }
- }
- }
- break;
-
- default:
- break;
- } // Switch event type
-
- if (ctx)
- qd_container_handle_event(qd_server->container, e, pn_conn, ctx);
-
- return true;
+qd_dispatch_t* qd_server_dispatch(qd_server_t* server) {
+ return server->qd;
}
-static void *thread_run(void *arg)
-{
- qd_server_t *qd_server = (qd_server_t*)arg;
- bool running = true;
- while (running) {
- pn_event_batch_t *events = pn_proactor_wait(qd_server->proactor);
- pn_event_t * e;
- qd_connection_t *qd_conn = 0;
- pn_connection_t *pn_conn = 0;
-
- while (running && (e = pn_event_batch_next(events))) {
- pn_connection_t *conn = pn_event_connection(e);
+// Permit replacement by dummy implementation in unit_tests
+__attribute__((noinline)) void qd_server_timeout(qd_server_t* server, qd_duration_t duration) {
+ pn_proactor_set_timeout(server->proactor, duration);
+}
- if (!pn_conn)
- pn_conn = conn;
- assert(pn_conn == conn);
+sys_mutex_t* qd_server_get_activation_lock(qd_server_t* server) {
+ return server->conn_activation_lock;
+}
- if (!qd_conn)
- qd_conn = !!pn_conn ? (qd_connection_t*) pn_connection_get_context(pn_conn) : 0;
+static void connection_wake(qd_connection_t* conn);
+static void connection_decorate(qd_server_t* server, qd_connection_t* conn, const qd_server_config_t* config);
- running = handle(qd_server, e, conn, qd_conn);
+// Construct a new qd_connection. Thread safe.
+qd_connection_t* qd_server_connection(qd_server_t* server, qd_server_config_t* config) {
+ qd_connection_t* conn = new_qd_connection_t();
- /* Free the connection after all other processing is complete */
- if (qd_conn && pn_event_type(e) == PN_TRANSPORT_CLOSED) {
- qd_conn_event_batch_complete(qd_server->container, qd_conn, true);
- pn_connection_set_context(pn_conn, NULL);
- qd_connection_free(qd_conn);
- qd_conn = 0;
- }
- }
-
- //
- // Notify the container that the batch is complete so it can do after-batch
- // processing.
- //
- if (qd_conn)
- qd_conn_event_batch_complete(qd_server->container, qd_conn, false);
-
- pn_proactor_done(qd_server->proactor, events);
+ if (!conn) {
+ return NULL;
}
- return NULL;
-}
+ ZERO(conn);
-static qd_failover_item_t *qd_connector_get_conn_info(qd_connector_t *ct) {
+ conn->pn_conn = pn_connection();
+ conn->deferred_call_lock = sys_mutex();
+ conn->role = strdup(config->role);
- qd_failover_item_t *item = DEQ_HEAD(ct->conn_info_list);
-
- if (DEQ_SIZE(ct->conn_info_list) > 1) {
- for (int i=1; i < ct->conn_index; i++) {
- item = DEQ_NEXT(item);
+ if (!conn->pn_conn || !conn->deferred_call_lock || !conn->role) {
+ if (conn->pn_conn) {
+ pn_connection_free(conn->pn_conn);
}
- }
- return item;
-}
+ if (conn->deferred_call_lock) {
+ sys_mutex_free(conn->deferred_call_lock);
+ }
-/* Timer callback to try/retry connection open */
-static void try_open_lh(qd_connector_t *ct)
-{
- if (ct->state != CXTR_STATE_CONNECTING && ct->state != CXTR_STATE_INIT) {
- /* No longer referenced by pn_connection or timer */
- qd_connector_decref(ct);
- return;
- }
+ free(conn->role);
+ free(conn);
- qd_connection_t *ctx = qd_server_connection(ct->server, &ct->config);
- if (!ctx) { /* Try again later */
- qd_log(ct->server->log_source, QD_LOG_CRITICAL, "Allocation failure connecting to %s",
- ct->config.host_port);
- ct->delay = 10000;
- sys_atomic_inc(&ct->ref_count);
- qd_timer_schedule(ct->timer, ct->delay);
- return;
+ return NULL;
}
- ctx->connector = ct;
- const qd_server_config_t *config = &ct->config;
-
- //
- // Set the hostname on the pn_connection. This hostname will be used by proton as the
- // hostname in the open frame.
- //
+ conn->server = server;
- qd_failover_item_t *item = qd_connector_get_conn_info(ct);
+ // The default. It is overridden for HTTP connections.
+ conn->wake = connection_wake;
- char *current_host = item->host;
- char *host_port = item->host_port;
+ pn_connection_set_context(conn->pn_conn, conn);
- pn_connection_set_hostname(ctx->pn_conn, current_host);
+ DEQ_ITEM_INIT(conn);
+ DEQ_INIT(conn->deferred_calls);
+ DEQ_INIT(conn->free_link_session_list);
- // Set the sasl user name and password on the proton connection object. This has to be
- // done before pn_proactor_connect which will bind a transport to the connection
- if(config->sasl_username)
- pn_connection_set_user(ctx->pn_conn, config->sasl_username);
- if (config->sasl_password)
- pn_connection_set_password(ctx->pn_conn, config->sasl_password);
+ sys_mutex_lock(server->lock);
+ conn->connection_id = server->next_connection_id++;
+ DEQ_INSERT_TAIL(server->conn_list, conn);
+ sys_mutex_unlock(server->lock);
- ctx->connector->state = CXTR_STATE_OPEN;
- ct->ctx = ctx;
- ct->delay = 5000;
+ connection_decorate(conn->server, conn, config);
- qd_log(ct->server->log_source, QD_LOG_TRACE,
- "[%"PRIu64"] Connecting to %s", ctx->connection_id, host_port);
- /* Note: the transport is configured in the PN_CONNECTION_BOUND event */
- pn_proactor_connect(ct->server->proactor, ctx->pn_conn, host_port);
+ return conn;
}
-static void setup_ssl_sasl_and_open(qd_connection_t *ctx)
-{
- qd_connector_t *ct = ctx->connector;
- const qd_server_config_t *config = &ct->config;
- pn_transport_t *tport = pn_connection_transport(ctx->pn_conn);
-
- //
- // Set up SSL if appropriate
- //
- if (config->ssl_profile) {
- pn_ssl_domain_t *domain = pn_ssl_domain(PN_SSL_MODE_CLIENT);
-
- if (!domain) {
- qd_error(QD_ERROR_RUNTIME, "SSL domain failed for connection to %s:%s",
- ct->config.host, ct->config.port);
- return;
- }
+qd_listener_t* qd_server_listener(qd_server_t* server) {
+ qd_listener_t* listener = new_qd_listener_t();
- // set our trusted database for checking the peer's cert:
- if (config->ssl_trusted_certificate_db) {
- if (pn_ssl_domain_set_trusted_ca_db(domain, config->ssl_trusted_certificate_db)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "SSL CA configuration failed for %s:%s",
- ct->config.host, ct->config.port);
- }
- }
- // should we force the peer to provide a cert?
- if (config->ssl_require_peer_authentication) {
- const char *trusted = (config->ssl_trusted_certificates)
- ? config->ssl_trusted_certificates
- : config->ssl_trusted_certificate_db;
- if (pn_ssl_domain_set_peer_authentication(domain,
- PN_SSL_VERIFY_PEER,
- trusted)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "SSL peer auth configuration failed for %s:%s",
- config->host, config->port);
- }
- }
-
- // configure our certificate if the peer requests one:
- if (config->ssl_certificate_file) {
- if (pn_ssl_domain_set_credentials(domain,
- config->ssl_certificate_file,
- config->ssl_private_key_file,
- config->ssl_password)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "SSL local configuration failed for %s:%s",
- config->host, config->port);
- }
- }
+ if (!listener) {
+ return 0;
+ }
- if (config->ssl_ciphers) {
- if (pn_ssl_domain_set_ciphers(domain, config->ssl_ciphers)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "SSL cipher configuration failed for %s:%s",
- config->host, config->port);
- }
- }
+ ZERO(listener);
+ sys_atomic_init(&listener->ref_count, 1);
- if (config->ssl_protocols) {
- if (pn_ssl_domain_set_protocols(domain, config->ssl_protocols)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "Permitted TLS protocols configuration failed %s:%s",
- config->host, config->port);
- }
- }
+ listener->server = server;
+ listener->http = NULL;
- //If ssl is enabled and verify_host_name is true, instruct proton to verify peer name
- if (config->verify_host_name) {
- if (pn_ssl_domain_set_peer_authentication(domain, PN_SSL_VERIFY_PEER_NAME, NULL)) {
- qd_log(ct->server->log_source, QD_LOG_ERROR,
- "SSL peer host name verification failed for %s:%s",
- config->host, config->port);
- }
- }
+ return listener;
+}
- ctx->ssl = pn_ssl(tport);
- pn_ssl_init(ctx->ssl, domain, 0);
- pn_ssl_domain_free(domain);
- }
+static void connector_try_open_handler(void* context);
- //
- // Set up SASL
- //
- sys_mutex_lock(ct->server->lock);
- pn_sasl_t *sasl = pn_sasl(tport);
- if (config->sasl_mechanisms)
- pn_sasl_allowed_mechs(sasl, config->sasl_mechanisms);
- pn_sasl_set_allow_insecure_mechs(sasl, config->allowInsecureAuthentication);
- sys_mutex_unlock(ct->server->lock);
-
- pn_connection_open(ctx->pn_conn);
-}
+qd_connector_t* qd_server_connector(qd_server_t* server) {
+ qd_connector_t* connector = new_qd_connector_t();
-static void try_open_cb(void *context) {
- qd_connector_t *ct = (qd_connector_t*) context;
- if (!qd_connector_decref(ct)) {
- sys_mutex_lock(ct->lock); /* TODO aconway 2017-05-09: this lock looks too big */
- try_open_lh(ct);
- sys_mutex_unlock(ct->lock);
+ if (!connector) {
+ return NULL;
}
-}
+ ZERO(connector);
+ sys_atomic_init(&connector->ref_count, 1);
-qd_server_t *qd_server(qd_dispatch_t *qd, int thread_count, const char *container_name,
- const char *sasl_config_path, const char *sasl_config_name)
-{
- /* Initialize const members, 0 initialize all others. */
- qd_server_t tmp = { .thread_count = thread_count };
- qd_server_t *qd_server = NEW(qd_server_t);
- if (qd_server == 0)
- return 0;
- memcpy(qd_server, &tmp, sizeof(tmp));
-
- qd_server->qd = qd;
- qd_server->log_source = qd_log_source("SERVER");
- qd_server->protocol_log_source = qd_log_source("PROTOCOL");
- qd_server->container_name = container_name;
- qd_server->sasl_config_path = sasl_config_path;
- qd_server->sasl_config_name = sasl_config_name;
- qd_server->proactor = pn_proactor();
- qd_server->container = 0;
- qd_server->start_context = 0;
- qd_server->lock = sys_mutex();
- qd_server->conn_activation_lock = sys_mutex();
- qd_server->cond = sys_cond();
- DEQ_INIT(qd_server->conn_list);
-
- qd_timer_initialize(qd_server->lock);
-
- qd_server->pause_requests = 0;
- qd_server->threads_paused = 0;
- qd_server->pause_next_sequence = 0;
- qd_server->pause_now_serving = 0;
- qd_server->next_connection_id = 1;
- qd_server->py_displayname_obj = 0;
-
- qd_server->http = qd_http_server(qd_server, qd_server->log_source);
-
- qd_log(qd_server->log_source, QD_LOG_INFO, "Container Name: %s", qd_server->container_name);
-
- return qd_server;
-}
+ connector->server = server;
+ qd_failover_item_list_t conn_info_list;
+ DEQ_INIT(conn_info_list);
+ connector->conn_info_list = conn_info_list;
+ connector->conn_index = 1;
+ connector->state = CXTR_STATE_INIT;
+ connector->lock = sys_mutex();
+ connector->timer = qd_timer(connector->server->qd, connector_try_open_handler, connector);
+ connector->conn_msg = (char*) malloc(300);
+ memset(connector->conn_msg, 0, 300);
-void qd_server_free(qd_server_t *qd_server)
-{
- if (!qd_server) return;
- qd_connection_t *ctx = DEQ_HEAD(qd_server->conn_list);
- while (ctx) {
- qd_log(qd_server->log_source, QD_LOG_INFO,
- "[C%"PRIu64"] Closing connection on shutdown",
- ctx->connection_id);
- DEQ_REMOVE_HEAD(qd_server->conn_list);
- if (ctx->pn_conn) {
- pn_transport_t *tport = pn_connection_transport(ctx->pn_conn);
- if (tport)
- pn_transport_set_context(tport, 0); /* for transport_tracer */
- qd_session_cleanup(ctx);
- pn_connection_set_context(ctx->pn_conn, 0);
- }
- if (ctx->free_user_id) free((char*)ctx->user_id);
- sys_mutex_free(ctx->deferred_call_lock);
- free(ctx->name);
- free(ctx->role);
- if (ctx->policy_settings)
- free_qd_policy_settings_t(ctx->policy_settings);
- free_qd_connection_t(ctx);
- ctx = DEQ_HEAD(qd_server->conn_list);
+ if (!connector->lock || !connector->timer) {
+ qd_connector_decref(connector);
+ return NULL;
}
- pn_proactor_free(qd_server->proactor);
- qd_timer_finalize();
- sys_mutex_free(qd_server->lock);
- sys_mutex_free(qd_server->conn_activation_lock);
- sys_cond_free(qd_server->cond);
- Py_XDECREF((PyObject *)qd_server->py_displayname_obj);
- free(qd_server);
-}
-void qd_server_set_container(qd_dispatch_t *qd, qd_container_t *container)
-{
- qd->server->container = container;
+ return connector;
}
-void qd_server_trace_all_connections()
-{
- qd_dispatch_t *qd = qd_dispatch_get_dispatch();
- if (qd->server) {
- sys_mutex_lock(qd->server->lock);
- qd_connection_list_t conn_list = qd->server->conn_list;
- qd_connection_t *conn = DEQ_HEAD(conn_list);
- while(conn) {
- //
- // If there is already a tracer on the transport, nothing to do, move on to the next connection.
- //
- pn_transport_t *tport = pn_connection_transport(conn->pn_conn);
- if (! pn_transport_get_tracer(tport)) {
- pn_transport_trace(tport, PN_TRACE_FRM);
- pn_transport_set_tracer(tport, transport_tracer);
- }
- conn = DEQ_NEXT(conn);
- }
- sys_mutex_unlock(qd->server->lock);
- }
-}
+static double normalize_memory_size(const uint64_t bytes, const char** suffix) {
+ static const char* const units[] = {"B", "KiB", "MiB", "GiB", "TiB"};
+ const int units_ct = 5;
+ const double base = 1024.0;
+ double value = (double) bytes;
-
-static double normalize_memory_size(const uint64_t bytes, const char **suffix)
-{
- static const char * const units[] = {"B", "KiB", "MiB", "GiB", "TiB"};
- const int units_ct = 5;
- const double base = 1024.0;
-
- double value = (double)bytes;
for (int i = 0; i < units_ct; ++i) {
if (value < base) {
- if (suffix)
+ if (suffix) {
*suffix = units[i];
+ }
+
return value;
}
+
value /= base;
}
- if (suffix)
+
+ if (suffix) {
*suffix = units[units_ct - 1];
+ }
+
return value;
}
-void qd_server_run(qd_dispatch_t *qd)
-{
- qd_server_t *qd_server = qd->server;
- int i;
- assert(qd_server);
- assert(qd_server->container); // Server can't run without a container
- qd_log(qd_server->log_source,
- QD_LOG_NOTICE, "Operational, %d Threads Running (process ID %ld)",
- qd_server->thread_count, (long)getpid());
+void qd_conn_event_batch_complete(qd_container_t* container, qd_connection_t* conn, bool conn_closed);
- const uintmax_t ram_size = qd_platform_memory_size();
- const uint64_t vm_size = qd_router_memory_usage();
- if (ram_size && vm_size) {
- const char *suffix_vm = 0;
- const char *suffix_ram = 0;
- double vm = normalize_memory_size(vm_size, &suffix_vm);
- double ram = normalize_memory_size(ram_size, &suffix_ram);
- qd_log(qd_server->log_source, QD_LOG_NOTICE,
- "Process VmSize %.2f %s (%.2f %s available memory)",
- vm, suffix_vm, ram, suffix_ram);
- }
-
-#ifndef NDEBUG
- qd_log(qd_server->log_source, QD_LOG_INFO, "Running in DEBUG Mode");
-#endif
- int n = qd_server->thread_count - 1; /* Start count-1 threads + use current thread */
- sys_thread_t **threads = (sys_thread_t **)calloc(n, sizeof(sys_thread_t*));
- for (i = 0; i < n; i++) {
- threads[i] = sys_thread(thread_run, qd_server);
- }
- thread_run(qd_server); /* Use the current thread */
- for (i = 0; i < n; i++) {
- sys_thread_join(threads[i]);
- sys_thread_free(threads[i]);
- }
- free(threads);
- qd_http_server_stop(qd_server->http); /* Stop HTTP threads immediately */
- qd_http_server_free(qd_server->http);
+static void* server_thread_run(void* arg) {
+ qd_server_t* server = (qd_server_t*) arg;
+ pn_proactor_t* proactor = server->proactor;
+ bool running = true;
- qd_log(qd_server->log_source, QD_LOG_NOTICE, "Shut Down");
-}
+ while (running) {
+ pn_event_batch_t* batch = pn_proactor_wait(proactor);
+ pn_event_t* event = pn_event_batch_next(batch);
+ pn_connection_t* pn_conn = pn_event_connection(event);
+ do {
+ running = qd_server_handle_event(server, event);
+ } while ((event = pn_event_batch_next(batch)));
-void qd_server_stop(qd_dispatch_t *qd)
-{
- /* Interrupt the proactor, async-signal-safe */
- pn_proactor_interrupt(qd->server->proactor);
-}
+ if (pn_conn) {
+ qd_connection_t* conn = (qd_connection_t*) pn_connection_get_context(pn_conn);
-void qd_server_activate(qd_connection_t *ctx)
-{
- if (ctx) ctx->wake(ctx);
-}
+ if (conn) {
+ qd_conn_event_batch_complete(server->container, conn, false);
+ }
+ }
+ pn_proactor_done(proactor, batch);
+ }
-void qd_connection_set_context(qd_connection_t *conn, void *context)
-{
- conn->user_context = context;
+ return NULL;
}
+// Wake function for proactor-managed connections
+static void connection_wake(qd_connection_t* conn) {
+ assert(conn);
-void *qd_connection_get_context(qd_connection_t *conn)
-{
- return conn->user_context;
+ pn_connection_wake(conn->pn_conn);
}
-
-void *qd_connection_get_config_context(qd_connection_t *conn)
-{
- return conn->context;
+static void data_put_symbol(pn_data_t* data, const char* value) {
+ pn_data_put_symbol(data, pn_bytes(strlen(value), value));
}
+static void data_put_string(pn_data_t* data, const char* value) {
+ pn_data_put_string(data, pn_bytes(strlen(value), value));
+}
-void qd_connection_set_link_context(qd_connection_t *conn, void *context)
-{
- conn->link_context = context;
+static void data_put_map_entry(pn_data_t* data, const char* key, const char* value) {
+ data_put_symbol(data, key);
+ data_put_string(data, value);
}
+static void connection_decorate(qd_server_t* server, qd_connection_t* conn, const qd_server_config_t* config) {
+ pn_connection_t* pn_conn = conn->pn_conn;
-void *qd_connection_get_link_context(qd_connection_t *conn)
-{
- return conn->link_context;
-}
+ // Set the container name
+ pn_connection_set_container(pn_conn, server->container_name);
+ // Offer ANONYMOUS_RELAY capability
+ size_t clen = strlen(QD_CAPABILITY_ANONYMOUS_RELAY);
+ pn_data_put_symbol(pn_connection_offered_capabilities(pn_conn),
+ pn_bytes(clen, (char*) QD_CAPABILITY_ANONYMOUS_RELAY));
-pn_connection_t *qd_connection_pn(qd_connection_t *conn)
-{
- return conn->pn_conn;
-}
+ // Create the connection properties map
+ pn_data_t* props = pn_connection_properties(pn_conn);
-bool qd_connection_inbound(qd_connection_t *conn)
-{
- return conn->listener != 0;
-}
+ pn_data_put_map(props);
+ pn_data_enter(props);
+ data_put_map_entry(props, QD_CONNECTION_PROPERTY_PRODUCT_KEY, QD_CONNECTION_PROPERTY_PRODUCT_VALUE);
+ data_put_map_entry(props, QD_CONNECTION_PROPERTY_VERSION_KEY, QPID_DISPATCH_VERSION);
-uint64_t qd_connection_connection_id(qd_connection_t *conn)
-{
- return conn->connection_id;
-}
+ data_put_symbol(props, QD_CONNECTION_PROPERTY_CONN_ID);
+ pn_data_put_int(props, conn->connection_id);
+ // XXX When is config going to be null?
+ if (config && config->inter_router_cost > 1) {
Review comment:
Agreed - if coverity ever encounters a path through the code where a pointer is checked for null, then it flags any use of that pointer that isn't checked for null as a possible error.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org