You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2020/08/04 18:57:51 UTC
[qpid-dispatch] 05/32: Dataplane: Added adaptor plugin capability.
Started first reference adaptor (TCP).
This is an automated email from the ASF dual-hosted git repository.
tross pushed a commit to branch dev-protocol-adaptors
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
commit e069627915d504a27c40a07da830ee39616c8b26
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 10:24:04 2020 -0400
Dataplane: Added adaptor plugin capability. Started first reference adaptor (TCP).
---
include/qpid/dispatch/protocol_adaptor.h | 42 ++++++++-
src/CMakeLists.txt | 1 +
src/adaptors/tcp_adaptor.c | 145 +++++++++++++++++++++++++++++++
src/router_core/router_core.c | 3 +
src/router_core/router_core_private.h | 1 +
src/router_core/router_core_thread.c | 51 +++++++++++
6 files changed, 241 insertions(+), 2 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index 9eb94c9..570b867 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -31,6 +31,44 @@ typedef struct qdr_connection_info_t qdr_connection_info_t;
/**
******************************************************************************
+ * Protocol adaptor declaration macro
+ ******************************************************************************
+ */
+/**
+ * Callback to initialize a protocol adaptor at core thread startup
+ *
+ * @param core Pointer to the core object
+ * @param adaptor_context [out] Returned adaptor context
+ */
+typedef void (*qdr_adaptor_init_t) (qdr_core_t *core, void **adaptor_context);
+
+
+/**
+ * Callback to finalize a protocol adaptor at core thread shutdown
+ *
+ * @param adaptor_context The context returned by the adaptor during the on_init call
+ */
+typedef void (*qdr_adaptor_final_t) (void *adaptor_context);
+
+
+/**
+ * Declaration of a protocol adaptor
+ *
+ * A protocol adaptor must declare itself by invoking the QDR_CORE_ADAPTOR_DECLARE macro in its body.
+ *
+ * @param name A null-terminated literal string naming the module
+ * @param on_init Pointer to a function for adaptor initialization, called at core thread startup
+ * @param on_final Pointer to a function for adaptor finalization, called at core thread shutdown
+ */
+#define QDR_CORE_ADAPTOR_DECLARE(name,on_init,on_final) \
+ static void adaptorstart() __attribute__((constructor)); \
+ void adaptorstart() { qdr_register_adaptor(name, on_init, on_final); }
+void qdr_register_adaptor(const char *name,
+ qdr_adaptor_init_t on_init,
+ qdr_adaptor_final_t on_final);
+
+/**
+ ******************************************************************************
* Callback function definitions
******************************************************************************
*/
@@ -91,7 +129,7 @@ typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
/**
* qdr_link_push_t callback
*/
-typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
+typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
/**
* qdr_link_deliver_t callback
@@ -703,7 +741,7 @@ void qdr_delivery_set_remote_extension_state(qdr_delivery_t *dlv, uint64_t remot
pn_data_t *qdr_delivery_take_local_extension_state(qdr_delivery_t *dlv, uint64_t *dispo);
-void qdr_connection_handlers(qdr_core_t *core,
+void qdr_connection_handlers(qdr_core_t *core,
void *context,
qdr_connection_activate_t activate,
qdr_link_first_attach_t first_attach,
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 88b09ea..8c0c7f7 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -38,6 +38,7 @@ add_custom_command (
# Build the qpid-dispatch library.
set(qpid_dispatch_SOURCES
+ adaptors/tcp_adaptor.c
alloc_pool.c
amqp.c
bitmask.c
diff --git a/src/adaptors/tcp_adaptor.c b/src/adaptors/tcp_adaptor.c
new file mode 100644
index 0000000..4ec9884
--- /dev/null
+++ b/src/adaptors/tcp_adaptor.c
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "qpid/dispatch/ctools.h"
+#include "qpid/dispatch/protocol_adaptor.h"
+#include "delivery.h"
+#include <stdio.h>
+#include <inttypes.h>
+
+typedef struct qdr_tcp_adaptor_t {
+ qdr_core_t *core;
+ qdr_protocol_adaptor_t *adaptor;
+} qdr_tcp_adaptor_t;
+
+
+static void qdr_tcp_first_attach(void *context, qdr_connection_t *conn, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target,
+ qd_session_class_t session_class)
+{
+}
+
+
+static void qdr_tcp_second_attach(void *context, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target)
+{
+}
+
+
+static void qdr_tcp_detach(void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close)
+{
+}
+
+
+static void qdr_tcp_flow(void *context, qdr_link_t *link, int credit)
+{
+}
+
+
+static void qdr_tcp_offer(void *context, qdr_link_t *link, int delivery_count)
+{
+}
+
+
+static void qdr_tcp_drained(void *context, qdr_link_t *link)
+{
+}
+
+
+static void qdr_tcp_drain(void *context, qdr_link_t *link, bool mode)
+{
+}
+
+
+static int qdr_tcp_push(void *context, qdr_link_t *link, int limit)
+{
+ return 0;
+}
+
+
+static uint64_t qdr_tcp_deliver(void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled)
+{
+ return 0;
+}
+
+
+static int qdr_tcp_get_credit(void *context, qdr_link_t *link)
+{
+ return 0;
+}
+
+
+static void qdr_tcp_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled)
+{
+}
+
+
+static void qdr_tcp_conn_close(void *context, qdr_connection_t *conn, qdr_error_t *error)
+{
+}
+
+
+static void qdr_tcp_conn_trace(void *context, qdr_connection_t *conn, bool trace)
+{
+}
+
+
+/**
+ * This initialization function will be invoked when the router core is ready for the protocol
+ * adaptor to be created. This function must:
+ *
+ * 1) Register the protocol adaptor with the router-core.
+ * 2) Prepare the protocol adaptor to be configured.
+ */
+static void qdr_tcp_adaptor_init(qdr_core_t *core, void **adaptor_context)
+{
+ qdr_tcp_adaptor_t *adaptor = NEW(qdr_tcp_adaptor_t);
+ adaptor->core = core;
+ adaptor->adaptor = qdr_protocol_adaptor(core,
+ "tcp",
+ adaptor,
+ 0, // activate
+ qdr_tcp_first_attach,
+ qdr_tcp_second_attach,
+ qdr_tcp_detach,
+ qdr_tcp_flow,
+ qdr_tcp_offer,
+ qdr_tcp_drained,
+ qdr_tcp_drain,
+ qdr_tcp_push,
+ qdr_tcp_deliver,
+ qdr_tcp_get_credit,
+ qdr_tcp_delivery_update,
+ qdr_tcp_conn_close,
+ qdr_tcp_conn_trace);
+ *adaptor_context = adaptor;
+}
+
+
+static void qdr_tcp_adaptor_final(void *adaptor_context)
+{
+ qdr_tcp_adaptor_t *adaptor = (qdr_tcp_adaptor_t*) adaptor_context;
+ qdr_protocol_adaptor_free(adaptor->core, adaptor->adaptor);
+ free(adaptor);
+}
+
+/**
+ * Declare the adaptor so that it will self-register on process startup.
+ */
+QDR_CORE_ADAPTOR_DECLARE("tcp-adaptor", qdr_tcp_adaptor_init, qdr_tcp_adaptor_final)
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index 08d2b7d..31ae0ad 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -213,6 +213,7 @@ void qdr_core_free(qdr_core_t *core)
// at this point all the conn identifiers have been freed
qd_hash_free(core->conn_id_hash);
+ qdr_adaptors_finalize(core);
qdr_modules_finalize(core);
qdr_agent_free(core->mgmt_agent);
@@ -939,6 +940,8 @@ qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
{
qdr_protocol_adaptor_t *adaptor = NEW(qdr_protocol_adaptor_t);
+ qd_log(core->log, QD_LOG_INFO, "Protocol adaptor registered: %s", name);
+
DEQ_ITEM_INIT(adaptor);
adaptor->name = name;
adaptor->user_context = context;
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index f91ac9f..7d8e09e 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -1007,6 +1007,7 @@ qdr_query_t *qdr_query(qdr_core_t *core,
qd_composed_field_t *body,
uint64_t conn_id);
void qdr_modules_finalize(qdr_core_t *core);
+void qdr_adaptors_finalize(qdr_core_t *core);
/**
* Create a new timer which will only be used inside the code thread.
diff --git a/src/router_core/router_core_thread.c b/src/router_core/router_core_thread.c
index 1e137ae..cc73d01 100644
--- a/src/router_core/router_core_thread.c
+++ b/src/router_core/router_core_thread.c
@@ -19,6 +19,7 @@
#include "router_core_private.h"
#include "module.h"
+#include <qpid/dispatch/protocol_adaptor.h>
#include "dispatch_private.h"
/**
@@ -57,6 +58,28 @@ void qdr_register_core_module(const char *name, qdrc_module_enable_t enable, qdr
}
+typedef struct qdrc_adaptor_t {
+ DEQ_LINKS(struct qdrc_adaptor_t);
+ const char *name;
+ qdr_adaptor_init_t on_init;
+ qdr_adaptor_final_t on_final;
+ void *context;
+} qdrc_adaptor_t;
+
+DEQ_DECLARE(qdrc_adaptor_t, qdrc_adaptor_list_t);
+static qdrc_adaptor_list_t registered_adaptors = {0,0};
+
+void qdr_register_adaptor(const char *name, qdr_adaptor_init_t on_init, qdr_adaptor_final_t on_final)
+{
+ qdrc_adaptor_t *adaptor = NEW(qdrc_adaptor_t);
+ ZERO(adaptor);
+ adaptor->name = name;
+ adaptor->on_init = on_init;
+ adaptor->on_final = on_final;
+ DEQ_INSERT_TAIL(registered_adaptors, adaptor);
+}
+
+
static void qdr_activate_connections_CT(qdr_core_t *core)
{
qdr_connection_t *conn = DEQ_HEAD(core->connections_to_activate);
@@ -123,6 +146,33 @@ void qdr_modules_finalize(qdr_core_t *core)
}
+void qdr_adaptors_init(qdr_core_t *core)
+{
+ //
+ // Initialize registered adaptors
+ //
+ qdrc_adaptor_t *adaptor = DEQ_HEAD(registered_adaptors);
+ while (adaptor) {
+ adaptor->on_init(core, &adaptor->context);
+ adaptor = DEQ_NEXT(adaptor);
+ }
+}
+
+
+void qdr_adaptors_finalize(qdr_core_t *core)
+{
+ //
+ // Finalize registered adaptors
+ //
+ qdrc_adaptor_t *adaptor = DEQ_TAIL(registered_adaptors);
+ while (adaptor) {
+ adaptor->on_final(adaptor->context);
+ adaptor = DEQ_PREV(adaptor);
+ }
+
+}
+
+
/*
* router_core_process_background_action_LH
*
@@ -159,6 +209,7 @@ void *router_core_thread(void *arg)
qdr_route_table_setup_CT(core);
qdr_modules_init(core);
+ qdr_adaptors_init(core);
qd_log(core->log, QD_LOG_INFO, "Router Core thread running. %s/%s", core->router_area, core->router_id);
while (core->running) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org