src/adaptors/http_common.c - qpid-dispatch - Git at Google

 /*
  * 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 "http_common.h"

 #include <proton/listener.h>

 #include <stdio.h>

 ALLOC_DECLARE(qd_http_listener_t);
 ALLOC_DEFINE(qd_http_listener_t);
 ALLOC_DECLARE(qd_http_connector_t);
 ALLOC_DEFINE(qd_http_connector_t);


 static qd_error_t load_bridge_config(qd_dispatch_t *qd, qd_http_bridge_config_t *config, qd_entity_t* entity)
 {
     char *version_str = 0;
     char *aggregation_str = 0;

     qd_error_clear();
     ZERO(config);

 #define CHECK() if (qd_error_code()) goto error
     config->name    = qd_entity_get_string(entity, "name");            CHECK();
     config->host    = qd_entity_get_string(entity, "host");            CHECK();
     config->port    = qd_entity_get_string(entity, "port");            CHECK();
     config->address = qd_entity_get_string(entity, "address");         CHECK();
     config->site    = qd_entity_opt_string(entity, "siteId", 0);       CHECK();
     version_str     = qd_entity_get_string(entity, "protocolVersion");  CHECK();
     aggregation_str = qd_entity_opt_string(entity, "aggregation", 0);  CHECK();
     config->event_channel = qd_entity_opt_bool(entity, "eventChannel", false); CHECK();
     config->host_override  = qd_entity_opt_string(entity, "hostOverride", 0);   CHECK();

     if (strcmp(version_str, "HTTP2") == 0) {
         config->version = VERSION_HTTP2;
     } else {
         config->version = VERSION_HTTP1;
     }
     free(version_str);
     version_str = 0;

     if (aggregation_str && strcmp(aggregation_str, "json") == 0) {
         config->aggregation = QD_AGGREGATION_JSON;
     } else if (aggregation_str && strcmp(aggregation_str, "multipart") == 0) {
         config->aggregation = QD_AGGREGATION_MULTIPART;
     } else {
         config->aggregation = QD_AGGREGATION_NONE;
     }
     free(aggregation_str);
     aggregation_str = 0;

     int hplen = strlen(config->host) + strlen(config->port) + 2;
     config->host_port = malloc(hplen);
     snprintf(config->host_port, hplen, "%s:%s", config->host, config->port);

     return QD_ERROR_NONE;

 error:
     qd_http_free_bridge_config(config);
     free(version_str);
     return qd_error_code();
 }


 void qd_http_free_bridge_config(qd_http_bridge_config_t *config)
 {
     if (!config) {
         return;
     }
     free(config->host);
     free(config->port);
     free(config->name);
     free(config->address);
     free(config->site);
     free(config->host_override);
     free(config->host_port);
 }


 //
 // HTTP Listener Management (HttpListenerEntity)
 //


 qd_http_listener_t *qd_dispatch_configure_http_listener(qd_dispatch_t *qd, qd_entity_t *entity)
 {
     qd_http_listener_t *listener = 0;
     qd_http_bridge_config_t config;

     if (load_bridge_config(qd, &config, entity) != QD_ERROR_NONE) {
         qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_ERROR,
                "Unable to create http listener: %s", qd_error_message());
         return 0;
     }

     switch (config.version) {
     case VERSION_HTTP1:
         listener = qd_http1_configure_listener(qd, &config, entity);
         break;
     case VERSION_HTTP2:
         listener = qd_http2_configure_listener(qd, &config, entity);
         break;
     }

     if (!listener)
         qd_http_free_bridge_config(&config);

     return listener;
 }


 void qd_dispatch_delete_http_listener(qd_dispatch_t *qd, void *impl)
 {
     qd_http_listener_t *listener = (qd_http_listener_t*) impl;
     if (listener) {
         switch (listener->config.version) {
         case VERSION_HTTP1:
             qd_http1_delete_listener(qd, listener);
             break;
         case VERSION_HTTP2:
             qd_http2_delete_listener(qd, listener);
             break;
         }
     }
 }


 qd_error_t qd_entity_refresh_httpListener(qd_entity_t* entity, void *impl)
 {
     return QD_ERROR_NONE;
 }


 //
 // HTTP Connector Management (HttpConnectorEntity)
 //


 qd_http_connector_t *qd_dispatch_configure_http_connector(qd_dispatch_t *qd, qd_entity_t *entity)
 {
     qd_http_connector_t *conn = 0;
     qd_http_bridge_config_t config;

     if (load_bridge_config(qd, &config, entity) != QD_ERROR_NONE) {
         qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_ERROR,
                "Unable to create http connector: %s", qd_error_message());
         return 0;
     }

     switch (config.version) {
     case VERSION_HTTP1:
         conn = qd_http1_configure_connector(qd, &config, entity);
         break;
     case VERSION_HTTP2:
         conn = qd_http2_configure_connector(qd, &config, entity);
         break;
     }

     if (!conn)
         qd_http_free_bridge_config(&config);

     return conn;
 }


 void qd_dispatch_delete_http_connector(qd_dispatch_t *qd, void *impl)
 {
     qd_http_connector_t *conn = (qd_http_connector_t*) impl;

     if (conn) {
         switch (conn->config.version) {
         case VERSION_HTTP1:
             qd_http1_delete_connector(qd, conn);
             break;
         case VERSION_HTTP2:
             qd_http2_delete_connector(qd, conn);
             break;
         }
     }
 }

 qd_error_t qd_entity_refresh_httpConnector(qd_entity_t* entity, void *impl)
 {
     return QD_ERROR_NONE;
 }

 //
 // qd_http_listener_t constructor
 //

 qd_http_listener_t *qd_http_listener(qd_server_t *server, qd_server_event_handler_t handler)
 {
     qd_http_listener_t *li = new_qd_http_listener_t();
     if (!li)
         return 0;
     ZERO(li);

     li->pn_listener = pn_listener();
     if (!li->pn_listener) {
         free_qd_http_listener_t(li);
         return 0;
     }

     sys_atomic_init(&li->ref_count, 1);
     li->server = server;
     li->context.context = li;
     li->context.handler = handler;
     pn_listener_set_context(li->pn_listener, &li->context);

     return li;
 }

 void qd_http_listener_decref(qd_http_listener_t* li)
 {
     if (li && sys_atomic_dec(&li->ref_count) == 1) {
         qd_http_free_bridge_config(&li->config);
         free_qd_http_listener_t(li);
     }
 }

 //
 // qd_http_connector_t constructor
 //

 qd_http_connector_t *qd_http_connector(qd_server_t *server)
 {
     qd_http_connector_t *c = new_qd_http_connector_t();
     if (!c) return 0;
     ZERO(c);
     sys_atomic_init(&c->ref_count, 1);
     c->server      = server;
     return c;
 }

 void qd_http_connector_decref(qd_http_connector_t* c)
 {
     if (c && sys_atomic_dec(&c->ref_count) == 1) {
         qd_http_free_bridge_config(&c->config);
         free_qd_http_connector_t(c);
     }
 }


 typedef struct qdr_http_method_status_t  qdr_http_method_status_t;

 struct qdr_http_method_status_t {
     DEQ_LINKS(qdr_http_method_status_t);

     char     *key;
     uint64_t requests;
 };
 DEQ_DECLARE(qdr_http_method_status_t, qdr_http_method_status_list_t);

 typedef struct qdr_http_request_info_t  qdr_http_request_info_t;

 struct qdr_http_request_info_t {
     DEQ_LINKS(qdr_http_request_info_t);

     char     *key;
     char     *address;
     char     *host;
     char     *site;
     bool      ingress;
     uint64_t  requests;
     uint64_t  bytes_in;
     uint64_t  bytes_out;
     uint64_t  max_latency;
     qdr_http_method_status_list_t detail;
 };
 DEQ_DECLARE(qdr_http_request_info_t, qdr_http_request_info_list_t);

 #define QDR_HTTP_REQUEST_INFO_NAME                   0
 #define QDR_HTTP_REQUEST_INFO_IDENTITY               1
 #define QDR_HTTP_REQUEST_INFO_ADDRESS                2
 #define QDR_HTTP_REQUEST_INFO_HOST                   3
 #define QDR_HTTP_REQUEST_INFO_SITE                   4
 #define QDR_HTTP_REQUEST_INFO_DIRECTION              5
 #define QDR_HTTP_REQUEST_INFO_REQUESTS               6
 #define QDR_HTTP_REQUEST_INFO_BYTES_IN               7
 #define QDR_HTTP_REQUEST_INFO_BYTES_OUT              8
 #define QDR_HTTP_REQUEST_INFO_MAX_LATENCY            9
 #define QDR_HTTP_REQUEST_INFO_DETAIL                10


 const char * const QDR_HTTP_REQUEST_INFO_DIRECTION_IN  = "in";
 const char * const QDR_HTTP_REQUEST_INFO_DIRECTION_OUT = "out";

 const char *qdr_http_request_info_columns[] =
     {"name",
      "identity",
      "address",
      "host",
      "site",
      "direction",
      "requests",
      "bytesIn",
      "bytesOut",
      "maxLatency",
      "details",
      0};

 const char *HTTP_REQUEST_INFO_TYPE = "org.apache.qpid.dispatch.httpRequestInfo";

 typedef struct {
     qdr_http_request_info_list_t records;
 } http_request_info_records_t;

 static http_request_info_records_t* request_info = 0;

 static http_request_info_records_t *_get_request_info()
 {
     if (!request_info) {
         request_info = NEW(http_request_info_records_t);
         DEQ_INIT(request_info->records);
     }
     return request_info;
 }

 static void insert_column(qdr_core_t *core, qdr_http_request_info_t *record, int col, qd_composed_field_t *body)
 {
     qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Insert column %i for %p", col, (void*) record);

     if (!record)
         return;

     switch(col) {
     case QDR_HTTP_REQUEST_INFO_NAME:
         qd_compose_insert_string(body, record->key);
         break;

     case QDR_HTTP_REQUEST_INFO_IDENTITY: {
         qd_compose_insert_string(body, record->key);
         break;
     }

     case QDR_HTTP_REQUEST_INFO_ADDRESS:
         qd_compose_insert_string(body, record->address);
         break;

     case QDR_HTTP_REQUEST_INFO_HOST:
         qd_compose_insert_string(body, record->host);
         break;

     case QDR_HTTP_REQUEST_INFO_SITE:
         qd_compose_insert_string(body, record->site);
         break;

     case QDR_HTTP_REQUEST_INFO_DIRECTION:
         if (record->ingress)
             qd_compose_insert_string(body, QDR_HTTP_REQUEST_INFO_DIRECTION_IN);
         else
             qd_compose_insert_string(body, QDR_HTTP_REQUEST_INFO_DIRECTION_OUT);
         break;

     case QDR_HTTP_REQUEST_INFO_REQUESTS:
         qd_compose_insert_uint(body, record->requests);
         break;

     case QDR_HTTP_REQUEST_INFO_BYTES_IN:
         qd_compose_insert_uint(body, record->bytes_in);
         break;

     case QDR_HTTP_REQUEST_INFO_BYTES_OUT:
         qd_compose_insert_uint(body, record->bytes_out);
         break;

     case QDR_HTTP_REQUEST_INFO_MAX_LATENCY:
         qd_compose_insert_uint(body, record->max_latency);
         break;

     case QDR_HTTP_REQUEST_INFO_DETAIL:
         qd_compose_start_map(body);
         for (qdr_http_method_status_t *item = DEQ_HEAD(record->detail); item; item = DEQ_NEXT(item)) {
             qd_compose_insert_string(body, item->key);
             qd_compose_insert_int(body, item->requests);
         }
         qd_compose_end_map(body);
         break;

     }
 }


 static void write_list(qdr_core_t *core, qdr_query_t *query,  qdr_http_request_info_t *record)
 {
     qd_composed_field_t *body = query->body;

     qd_compose_start_list(body);

     if (record) {
         int i = 0;
         while (query->columns[i] >= 0) {
             insert_column(core, record, query->columns[i], body);
             i++;
         }
     }
     qd_compose_end_list(body);
 }

 static void write_map(qdr_core_t           *core,
                       qdr_http_request_info_t *record,
                       qd_composed_field_t  *body,
                       const char           *qdr_connection_columns[])
 {
     qd_compose_start_map(body);

     for(int i = 0; i < QDR_HTTP_REQUEST_INFO_COLUMN_COUNT; i++) {
         qd_compose_insert_string(body, qdr_connection_columns[i]);
         insert_column(core, record, i, body);
     }

     qd_compose_end_map(body);
 }

 static void advance(qdr_query_t *query, qdr_http_request_info_t *record)
 {
     if (record) {
         query->next_offset++;
         record = DEQ_NEXT(record);
         query->more = !!record;
     }
     else {
         query->more = false;
     }
 }

 static qdr_http_request_info_t *find_by_identity(qdr_core_t *core, qd_iterator_t *identity)
 {
     if (!identity)
         return 0;

     qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records);
     while (record) {
         // Convert the passed in identity to a char*
         if (qd_iterator_equal(identity, (const unsigned char*) record->key))
             break;
         record = DEQ_NEXT(record);
     }

     return record;

 }

 void qdra_http_request_info_get_first_CT(qdr_core_t *core, qdr_query_t *query, int offset)
 {
     qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "query for first http request info (%i)", offset);
     query->status = QD_AMQP_OK;

     if (offset >= DEQ_SIZE(_get_request_info()->records)) {
         query->more = false;
         qdr_agent_enqueue_response_CT(core, query);
         return;
     }

     qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records);
     for (int i = 0; i < offset && record; i++)
         record = DEQ_NEXT(record);
     assert(record);

     if (record) {
         write_list(core, query, record);
         query->next_offset = offset;
         advance(query, record);
     } else {
         query->more = false;
     }

     qdr_agent_enqueue_response_CT(core, query);
 }

 void qdra_http_request_info_get_next_CT(qdr_core_t *core, qdr_query_t *query)
 {
     qdr_http_request_info_t *record = 0;

     if (query->next_offset < DEQ_SIZE(_get_request_info()->records)) {
         record = DEQ_HEAD(_get_request_info()->records);
         for (int i = 0; i < query->next_offset && record; i++)
             record = DEQ_NEXT(record);
     }

     if (record) {
         write_list(core, query, record);
         advance(query, record);
     } else {
         query->more = false;
     }
     qdr_agent_enqueue_response_CT(core, query);
 }

 void qdra_http_request_info_get_CT(qdr_core_t          *core,
                                qd_iterator_t       *name,
                                qd_iterator_t       *identity,
                                qdr_query_t         *query,
                                const char          *qdr_http_request_info_columns[])
 {
     qdr_http_request_info_t *record = 0;

     if (!identity) {
         query->status = QD_AMQP_BAD_REQUEST;
         query->status.description = "Name not supported. Identity required";
         qd_log(core->agent_log, QD_LOG_ERROR, "Error performing READ of %s: %s", HTTP_REQUEST_INFO_TYPE, query->status.description);
     } else {
         record = find_by_identity(core, identity);

         if (record == 0) {
             query->status = QD_AMQP_NOT_FOUND;
         } else {
             write_map(core, record, query->body, qdr_http_request_info_columns);
             query->status = QD_AMQP_OK;
         }
     }
     qdr_agent_enqueue_response_CT(core, query);
 }

 static const char* UNKNOWN = "unknown";

 static qdr_http_method_status_t* _new_qdr_http_method_status_t(const char *const method, int status)
 {
     qdr_http_method_status_t* record = NEW(qdr_http_method_status_t);
     ZERO(record);

     if (status >= 600 || status < 100) {
         status = 500;
     }
     if (method) {
         size_t key_len = strlen(method) + 5;
         record->key = malloc(key_len);
         snprintf(record->key, key_len, "%s:%03i", method, status);
     } else {
         record->key = qd_strdup(UNKNOWN);
     }

     return record;
 }

 static void _free_qdr_http_method_status(qdr_http_method_status_t* record)
 {
     free(record->key);
     free(record);
 }

 static void _free_qdr_http_request_info(qdr_http_request_info_t* record)
 {
     if (record->key) {
         free(record->key);
     }
     if (record->address) {
         free(record->address);
     }
     if (record->host) {
         free(record->host);
     }
     if (record->site) {
         free(record->site);
     }
     for (qdr_http_method_status_t *item = DEQ_HEAD(record->detail); item; item = DEQ_HEAD(record->detail)) {
         _free_qdr_http_method_status(item);
     }
     free(record);
 }

 static void _update_http_method_status_detail(qdr_http_method_status_list_t *detail, qdr_http_method_status_t *addition)
 {
     bool updated = false;
     for (qdr_http_method_status_t *item = DEQ_HEAD(*detail); item && !updated; item = DEQ_NEXT(item)) {
         if (strcmp(item->key, addition->key) == 0) {
             item->requests += addition->requests;
             _free_qdr_http_method_status(addition);
             updated = true;
         }
     }
     if (!updated) {
         DEQ_INSERT_TAIL(*detail, addition);
     }
 }

 static bool _update_qdr_http_request_info(qdr_http_request_info_t* record, qdr_http_request_info_t* additions)
 {
     if (strcmp(record->key, additions->key) == 0) {
         record->requests += additions->requests;
         record->bytes_in += additions->bytes_in;
         record->bytes_out += additions->bytes_out;
         if (additions->max_latency > record->max_latency) {
             record->max_latency = additions->max_latency;
         }
         for (qdr_http_method_status_t *item = DEQ_HEAD(additions->detail); item; item = DEQ_HEAD(additions->detail)) {
             DEQ_REMOVE_HEAD(additions->detail);
             _update_http_method_status_detail(&record->detail, item);
         }
         return true;
     } else {
         return false;
     }
 }

 static void _add_http_request_info_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
 {
     qdr_http_request_info_t *update = (qdr_http_request_info_t*) action->args.general.context_1;
     bool updated = false;
     for (qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records); record && !updated; record = DEQ_NEXT(record)) {
         if (_update_qdr_http_request_info(record, update)) {
             updated = true;
             _free_qdr_http_request_info(update);
             qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Updated http request info %s", record->key);
         }
     }
     if (!updated) {
         DEQ_INSERT_TAIL(_get_request_info()->records, update);
         qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Added http request info %s (%zu)", update->key, DEQ_SIZE(_get_request_info()->records));
     }
 }

 static void _add_http_request_info(qdr_core_t *core, qdr_http_request_info_t* record)
 {
     qdr_action_t *action = qdr_action(_add_http_request_info_CT, "add_http_request_info");
     action->args.general.context_1 = record;
     qdr_action_enqueue(core, action);
 }

 static qdr_http_request_info_t* _new_qdr_http_request_info_t()
 {
     qdr_http_request_info_t* record = NEW(qdr_http_request_info_t);
     ZERO(record);
     DEQ_INIT(record->detail);
     return record;
 }

 static char *_record_key(const char *host, const char *address, const char* site, bool ingress)
 {
     if (!host) return 0;
     size_t hostlen = strlen(host);
     size_t addresslen = address ? strlen(address) + 1 : 0;
     size_t sitelen = site ? strlen(site) + 1 : 0;
     char *key = malloc(hostlen + addresslen + sitelen + 3);
     size_t i = 0;
     key[i++] = ingress ? 'i' : 'o';
     key[i++] = '_';
     strcpy(key+i, host);
     i += hostlen;
     if (address) {
         key[i++] = '_';
         strcpy(key+i, address);
         i += (addresslen-1);
     }
     if (site) {
         key[i++] = '@';
         strcpy(key+i, site);
     }
     return key;
 }

 void qd_http_record_request(qdr_core_t *core, const char * method, uint32_t status_code, const char *address, const char *host,
                              const char *local_site, const char *remote_site, bool ingress,
                              uint64_t bytes_in, uint64_t bytes_out, uint64_t latency)
 {
     qdr_http_request_info_t* record = _new_qdr_http_request_info_t();
     record->ingress = ingress;
     record->address = address ? qd_strdup(address) : 0;
     record->host = host ? qd_strdup(host) : 0;
     record->site = remote_site ? qd_strdup(remote_site) : 0;
     record->key = _record_key(record->host, record->address, remote_site, record->ingress);
     record->requests = 1;
     record->bytes_in = bytes_in;
     record->bytes_out = bytes_out;
     record->max_latency = latency;

     qdr_http_method_status_t *detail = _new_qdr_http_method_status_t(method, (int) status_code);
     detail->requests = 1;
     DEQ_INSERT_TAIL(record->detail, detail);

     qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Adding http request info %s", record->key);
     _add_http_request_info(core, record);
 }

 char *qd_get_host_from_host_port(const char *host_port)
 {
     char *end = strchr(host_port, ':');
     if (end == NULL) {
         return 0;
     } else {
         size_t len = end - host_port;
         char *host = malloc(len + 1);
         strncpy(host, host_port, len);
         host[len] = '\0';
         return host;
     }
 }
	/*
	* 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 "http_common.h"

	#include <proton/listener.h>

	#include <stdio.h>

	ALLOC_DECLARE(qd_http_listener_t);
	ALLOC_DEFINE(qd_http_listener_t);
	ALLOC_DECLARE(qd_http_connector_t);
	ALLOC_DEFINE(qd_http_connector_t);


	static qd_error_t load_bridge_config(qd_dispatch_t qd, qd_http_bridge_config_t config, qd_entity_t* entity)
	{
	char *version_str = 0;
	char *aggregation_str = 0;

	qd_error_clear();
	ZERO(config);

	#define CHECK() if (qd_error_code()) goto error
	config->name = qd_entity_get_string(entity, "name"); CHECK();
	config->host = qd_entity_get_string(entity, "host"); CHECK();
	config->port = qd_entity_get_string(entity, "port"); CHECK();
	config->address = qd_entity_get_string(entity, "address"); CHECK();
	config->site = qd_entity_opt_string(entity, "siteId", 0); CHECK();
	version_str = qd_entity_get_string(entity, "protocolVersion"); CHECK();
	aggregation_str = qd_entity_opt_string(entity, "aggregation", 0); CHECK();
	config->event_channel = qd_entity_opt_bool(entity, "eventChannel", false); CHECK();
	config->host_override = qd_entity_opt_string(entity, "hostOverride", 0); CHECK();

	if (strcmp(version_str, "HTTP2") == 0) {
	config->version = VERSION_HTTP2;
	} else {
	config->version = VERSION_HTTP1;
	}
	free(version_str);
	version_str = 0;

	if (aggregation_str && strcmp(aggregation_str, "json") == 0) {
	config->aggregation = QD_AGGREGATION_JSON;
	} else if (aggregation_str && strcmp(aggregation_str, "multipart") == 0) {
	config->aggregation = QD_AGGREGATION_MULTIPART;
	} else {
	config->aggregation = QD_AGGREGATION_NONE;
	}
	free(aggregation_str);
	aggregation_str = 0;

	int hplen = strlen(config->host) + strlen(config->port) + 2;
	config->host_port = malloc(hplen);
	snprintf(config->host_port, hplen, "%s:%s", config->host, config->port);

	return QD_ERROR_NONE;

	error:
	qd_http_free_bridge_config(config);
	free(version_str);
	return qd_error_code();
	}


	void qd_http_free_bridge_config(qd_http_bridge_config_t *config)
	{
	if (!config) {
	return;
	}
	free(config->host);
	free(config->port);
	free(config->name);
	free(config->address);
	free(config->site);
	free(config->host_override);
	free(config->host_port);
	}


	//
	// HTTP Listener Management (HttpListenerEntity)
	//


	qd_http_listener_t qd_dispatch_configure_http_listener(qd_dispatch_t qd, qd_entity_t *entity)
	{
	qd_http_listener_t *listener = 0;
	qd_http_bridge_config_t config;

	if (load_bridge_config(qd, &config, entity) != QD_ERROR_NONE) {
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_ERROR,
	"Unable to create http listener: %s", qd_error_message());
	return 0;
	}

	switch (config.version) {
	case VERSION_HTTP1:
	listener = qd_http1_configure_listener(qd, &config, entity);
	break;
	case VERSION_HTTP2:
	listener = qd_http2_configure_listener(qd, &config, entity);
	break;
	}

	if (!listener)
	qd_http_free_bridge_config(&config);

	return listener;
	}


	void qd_dispatch_delete_http_listener(qd_dispatch_t qd, void impl)
	{
	qd_http_listener_t listener = (qd_http_listener_t) impl;
	if (listener) {
	switch (listener->config.version) {
	case VERSION_HTTP1:
	qd_http1_delete_listener(qd, listener);
	break;
	case VERSION_HTTP2:
	qd_http2_delete_listener(qd, listener);
	break;
	}
	}
	}


	qd_error_t qd_entity_refresh_httpListener(qd_entity_t* entity, void *impl)
	{
	return QD_ERROR_NONE;
	}


	//
	// HTTP Connector Management (HttpConnectorEntity)
	//


	qd_http_connector_t qd_dispatch_configure_http_connector(qd_dispatch_t qd, qd_entity_t *entity)
	{
	qd_http_connector_t *conn = 0;
	qd_http_bridge_config_t config;

	if (load_bridge_config(qd, &config, entity) != QD_ERROR_NONE) {
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_ERROR,
	"Unable to create http connector: %s", qd_error_message());
	return 0;
	}

	switch (config.version) {
	case VERSION_HTTP1:
	conn = qd_http1_configure_connector(qd, &config, entity);
	break;
	case VERSION_HTTP2:
	conn = qd_http2_configure_connector(qd, &config, entity);
	break;
	}

	if (!conn)
	qd_http_free_bridge_config(&config);

	return conn;
	}


	void qd_dispatch_delete_http_connector(qd_dispatch_t qd, void impl)
	{
	qd_http_connector_t conn = (qd_http_connector_t) impl;

	if (conn) {
	switch (conn->config.version) {
	case VERSION_HTTP1:
	qd_http1_delete_connector(qd, conn);
	break;
	case VERSION_HTTP2:
	qd_http2_delete_connector(qd, conn);
	break;
	}
	}
	}

	qd_error_t qd_entity_refresh_httpConnector(qd_entity_t* entity, void *impl)
	{
	return QD_ERROR_NONE;
	}

	//
	// qd_http_listener_t constructor
	//

	qd_http_listener_t qd_http_listener(qd_server_t server, qd_server_event_handler_t handler)
	{
	qd_http_listener_t *li = new_qd_http_listener_t();
	if (!li)
	return 0;
	ZERO(li);

	li->pn_listener = pn_listener();
	if (!li->pn_listener) {
	free_qd_http_listener_t(li);
	return 0;
	}

	sys_atomic_init(&li->ref_count, 1);
	li->server = server;
	li->context.context = li;
	li->context.handler = handler;
	pn_listener_set_context(li->pn_listener, &li->context);

	return li;
	}

	void qd_http_listener_decref(qd_http_listener_t* li)
	{
	if (li && sys_atomic_dec(&li->ref_count) == 1) {
	qd_http_free_bridge_config(&li->config);
	free_qd_http_listener_t(li);
	}
	}

	//
	// qd_http_connector_t constructor
	//

	qd_http_connector_t qd_http_connector(qd_server_t server)
	{
	qd_http_connector_t *c = new_qd_http_connector_t();
	if (!c) return 0;
	ZERO(c);
	sys_atomic_init(&c->ref_count, 1);
	c->server = server;
	return c;
	}

	void qd_http_connector_decref(qd_http_connector_t* c)
	{
	if (c && sys_atomic_dec(&c->ref_count) == 1) {
	qd_http_free_bridge_config(&c->config);
	free_qd_http_connector_t(c);
	}
	}


	typedef struct qdr_http_method_status_t qdr_http_method_status_t;

	struct qdr_http_method_status_t {
	DEQ_LINKS(qdr_http_method_status_t);

	char *key;
	uint64_t requests;
	};
	DEQ_DECLARE(qdr_http_method_status_t, qdr_http_method_status_list_t);

	typedef struct qdr_http_request_info_t qdr_http_request_info_t;

	struct qdr_http_request_info_t {
	DEQ_LINKS(qdr_http_request_info_t);

	char *key;
	char *address;
	char *host;
	char *site;
	bool ingress;
	uint64_t requests;
	uint64_t bytes_in;
	uint64_t bytes_out;
	uint64_t max_latency;
	qdr_http_method_status_list_t detail;
	};
	DEQ_DECLARE(qdr_http_request_info_t, qdr_http_request_info_list_t);

	#define QDR_HTTP_REQUEST_INFO_NAME 0
	#define QDR_HTTP_REQUEST_INFO_IDENTITY 1
	#define QDR_HTTP_REQUEST_INFO_ADDRESS 2
	#define QDR_HTTP_REQUEST_INFO_HOST 3
	#define QDR_HTTP_REQUEST_INFO_SITE 4
	#define QDR_HTTP_REQUEST_INFO_DIRECTION 5
	#define QDR_HTTP_REQUEST_INFO_REQUESTS 6
	#define QDR_HTTP_REQUEST_INFO_BYTES_IN 7
	#define QDR_HTTP_REQUEST_INFO_BYTES_OUT 8
	#define QDR_HTTP_REQUEST_INFO_MAX_LATENCY 9
	#define QDR_HTTP_REQUEST_INFO_DETAIL 10


	const char * const QDR_HTTP_REQUEST_INFO_DIRECTION_IN = "in";
	const char * const QDR_HTTP_REQUEST_INFO_DIRECTION_OUT = "out";

	const char *qdr_http_request_info_columns[] =
	{"name",
	"identity",
	"address",
	"host",
	"site",
	"direction",
	"requests",
	"bytesIn",
	"bytesOut",
	"maxLatency",
	"details",
	0};

	const char *HTTP_REQUEST_INFO_TYPE = "org.apache.qpid.dispatch.httpRequestInfo";

	typedef struct {
	qdr_http_request_info_list_t records;
	} http_request_info_records_t;

	static http_request_info_records_t* request_info = 0;

	static http_request_info_records_t *_get_request_info()
	{
	if (!request_info) {
	request_info = NEW(http_request_info_records_t);
	DEQ_INIT(request_info->records);
	}
	return request_info;
	}

	static void insert_column(qdr_core_t core, qdr_http_request_info_t record, int col, qd_composed_field_t *body)
	{
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Insert column %i for %p", col, (void*) record);

	if (!record)
	return;

	switch(col) {
	case QDR_HTTP_REQUEST_INFO_NAME:
	qd_compose_insert_string(body, record->key);
	break;

	case QDR_HTTP_REQUEST_INFO_IDENTITY: {
	qd_compose_insert_string(body, record->key);
	break;
	}

	case QDR_HTTP_REQUEST_INFO_ADDRESS:
	qd_compose_insert_string(body, record->address);
	break;

	case QDR_HTTP_REQUEST_INFO_HOST:
	qd_compose_insert_string(body, record->host);
	break;

	case QDR_HTTP_REQUEST_INFO_SITE:
	qd_compose_insert_string(body, record->site);
	break;

	case QDR_HTTP_REQUEST_INFO_DIRECTION:
	if (record->ingress)
	qd_compose_insert_string(body, QDR_HTTP_REQUEST_INFO_DIRECTION_IN);
	else
	qd_compose_insert_string(body, QDR_HTTP_REQUEST_INFO_DIRECTION_OUT);
	break;

	case QDR_HTTP_REQUEST_INFO_REQUESTS:
	qd_compose_insert_uint(body, record->requests);
	break;

	case QDR_HTTP_REQUEST_INFO_BYTES_IN:
	qd_compose_insert_uint(body, record->bytes_in);
	break;

	case QDR_HTTP_REQUEST_INFO_BYTES_OUT:
	qd_compose_insert_uint(body, record->bytes_out);
	break;

	case QDR_HTTP_REQUEST_INFO_MAX_LATENCY:
	qd_compose_insert_uint(body, record->max_latency);
	break;

	case QDR_HTTP_REQUEST_INFO_DETAIL:
	qd_compose_start_map(body);
	for (qdr_http_method_status_t *item = DEQ_HEAD(record->detail); item; item = DEQ_NEXT(item)) {
	qd_compose_insert_string(body, item->key);
	qd_compose_insert_int(body, item->requests);
	}
	qd_compose_end_map(body);
	break;

	}
	}


	static void write_list(qdr_core_t core, qdr_query_t query, qdr_http_request_info_t *record)
	{
	qd_composed_field_t *body = query->body;

	qd_compose_start_list(body);

	if (record) {
	int i = 0;
	while (query->columns[i] >= 0) {
	insert_column(core, record, query->columns[i], body);
	i++;
	}
	}
	qd_compose_end_list(body);
	}

	static void write_map(qdr_core_t *core,
	qdr_http_request_info_t *record,
	qd_composed_field_t *body,
	const char *qdr_connection_columns[])
	{
	qd_compose_start_map(body);

	for(int i = 0; i < QDR_HTTP_REQUEST_INFO_COLUMN_COUNT; i++) {
	qd_compose_insert_string(body, qdr_connection_columns[i]);
	insert_column(core, record, i, body);
	}

	qd_compose_end_map(body);
	}

	static void advance(qdr_query_t query, qdr_http_request_info_t record)
	{
	if (record) {
	query->next_offset++;
	record = DEQ_NEXT(record);
	query->more = !!record;
	}
	else {
	query->more = false;
	}
	}

	static qdr_http_request_info_t find_by_identity(qdr_core_t core, qd_iterator_t *identity)
	{
	if (!identity)
	return 0;

	qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records);
	while (record) {
	// Convert the passed in identity to a char*
	if (qd_iterator_equal(identity, (const unsigned char*) record->key))
	break;
	record = DEQ_NEXT(record);
	}

	return record;

	}

	void qdra_http_request_info_get_first_CT(qdr_core_t core, qdr_query_t query, int offset)
	{
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "query for first http request info (%i)", offset);
	query->status = QD_AMQP_OK;

	if (offset >= DEQ_SIZE(_get_request_info()->records)) {
	query->more = false;
	qdr_agent_enqueue_response_CT(core, query);
	return;
	}

	qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records);
	for (int i = 0; i < offset && record; i++)
	record = DEQ_NEXT(record);
	assert(record);

	if (record) {
	write_list(core, query, record);
	query->next_offset = offset;
	advance(query, record);
	} else {
	query->more = false;
	}

	qdr_agent_enqueue_response_CT(core, query);
	}

	void qdra_http_request_info_get_next_CT(qdr_core_t core, qdr_query_t query)
	{
	qdr_http_request_info_t *record = 0;

	if (query->next_offset < DEQ_SIZE(_get_request_info()->records)) {
	record = DEQ_HEAD(_get_request_info()->records);
	for (int i = 0; i < query->next_offset && record; i++)
	record = DEQ_NEXT(record);
	}

	if (record) {
	write_list(core, query, record);
	advance(query, record);
	} else {
	query->more = false;
	}
	qdr_agent_enqueue_response_CT(core, query);
	}

	void qdra_http_request_info_get_CT(qdr_core_t *core,
	qd_iterator_t *name,
	qd_iterator_t *identity,
	qdr_query_t *query,
	const char *qdr_http_request_info_columns[])
	{
	qdr_http_request_info_t *record = 0;

	if (!identity) {
	query->status = QD_AMQP_BAD_REQUEST;
	query->status.description = "Name not supported. Identity required";
	qd_log(core->agent_log, QD_LOG_ERROR, "Error performing READ of %s: %s", HTTP_REQUEST_INFO_TYPE, query->status.description);
	} else {
	record = find_by_identity(core, identity);

	if (record == 0) {
	query->status = QD_AMQP_NOT_FOUND;
	} else {
	write_map(core, record, query->body, qdr_http_request_info_columns);
	query->status = QD_AMQP_OK;
	}
	}
	qdr_agent_enqueue_response_CT(core, query);
	}

	static const char* UNKNOWN = "unknown";

	static qdr_http_method_status_t* _new_qdr_http_method_status_t(const char *const method, int status)
	{
	qdr_http_method_status_t* record = NEW(qdr_http_method_status_t);
	ZERO(record);

	if (status >= 600 \|\| status < 100) {
	status = 500;
	}
	if (method) {
	size_t key_len = strlen(method) + 5;
	record->key = malloc(key_len);
	snprintf(record->key, key_len, "%s:%03i", method, status);
	} else {
	record->key = qd_strdup(UNKNOWN);
	}

	return record;
	}

	static void _free_qdr_http_method_status(qdr_http_method_status_t* record)
	{
	free(record->key);
	free(record);
	}

	static void _free_qdr_http_request_info(qdr_http_request_info_t* record)
	{
	if (record->key) {
	free(record->key);
	}
	if (record->address) {
	free(record->address);
	}
	if (record->host) {
	free(record->host);
	}
	if (record->site) {
	free(record->site);
	}
	for (qdr_http_method_status_t *item = DEQ_HEAD(record->detail); item; item = DEQ_HEAD(record->detail)) {
	_free_qdr_http_method_status(item);
	}
	free(record);
	}

	static void _update_http_method_status_detail(qdr_http_method_status_list_t detail, qdr_http_method_status_t addition)
	{
	bool updated = false;
	for (qdr_http_method_status_t item = DEQ_HEAD(detail); item && !updated; item = DEQ_NEXT(item)) {
	if (strcmp(item->key, addition->key) == 0) {
	item->requests += addition->requests;
	_free_qdr_http_method_status(addition);
	updated = true;
	}
	}
	if (!updated) {
	DEQ_INSERT_TAIL(*detail, addition);
	}
	}

	static bool _update_qdr_http_request_info(qdr_http_request_info_t* record, qdr_http_request_info_t* additions)
	{
	if (strcmp(record->key, additions->key) == 0) {
	record->requests += additions->requests;
	record->bytes_in += additions->bytes_in;
	record->bytes_out += additions->bytes_out;
	if (additions->max_latency > record->max_latency) {
	record->max_latency = additions->max_latency;
	}
	for (qdr_http_method_status_t *item = DEQ_HEAD(additions->detail); item; item = DEQ_HEAD(additions->detail)) {
	DEQ_REMOVE_HEAD(additions->detail);
	_update_http_method_status_detail(&record->detail, item);
	}
	return true;
	} else {
	return false;
	}
	}

	static void _add_http_request_info_CT(qdr_core_t core, qdr_action_t action, bool discard)
	{
	qdr_http_request_info_t update = (qdr_http_request_info_t) action->args.general.context_1;
	bool updated = false;
	for (qdr_http_request_info_t *record = DEQ_HEAD(_get_request_info()->records); record && !updated; record = DEQ_NEXT(record)) {
	if (_update_qdr_http_request_info(record, update)) {
	updated = true;
	_free_qdr_http_request_info(update);
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Updated http request info %s", record->key);
	}
	}
	if (!updated) {
	DEQ_INSERT_TAIL(_get_request_info()->records, update);
	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Added http request info %s (%zu)", update->key, DEQ_SIZE(_get_request_info()->records));
	}
	}

	static void _add_http_request_info(qdr_core_t core, qdr_http_request_info_t record)
	{
	qdr_action_t *action = qdr_action(_add_http_request_info_CT, "add_http_request_info");
	action->args.general.context_1 = record;
	qdr_action_enqueue(core, action);
	}

	static qdr_http_request_info_t* _new_qdr_http_request_info_t()
	{
	qdr_http_request_info_t* record = NEW(qdr_http_request_info_t);
	ZERO(record);
	DEQ_INIT(record->detail);
	return record;
	}

	static char _record_key(const char host, const char address, const char site, bool ingress)
	{
	if (!host) return 0;
	size_t hostlen = strlen(host);
	size_t addresslen = address ? strlen(address) + 1 : 0;
	size_t sitelen = site ? strlen(site) + 1 : 0;
	char *key = malloc(hostlen + addresslen + sitelen + 3);
	size_t i = 0;
	key[i++] = ingress ? 'i' : 'o';
	key[i++] = '_';
	strcpy(key+i, host);
	i += hostlen;
	if (address) {
	key[i++] = '_';
	strcpy(key+i, address);
	i += (addresslen-1);
	}
	if (site) {
	key[i++] = '@';
	strcpy(key+i, site);
	}
	return key;
	}

	void qd_http_record_request(qdr_core_t core, const char method, uint32_t status_code, const char address, const char host,
	const char local_site, const char remote_site, bool ingress,
	uint64_t bytes_in, uint64_t bytes_out, uint64_t latency)
	{
	qdr_http_request_info_t* record = _new_qdr_http_request_info_t();
	record->ingress = ingress;
	record->address = address ? qd_strdup(address) : 0;
	record->host = host ? qd_strdup(host) : 0;
	record->site = remote_site ? qd_strdup(remote_site) : 0;
	record->key = _record_key(record->host, record->address, remote_site, record->ingress);
	record->requests = 1;
	record->bytes_in = bytes_in;
	record->bytes_out = bytes_out;
	record->max_latency = latency;

	qdr_http_method_status_t *detail = _new_qdr_http_method_status_t(method, (int) status_code);
	detail->requests = 1;
	DEQ_INSERT_TAIL(record->detail, detail);

	qd_log(qd_log_source(QD_HTTP_LOG_SOURCE), QD_LOG_DEBUG, "Adding http request info %s", record->key);
	_add_http_request_info(core, record);
	}

	char qd_get_host_from_host_port(const char host_port)
	{
	char *end = strchr(host_port, ':');
	if (end == NULL) {
	return 0;
	} else {
	size_t len = end - host_port;
	char *host = malloc(len + 1);
	strncpy(host, host_port, len);
	host[len] = '\0';
	return host;
	}
	}