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:46 UTC
[qpid-dispatch] branch dev-protocol-adaptors created (now bcc2ddf)
This is an automated email from the ASF dual-hosted git repository.
tross pushed a change to branch dev-protocol-adaptors
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git.
at bcc2ddf DISPATCH-1742 Dataplane: Fixups from rebase to master.
This branch includes the following new commits:
new 1e7d562 Dataplane: Moved protocol-adapter functions from router_core.h into proto_adaptor.h
new 97fe8ff Dataplane: Re-factored direct-AMQP to use the protocol-adaptor interface.
new f4a4156 Dataplane: Exposed the protocol name with the connection entity.
new 4c7b94d Dataplane: Improved sasl-plain test by using symbolic attribute names, not positional indices.
new e069627 Dataplane: Added adaptor plugin capability. Started first reference adaptor (TCP).
new b75f05f Dataplane: Added documentation for the protocol adaptor callbacks.
new 7275350 Dataplane: Removed old handler call which is not used anymore.
new 1ab9b7f Dataplane: minor cleanup
new 6ea13c0 Dataplane: Added setter for dynamic in qdr_terminus_t
new 3b58b21 Dataplane: Renamed tcp_adaptor to reference_adaptor. Added more test content to the reference adaptor. It now sends messages to a fixed address. Fixed qdr_terminus_format to show the dynamically-assigned address for dynamis termini.
new f91328c Dataplane: connection-activate is now routed through the protocol adapter that handles the connection.
new 9545f35 Dataplane: Updated the reference adaptor to implement connection activation
new 084daa1 Dataplane: Fixed order problem in shutting down the router. Disabled the reference adaptor by default (uncomment the last line to re-enable). The reference adaptor causes test failures.
new 0226134 Dataplane: Added a 5th message compose variant to provide: - optional properties - optional application-properties - optional body in the form of a buffer list - indication of receive-complete
new ac5aa77 Dataplane: Changed the new compose function to have only one field for headers. This field can have both properties and application properties. It's more efficient put together like this.
new 9bf4ffd Dataplane: Added message method to set send-complete. Added reference code to receive messages (non streamed).
new c281064 Dataplane: Exposed access to connection-ids from server. Moved the generation of the "connection opened" log from router_node.c to the core module. This causes the log to be raised for all protocol adaptors.
new a307cf4 Dataplane: Added calls in message.h for streaming putput from adaptors. Renamed qdr_deliver_continue* to qdr_delivery_continue*
new 777f0b7 Dataplane: Set proper buffer refcount in messages during buffer-extend. This ensures that the streaming buffers are properly freed when no longer needed.
new 1463337 Dataplane: Updates to the message-extend (return buffer count for flow control). Added bidirectional streaming test to ref adaptor.
new 7cb6c07 Dataplane: Added API for streaming data out of messages. This commit adds the requirement for Proton raw-connection support.
new b3af7de Dataplane: Added no_route and initial_delivery on link-first-attach.
new 20f841a Dataplane: (from gsim) Implementation of qd_message_read_body.
new 8a53cbb Dataplane: Added implementation of qd_message_release_body.
new febda9e Dataplane - Added qd_buffer_list_append function to efficiently accumulate data in buffer lists.
new 4c11d64 Dataplane: WIP changes
new 3e2418e Dataplane: Message parsing bug fixed: now properly handles empty var-length fields.
new df0568a Dataplane: Fixed message parsing so it can handle partial and streaming content.
new 2898359 Dataplane: disabled reference adaptor
new 5bb50a1 Dataplane: WIP
new ee033ef Dataplane - Added the body_data data structure for reading streaming messages. WIP - The following functions (in message.c) need to be implemented: find_last_buffer qd_message_body_data_iterator qd_message_body_data_buffer_count qd_message_body_data_buffers qd_message_body_data_release
new bcc2ddf DISPATCH-1742 Dataplane: Fixups from rebase to master.
The 32 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 24/32: Dataplane: Added implementation of
qd_message_release_body.
Posted by tr...@apache.org.
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 8a53cbbe22ab8aff7017714ca131949ed33fe345
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jun 22 13:17:38 2020 -0400
Dataplane: Added implementation of qd_message_release_body.
---
include/qpid/dispatch/buffer.h | 9 +++++++++
src/message.c | 30 ++++++++++++++++++++++++++++++
2 files changed, 39 insertions(+)
diff --git a/include/qpid/dispatch/buffer.h b/include/qpid/dispatch/buffer.h
index 4de49bd..e93b5df 100644
--- a/include/qpid/dispatch/buffer.h
+++ b/include/qpid/dispatch/buffer.h
@@ -147,6 +147,15 @@ static inline uint32_t qd_buffer_set_fanout(qd_buffer_t *buf, uint32_t value)
}
/**
+ * Get the fanout value on the buffer.
+ * @return the count
+ */
+static inline uint32_t qd_buffer_get_fanout(const qd_buffer_t *buf)
+{
+ return buf->bfanout;
+}
+
+/**
* Increase the fanout by 1. How many receivers should this buffer be sent to.
* @return the _old_ count (pre increment)
*/
diff --git a/src/message.c b/src/message.c
index 55d134e..78f2888 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2292,6 +2292,36 @@ int qd_message_read_body(qd_message_t *in_msg, pn_raw_buffer_t* buffers, int len
}
+void qd_message_release_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int buffer_count)
+{
+ qd_message_pvt_t *pvt = (qd_message_pvt_t*) msg;
+ qd_message_content_t *content = MSG_CONTENT(msg);
+ qd_buffer_t *buf;
+
+ LOCK(content->lock);
+ //
+ // Decrement the buffer fanout for each of the referenced buffers.
+ //
+ if (pvt->is_fanout) {
+ for (int i = 0; i < buffer_count; i++) {
+ buf = (qd_buffer_t*) buffers[i].context;
+ qd_buffer_dec_fanout(buf);
+ }
+ }
+
+ //
+ // Free buffers at the head of the list that have zero refcounts.
+ //
+ buf = DEQ_HEAD(content->buffers);
+ while (buf && qd_buffer_get_fanout(buf) == 0) {
+ DEQ_REMOVE_HEAD(content->buffers);
+ qd_buffer_free(buf);
+ buf = DEQ_HEAD(content->buffers);
+ }
+ UNLOCK(content->lock);
+}
+
+
qd_parsed_field_t *qd_message_get_ingress (qd_message_t *msg)
{
return ((qd_message_pvt_t*)msg)->content->ma_pf_ingress;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 25/32: Dataplane - Added qd_buffer_list_append
function to efficiently accumulate data in buffer lists.
Posted by tr...@apache.org.
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 febda9e0598da7566fefee940289586ab6703993
Author: Ted Ross <tr...@apache.org>
AuthorDate: Fri Jun 26 16:30:28 2020 -0400
Dataplane - Added qd_buffer_list_append function to efficiently accumulate data in buffer lists.
---
include/qpid/dispatch/buffer.h | 16 ++++++++++++++--
src/buffer.c | 34 ++++++++++++++++++++++++++++++++++
tests/buffer_test.c | 35 +++++++++++++++++++++++++++++++++++
3 files changed, 83 insertions(+), 2 deletions(-)
diff --git a/include/qpid/dispatch/buffer.h b/include/qpid/dispatch/buffer.h
index e93b5df..715c813 100644
--- a/include/qpid/dispatch/buffer.h
+++ b/include/qpid/dispatch/buffer.h
@@ -38,8 +38,8 @@ extern size_t BUFFER_SIZE;
/** A raw byte buffer .*/
struct qd_buffer_t {
DEQ_LINKS(qd_buffer_t);
- unsigned int size; ///< Size of data content
- sys_atomic_t bfanout; // The number of receivers for this buffer
+ unsigned int size; ///< Size of data content
+ sys_atomic_t bfanout; ///< The number of receivers for this buffer
};
/**
@@ -185,6 +185,18 @@ static inline unsigned char *qd_buffer_at(const qd_buffer_t *buf, size_t len)
}
+/**
+ * qd_buffer_list_append
+ *
+ * Append new data to a buffer list using freespace efficiently and adding new buffers when necessary.
+ *
+ * @param buflist Pointer to a buffer list that will possibly be changed by adding new buffers
+ * @param data Pointer to raw binary data to be added to the buffer list
+ * @param len The number of bytes of data to append
+ */
+void qd_buffer_list_append(qd_buffer_list_t *buflist, uint8_t *data, size_t len);
+
+
///@}
#endif
diff --git a/src/buffer.c b/src/buffer.c
index 40af6c2..66a4af2 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -106,3 +106,37 @@ unsigned int qd_buffer_list_length(const qd_buffer_list_t *list)
}
return len;
}
+
+
+void qd_buffer_list_append(qd_buffer_list_t *buflist, uint8_t *data, size_t len)
+{
+ //
+ // If len is zero, there's no work to do.
+ //
+ if (len == 0)
+ return;
+
+ //
+ // If the buffer list is empty and there's some data, add one empty buffer before we begin.
+ //
+ if (DEQ_SIZE(*buflist) == 0) {
+ qd_buffer_t *buf = qd_buffer();
+ DEQ_INSERT_TAIL(*buflist, buf);
+ }
+
+ qd_buffer_t *tail = DEQ_TAIL(*buflist);
+
+ while (len > 0) {
+ size_t to_copy = MIN(len, qd_buffer_capacity(tail));
+ if (to_copy > 0) {
+ memcpy(qd_buffer_cursor(tail), data, to_copy);
+ qd_buffer_insert(tail, to_copy);
+ data += to_copy;
+ len -= to_copy;
+ }
+ if (len > 0) {
+ tail = qd_buffer();
+ DEQ_INSERT_TAIL(*buflist, tail);
+ }
+ }
+}
diff --git a/tests/buffer_test.c b/tests/buffer_test.c
index 6dd61f6..4b15450 100644
--- a/tests/buffer_test.c
+++ b/tests/buffer_test.c
@@ -87,12 +87,47 @@ static char *test_buffer_list_clone(void *context)
}
+static char *test_buffer_list_append(void *context)
+{
+ qd_buffer_list_t list;
+ qd_buffer_t *buf = qd_buffer();
+ size_t buffer_size = qd_buffer_capacity(buf);
+ qd_buffer_free(buf);
+
+ DEQ_INIT(list);
+
+ qd_buffer_list_append(&list, (uint8_t*) "", 0);
+ if (DEQ_SIZE(list) > 0) return "Buffer list should be empty";
+
+ qd_buffer_list_append(&list, (uint8_t*) "ABCDEFGHIJ", 10);
+ if (DEQ_SIZE(list) != (10 / buffer_size) + ((10 % buffer_size) ? 1 : 0)) return "Incorrect buffer count for size 10";
+
+ qd_buffer_list_append(&list, (uint8_t*) "KLMNOPQRSTUVWXYZ", 16);
+ if (DEQ_SIZE(list) != (26 / buffer_size) + ((26 % buffer_size) ? 1 : 0)) return "Incorrect buffer count for size 26";
+
+ size_t list_len = 0;
+ buf = DEQ_HEAD(list);
+ while (buf) {
+ list_len += qd_buffer_size(buf);
+ buf = DEQ_NEXT(buf);
+ }
+ if (list_len != 26) {
+ static char error[100];
+ sprintf(error, "Incorrect accumulated buffer size: %ld", list_len);
+ return error;
+ }
+
+ return 0;
+}
+
+
int buffer_tests()
{
int result = 0;
char *test_group = "buffer_tests";
TEST_CASE(test_buffer_list_clone, 0);
+ TEST_CASE(test_buffer_list_append, 0);
return result;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 02/32: Dataplane: Re-factored direct-AMQP to use
the protocol-adaptor interface.
Posted by tr...@apache.org.
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 97fe8ff22182ba56678dc6b766479c43e49ce2d5
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jun 1 16:40:53 2020 -0400
Dataplane: Re-factored direct-AMQP to use the protocol-adaptor interface.
---
include/qpid/dispatch/protocol_adaptor.h | 133 ++++++++++++++++++++++++-------
src/router_core/connections.c | 130 ++++++++++++------------------
src/router_core/router_core.c | 49 ++++++++++++
src/router_core/router_core_private.h | 57 +++++++------
src/router_core/transfer.c | 4 +-
src/router_node.c | 46 +++++++----
6 files changed, 267 insertions(+), 152 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index 33916ab..9eb94c9 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -50,22 +50,73 @@ typedef struct qdr_connection_info_t qdr_connection_info_t;
*/
typedef void (*qdr_connection_activate_t) (void *context, qdr_connection_t *conn);
-typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target,
- qd_session_class_t);
+/**
+ * qdr_link_first_attach_t callback
+ */
+typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target,
+ qd_session_class_t);
+
+/**
+ * qdr_link_second_attach_t callback
+ */
typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
qdr_terminus_t *source, qdr_terminus_t *target);
-typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
-typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
-typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
-typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
-typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
-typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
-typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
-typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
-typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
+
+/**
+ * qdr_link_detach_t callback
+ */
+typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
+
+/**
+ * qdr_link_flow_t callback
+ */
+typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
+
+/**
+ * qdr_link_offer_t callback
+ */
+typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
+
+/**
+ * qdr_link_drained_t callback
+ */
+typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
+
+/**
+ * qdr_link_drain_t callback
+ */
+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);
+
+/**
+ * qdr_link_deliver_t callback
+ */
+typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
+
+/**
+ * qdr_link_get_credit_t callback
+ */
+typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
+
+/**
+ * qdr_delivery_update_t callback
+ */
+typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
+
+/**
+ * qdr_connection_close_t callback
+ */
+typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
+
+/**
+ * qdr_connection_trace_t callback
+ */
+typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
/**
@@ -74,6 +125,26 @@ typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn,
******************************************************************************
*/
+qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
+ const char *name,
+ void *context,
+ qdr_connection_activate_t activate,
+ qdr_link_first_attach_t first_attach,
+ qdr_link_second_attach_t second_attach,
+ qdr_link_detach_t detach,
+ qdr_link_flow_t flow,
+ qdr_link_offer_t offer,
+ qdr_link_drained_t drained,
+ qdr_link_drain_t drain,
+ qdr_link_push_t push,
+ qdr_link_deliver_t deliver,
+ qdr_link_get_credit_t get_credit,
+ qdr_delivery_update_t delivery_update,
+ qdr_connection_close_t conn_close,
+ qdr_connection_trace_t conn_trace);
+
+void qdr_protocol_adaptor_free(qdr_core_t *core, qdr_protocol_adaptor_t *adaptor);
+
/**
******************************************************************************
@@ -104,6 +175,7 @@ typedef void (*qdr_connection_bind_context_t) (qdr_connection_t *context, void*
* Once a new connection has been both remotely and locally opened, the core must be notified.
*
* @param core Pointer to the core object
+ * @param protocol_adaptor Pointer to the protocol adaptor handling the connection
* @param incoming True iff this connection is associated with a listener, False if a connector
* @param role The configured role of this connection
* @param cost If the role is inter_router, this is the configured cost for the connection.
@@ -118,22 +190,23 @@ typedef void (*qdr_connection_bind_context_t) (qdr_connection_t *context, void*
* @param vhost If non-null, this is the vhost of the connection to be used for multi-tenancy.
* @return Pointer to a connection object that can be used to refer to this connection over its lifetime.
*/
-qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
- bool incoming,
- qdr_connection_role_t role,
- int cost,
- uint64_t management_id,
- const char *label,
- const char *remote_container_id,
- bool strip_annotations_in,
- bool strip_annotations_out,
- bool policy_allow_dynamic_link_routes,
- bool policy_allow_admin_status_update,
- int link_capacity,
- const char *vhost,
- qdr_connection_info_t *connection_info,
- qdr_connection_bind_context_t context_binder,
- void* bind_token);
+qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
+ qdr_protocol_adaptor_t *protocol_adaptor,
+ bool incoming,
+ qdr_connection_role_t role,
+ int cost,
+ uint64_t management_id,
+ const char *label,
+ const char *remote_container_id,
+ bool strip_annotations_in,
+ bool strip_annotations_out,
+ bool policy_allow_dynamic_link_routes,
+ bool policy_allow_admin_status_update,
+ int link_capacity,
+ const char *vhost,
+ qdr_connection_info_t *connection_info,
+ qdr_connection_bind_context_t context_binder,
+ void *bind_token);
/**
* qdr_connection_closed
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 2d05061..67ee1cb 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -67,27 +67,29 @@ qdr_terminus_t *qdr_terminus_router_data(void)
// Interface Functions
//==================================================================================
-qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
- bool incoming,
- qdr_connection_role_t role,
- int cost,
- uint64_t management_id,
- const char *label,
- const char *remote_container_id,
- bool strip_annotations_in,
- bool strip_annotations_out,
- bool policy_allow_dynamic_link_routes,
- bool policy_allow_admin_status_update,
- int link_capacity,
- const char *vhost,
- qdr_connection_info_t *connection_info,
+qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
+ qdr_protocol_adaptor_t *protocol_adaptor,
+ bool incoming,
+ qdr_connection_role_t role,
+ int cost,
+ uint64_t management_id,
+ const char *label,
+ const char *remote_container_id,
+ bool strip_annotations_in,
+ bool strip_annotations_out,
+ bool policy_allow_dynamic_link_routes,
+ bool policy_allow_admin_status_update,
+ int link_capacity,
+ const char *vhost,
+ qdr_connection_info_t *connection_info,
qdr_connection_bind_context_t context_binder,
- void *bind_token)
+ void *bind_token)
{
qdr_action_t *action = qdr_action(qdr_connection_opened_CT, "connection_opened");
qdr_connection_t *conn = new_qdr_connection_t();
ZERO(conn);
+ conn->protocol_adaptor = protocol_adaptor;
conn->identity = management_id;
conn->connection_info = connection_info;
conn->core = core;
@@ -236,26 +238,28 @@ void qdr_record_link_credit(qdr_core_t *core, qdr_link_t *link)
//
// Get Proton's view of this link's available credit.
//
- int pn_credit = core->get_credit_handler(core->user_context, link);
+ if (link && link->conn && link->conn->protocol_adaptor) {
+ int pn_credit = link->conn->protocol_adaptor->get_credit_handler(link->conn->protocol_adaptor->user_context, link);
- if (link->credit_reported > 0 && pn_credit == 0) {
- //
- // The link has transitioned from positive credit to zero credit.
- //
- link->zero_credit_time = core->uptime_ticks;
- } else if (link->credit_reported == 0 && pn_credit > 0) {
- //
- // The link has transitioned from zero credit to positive credit.
- // Clear the recorded time.
- //
- link->zero_credit_time = 0;
- if (link->reported_as_blocked) {
- link->reported_as_blocked = false;
- core->links_blocked--;
+ if (link->credit_reported > 0 && pn_credit == 0) {
+ //
+ // The link has transitioned from positive credit to zero credit.
+ //
+ link->zero_credit_time = core->uptime_ticks;
+ } else if (link->credit_reported == 0 && pn_credit > 0) {
+ //
+ // The link has transitioned from zero credit to positive credit.
+ // Clear the recorded time.
+ //
+ link->zero_credit_time = 0;
+ if (link->reported_as_blocked) {
+ link->reported_as_blocked = false;
+ core->links_blocked--;
+ }
}
- }
- link->credit_reported = pn_credit;
+ link->credit_reported = pn_credit;
+ }
}
@@ -273,7 +277,7 @@ int qdr_connection_process(qdr_connection_t *conn)
int event_count = 0;
if (conn->closed) {
- core->conn_close_handler(core->user_context, conn, conn->error);
+ conn->protocol_adaptor->conn_close_handler(conn->protocol_adaptor->user_context, conn, conn->error);
return 0;
}
@@ -302,19 +306,19 @@ int qdr_connection_process(qdr_connection_t *conn)
switch (work->work_type) {
case QDR_CONNECTION_WORK_FIRST_ATTACH :
- core->first_attach_handler(core->user_context, conn, work->link, work->source, work->target, work->ssn_class);
+ conn->protocol_adaptor->first_attach_handler(conn->protocol_adaptor->user_context, conn, work->link, work->source, work->target, work->ssn_class);
break;
case QDR_CONNECTION_WORK_SECOND_ATTACH :
- core->second_attach_handler(core->user_context, work->link, work->source, work->target);
+ conn->protocol_adaptor->second_attach_handler(conn->protocol_adaptor->user_context, work->link, work->source, work->target);
break;
case QDR_CONNECTION_WORK_TRACING_ON :
- core->conn_trace_handler(core->user_context, conn, true);
+ conn->protocol_adaptor->conn_trace_handler(conn->protocol_adaptor->user_context, conn, true);
break;
case QDR_CONNECTION_WORK_TRACING_OFF :
- core->conn_trace_handler(core->user_context, conn, false);
+ conn->protocol_adaptor->conn_trace_handler(conn->protocol_adaptor->user_context, conn, false);
break;
}
@@ -353,7 +357,7 @@ int qdr_connection_process(qdr_connection_t *conn)
qdr_delivery_ref_t *dref = DEQ_HEAD(updated_deliveries);
while (dref) {
- core->delivery_update_handler(core->user_context, dref->dlv, dref->dlv->disposition, dref->dlv->settled);
+ conn->protocol_adaptor->delivery_update_handler(conn->protocol_adaptor->user_context, dref->dlv, dref->dlv->disposition, dref->dlv->settled);
qdr_delivery_decref(core, dref->dlv, "qdr_connection_process - remove from updated list");
qdr_del_delivery_ref(&updated_deliveries, dref);
dref = DEQ_HEAD(updated_deliveries);
@@ -364,7 +368,7 @@ int qdr_connection_process(qdr_connection_t *conn)
switch (link_work->work_type) {
case QDR_LINK_WORK_DELIVERY :
{
- int count = core->push_handler(core->user_context, link, link_work->value);
+ int count = conn->protocol_adaptor->push_handler(conn->protocol_adaptor->user_context, link, link_work->value);
assert(count <= link_work->value);
link_work->value -= count;
break;
@@ -372,20 +376,20 @@ int qdr_connection_process(qdr_connection_t *conn)
case QDR_LINK_WORK_FLOW :
if (link_work->value > 0)
- core->flow_handler(core->user_context, link, link_work->value);
+ conn->protocol_adaptor->flow_handler(conn->protocol_adaptor->user_context, link, link_work->value);
if (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_SET)
- core->drain_handler(core->user_context, link, true);
+ conn->protocol_adaptor->drain_handler(conn->protocol_adaptor->user_context, link, true);
else if (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_CLEAR)
- core->drain_handler(core->user_context, link, false);
+ conn->protocol_adaptor->drain_handler(conn->protocol_adaptor->user_context, link, false);
else if (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_DRAINED)
- core->drained_handler(core->user_context, link);
+ conn->protocol_adaptor->drained_handler(conn->protocol_adaptor->user_context, link);
break;
case QDR_LINK_WORK_FIRST_DETACH :
case QDR_LINK_WORK_SECOND_DETACH :
- core->detach_handler(core->user_context, link, link_work->error,
- link_work->work_type == QDR_LINK_WORK_FIRST_DETACH,
- link_work->close_link);
+ conn->protocol_adaptor->detach_handler(conn->protocol_adaptor->user_context, link, link_work->error,
+ link_work->work_type == QDR_LINK_WORK_FIRST_DETACH,
+ link_work->close_link);
detach_sent = true;
break;
}
@@ -632,40 +636,6 @@ static void qdr_link_processing_complete(qdr_core_t *core, qdr_link_t *link)
-void qdr_connection_handlers(qdr_core_t *core,
- void *context,
- qdr_connection_activate_t activate,
- qdr_link_first_attach_t first_attach,
- qdr_link_second_attach_t second_attach,
- qdr_link_detach_t detach,
- qdr_link_flow_t flow,
- qdr_link_offer_t offer,
- qdr_link_drained_t drained,
- qdr_link_drain_t drain,
- qdr_link_push_t push,
- qdr_link_deliver_t deliver,
- qdr_link_get_credit_t get_credit,
- qdr_delivery_update_t delivery_update,
- qdr_connection_close_t conn_close,
- qdr_connection_trace_t conn_trace)
-{
- core->user_context = context;
- core->first_attach_handler = first_attach;
- core->second_attach_handler = second_attach;
- core->detach_handler = detach;
- core->flow_handler = flow;
- core->offer_handler = offer;
- core->drained_handler = drained;
- core->drain_handler = drain;
- core->push_handler = push;
- core->deliver_handler = deliver;
- core->get_credit_handler = get_credit;
- core->delivery_update_handler = delivery_update;
- core->conn_close_handler = conn_close;
- core->conn_trace_handler = conn_trace;
-}
-
-
//==================================================================================
// In-Thread Functions
//==================================================================================
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index d2d24d9..08d2b7d 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -908,6 +908,7 @@ static void qdr_global_stats_request_CT(qdr_core_t *core, qdr_action_t *action,
qdr_post_general_work_CT(core, work);
}
+
void qdr_request_global_stats(qdr_core_t *core, qdr_global_stats_t *stats, qdr_global_stats_handler_t callback, void *context)
{
qdr_action_t *action = qdr_action(qdr_global_stats_request_CT, "global_stats_request");
@@ -917,3 +918,51 @@ void qdr_request_global_stats(qdr_core_t *core, qdr_global_stats_t *stats, qdr_g
qdr_action_enqueue(core, action);
}
+
+qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
+ const char *name,
+ void *context,
+ qdr_connection_activate_t activate,
+ qdr_link_first_attach_t first_attach,
+ qdr_link_second_attach_t second_attach,
+ qdr_link_detach_t detach,
+ qdr_link_flow_t flow,
+ qdr_link_offer_t offer,
+ qdr_link_drained_t drained,
+ qdr_link_drain_t drain,
+ qdr_link_push_t push,
+ qdr_link_deliver_t deliver,
+ qdr_link_get_credit_t get_credit,
+ qdr_delivery_update_t delivery_update,
+ qdr_connection_close_t conn_close,
+ qdr_connection_trace_t conn_trace)
+{
+ qdr_protocol_adaptor_t *adaptor = NEW(qdr_protocol_adaptor_t);
+
+ DEQ_ITEM_INIT(adaptor);
+ adaptor->name = name;
+ adaptor->user_context = context;
+ adaptor->first_attach_handler = first_attach;
+ adaptor->second_attach_handler = second_attach;
+ adaptor->detach_handler = detach;
+ adaptor->flow_handler = flow;
+ adaptor->offer_handler = offer;
+ adaptor->drained_handler = drained;
+ adaptor->drain_handler = drain;
+ adaptor->push_handler = push;
+ adaptor->deliver_handler = deliver;
+ adaptor->get_credit_handler = get_credit;
+ adaptor->delivery_update_handler = delivery_update;
+ adaptor->conn_close_handler = conn_close;
+ adaptor->conn_trace_handler = conn_trace;
+
+ DEQ_INSERT_TAIL(core->protocol_adaptors, adaptor);
+ return adaptor;
+}
+
+
+void qdr_protocol_adaptor_free(qdr_core_t *core, qdr_protocol_adaptor_t *adaptor)
+{
+ DEQ_REMOVE(core->protocol_adaptors, adaptor);
+ free(adaptor);
+}
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 49d9650..f91ac9f 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -656,6 +656,7 @@ typedef enum {
struct qdr_connection_t {
DEQ_LINKS(qdr_connection_t);
DEQ_LINKS_N(ACTIVATE, qdr_connection_t);
+ qdr_protocol_adaptor_t *protocol_adaptor;
uint64_t identity;
qdr_core_t *core;
bool incoming;
@@ -784,6 +785,33 @@ typedef struct qdr_priority_sheaf_t {
int count;
} qdr_priority_sheaf_t;
+
+struct qdr_protocol_adaptor_t {
+ DEQ_LINKS(qdr_protocol_adaptor_t);
+ const char *name;
+
+ //
+ // Callbacks
+ //
+ void *user_context;
+ qdr_link_first_attach_t first_attach_handler;
+ qdr_link_second_attach_t second_attach_handler;
+ qdr_link_detach_t detach_handler;
+ qdr_link_flow_t flow_handler;
+ qdr_link_offer_t offer_handler;
+ qdr_link_drained_t drained_handler;
+ qdr_link_drain_t drain_handler;
+ qdr_link_push_t push_handler;
+ qdr_link_deliver_t deliver_handler;
+ qdr_link_get_credit_t get_credit_handler;
+ qdr_delivery_update_t delivery_update_handler;
+ qdr_connection_close_t conn_close_handler;
+ qdr_connection_trace_t conn_trace_handler;
+};
+
+DEQ_DECLARE(qdr_protocol_adaptor_t, qdr_protocol_adaptor_list_t);
+
+
struct qdr_core_t {
qd_dispatch_t *qd;
qd_log_source_t *log;
@@ -801,11 +829,12 @@ struct qdr_core_t {
qd_timer_t *work_timer;
uint32_t uptime_ticks;
- qdr_connection_list_t open_connections;
- qdr_connection_t *active_edge_connection;
- qdr_connection_list_t connections_to_activate;
- qdr_link_list_t open_links;
- qdr_connection_ref_list_t streaming_connections;
+ qdr_protocol_adaptor_list_t protocol_adaptors;
+ qdr_connection_list_t open_connections;
+ qdr_connection_t *active_edge_connection;
+ qdr_connection_list_t connections_to_activate;
+ qdr_link_list_t open_links;
+ qdr_connection_ref_list_t streaming_connections;
qdrc_attach_addr_lookup_t addr_lookup_handler;
void *addr_lookup_context;
@@ -821,24 +850,6 @@ struct qdr_core_t {
qdr_link_lost_t rt_link_lost;
//
- // Connection section
- //
- void *user_context;
- qdr_link_first_attach_t first_attach_handler;
- qdr_link_second_attach_t second_attach_handler;
- qdr_link_detach_t detach_handler;
- qdr_link_flow_t flow_handler;
- qdr_link_offer_t offer_handler;
- qdr_link_drained_t drained_handler;
- qdr_link_drain_t drain_handler;
- qdr_link_push_t push_handler;
- qdr_link_deliver_t deliver_handler;
- qdr_link_get_credit_t get_credit_handler;
- qdr_delivery_update_t delivery_update_handler;
- qdr_connection_close_t conn_close_handler;
- qdr_connection_trace_t conn_trace_handler;
-
- //
// Events section
//
qdrc_event_subscription_list_t conn_event_subscriptions;
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index 7d7aa9c..add176a 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -163,7 +163,7 @@ int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
do {
settled = dlv->settled;
sys_mutex_unlock(conn->work_lock);
- new_disp = core->deliver_handler(core->user_context, link, dlv, settled);
+ new_disp = conn->protocol_adaptor->deliver_handler(conn->protocol_adaptor->user_context, link, dlv, settled);
sys_mutex_lock(conn->work_lock);
} while (settled != dlv->settled); // oops missed the settlement
send_complete = qdr_delivery_send_complete(dlv);
@@ -238,7 +238,7 @@ int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
}
if (offer != -1)
- core->offer_handler(core->user_context, link, offer);
+ conn->protocol_adaptor->offer_handler(conn->protocol_adaptor->user_context, link, offer);
}
return num_deliveries_completed;
diff --git a/src/router_node.c b/src/router_node.c
index ce59329..a5278dc 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -37,6 +37,8 @@ const char *QD_ROUTER_NODE_TYPE = "router.node";
const char *QD_ROUTER_ADDRESS_TYPE = "router.address";
const char *QD_ROUTER_LINK_TYPE = "router.link";
+static qdr_protocol_adaptor_t *amqp_direct_adaptor = 0;
+
static char *router_role = "inter-router";
static char *container_role = "route-container";
static char *edge_role = "edge";
@@ -1199,7 +1201,13 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
is_ssl,
(rversion_found) ? &rversion : 0);
- qdr_connection_opened(router->router_core, inbound, role, cost, connection_id, name,
+ qdr_connection_opened(router->router_core,
+ amqp_direct_adaptor,
+ inbound,
+ role,
+ cost,
+ connection_id,
+ name,
pn_connection_remote_container(pn_conn),
conn->strip_annotations_in,
conn->strip_annotations_out,
@@ -1208,7 +1216,8 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
link_capacity,
vhost,
connection_info,
- bind_connection_context, conn);
+ bind_connection_context,
+ conn);
char props_str[1000];
size_t props_len = 1000;
@@ -1968,21 +1977,23 @@ void qd_router_setup_late(qd_dispatch_t *qd)
qd->router->tracemask = qd_tracemask();
qd->router->router_core = qdr_core(qd, qd->router->router_mode, qd->router->router_area, qd->router->router_id);
- qdr_connection_handlers(qd->router->router_core, (void*) qd->router,
- CORE_connection_activate,
- CORE_link_first_attach,
- CORE_link_second_attach,
- CORE_link_detach,
- CORE_link_flow,
- CORE_link_offer,
- CORE_link_drained,
- CORE_link_drain,
- CORE_link_push,
- CORE_link_deliver,
- CORE_link_get_credit,
- CORE_delivery_update,
- CORE_close_connection,
- CORE_conn_trace);
+ amqp_direct_adaptor = qdr_protocol_adaptor(qd->router->router_core,
+ "amqp-direct",
+ (void*) qd->router,
+ CORE_connection_activate,
+ CORE_link_first_attach,
+ CORE_link_second_attach,
+ CORE_link_detach,
+ CORE_link_flow,
+ CORE_link_offer,
+ CORE_link_drained,
+ CORE_link_drain,
+ CORE_link_push,
+ CORE_link_deliver,
+ CORE_link_get_credit,
+ CORE_delivery_update,
+ CORE_close_connection,
+ CORE_conn_trace);
qd_router_python_setup(qd->router);
qd_timer_schedule(qd->router->timer, 1000);
@@ -1994,6 +2005,7 @@ void qd_router_free(qd_router_t *router)
qd_container_set_default_node_type(router->qd, 0, 0, QD_DIST_BOTH);
+ qdr_protocol_adaptor_free(router->router_core, amqp_direct_adaptor);
qdr_core_free(router->router_core);
qd_tracemask_free(router->tracemask);
qd_timer_free(router->timer);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 14/32: Dataplane: Added a 5th message compose
variant to provide: - optional properties - optional application-properties
- optional body in the form of a buffer list - indication of
receive-complete
Posted by tr...@apache.org.
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 02261342d835f8cf3a09e9e15121e43da5222ab8
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 18:02:25 2020 -0400
Dataplane: Added a 5th message compose variant to provide: - optional properties - optional application-properties - optional body in the form of a buffer list - indication of receive-complete
---
include/qpid/dispatch/message.h | 6 +++
src/adaptors/reference_adaptor.c | 80 ++++++++++++++++++++++++++++++----------
src/message.c | 27 ++++++++++++++
3 files changed, 93 insertions(+), 20 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 63f0b47..55c3dba 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -278,6 +278,12 @@ void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *content);
void qd_message_compose_3(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2);
void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2, qd_composed_field_t *content3);
+void qd_message_compose_5(qd_message_t *msg,
+ qd_composed_field_t *properties,
+ qd_composed_field_t *application_properties,
+ qd_buffer_list_t *body,
+ bool complete);
+
/** Put string representation of a message suitable for logging in buffer.
* @return buffer
*/
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index ea0ca6f..b6ea5b4 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -25,6 +25,8 @@
#include <stdio.h>
#include <inttypes.h>
+static char *address = "echo-service";
+
typedef struct qdr_ref_adaptor_t {
qdr_core_t *core;
qdr_protocol_adaptor_t *adaptor;
@@ -33,6 +35,7 @@ typedef struct qdr_ref_adaptor_t {
qdr_connection_t *conn;
qdr_link_t *out_link;
qdr_link_t *in_link;
+ char *reply_to;
} qdr_ref_adaptor_t;
@@ -61,6 +64,7 @@ static void qdr_ref_first_attach(void *context, qdr_connection_t *conn, qdr_link
static void qdr_ref_second_attach(void *context, qdr_link_t *link,
qdr_terminus_t *source, qdr_terminus_t *target)
{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
#define TERM_SIZE 200
char ftarget[TERM_SIZE];
char fsource[TERM_SIZE];
@@ -79,6 +83,25 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
}
printf("qdr_ref_second_attach: source=%s target=%s\n", fsource, ftarget);
+
+ if (link == adaptor->in_link) {
+ uint64_t link_id;
+ qdr_terminus_t *target = qdr_terminus(0);
+
+ qdr_terminus_set_address(target, address);
+
+ adaptor->out_link = qdr_link_first_attach(adaptor->conn,
+ QD_INCOMING,
+ qdr_terminus(0), //qdr_terminus_t *source,
+ target, //qdr_terminus_t *target,
+ "ref.1", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+
+ qd_iterator_t *reply_iter = qdr_terminus_get_address(source);
+ adaptor->reply_to = (char*) qd_iterator_copy(reply_iter);
+ printf("qdr_ref_second_attach: reply-to=%s\n", adaptor->reply_to);
+ }
}
@@ -95,16 +118,40 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
printf("qdr_ref_flow: %d credits issued\n", credit);
- qd_message_t *msg = qd_message();
- DEQ_INIT(buffers);
- buf = qd_buffer();
- char *insert = (char*) qd_buffer_cursor(buf);
- strcpy(insert, "Test Payload");
- qd_buffer_insert(buf, 13);
- DEQ_INSERT_HEAD(buffers, buf);
- qd_message_compose_1(msg, "echo-service", &buffers);
-
- qdr_link_deliver(adaptor->out_link, msg, 0, false, 0, 0);
+ if (link == adaptor->out_link) {
+ qd_composed_field_t *props = qd_compose(QD_PERFORMATIVE_PROPERTIES, 0);
+ qd_compose_start_list(props);
+ qd_compose_insert_null(props); // message-id
+ qd_compose_insert_null(props); // user-id
+ qd_compose_insert_null(props); // to
+ qd_compose_insert_null(props); // subject
+ qd_compose_insert_string(props, adaptor->reply_to); // reply-to
+ /*
+ qd_compose_insert_null(props); // correlation-id
+ qd_compose_insert_null(props); // content-type
+ qd_compose_insert_null(props); // content-encoding
+ qd_compose_insert_timestamp(props, 0); // absolute-expiry-time
+ qd_compose_insert_timestamp(props, 0); // creation-time
+ qd_compose_insert_null(props); // group-id
+ qd_compose_insert_uint(props, 0); // group-sequence
+ qd_compose_insert_null(props); // reply-to-group-id
+ */
+ qd_compose_end_list(props);
+
+ qd_message_t *msg = qd_message();
+ DEQ_INIT(buffers);
+ buf = qd_buffer();
+ char *insert = (char*) qd_buffer_cursor(buf);
+ strcpy(insert, "Test Payload");
+ qd_buffer_insert(buf, 13);
+ DEQ_INSERT_HEAD(buffers, buf);
+
+ qd_message_compose_5(msg, props, 0, &buffers, true);
+ qd_compose_free(props);
+
+ qdr_delivery_t *dlv = qdr_link_deliver(adaptor->out_link, msg, 0, false, 0, 0);
+ qdr_delivery_decref(adaptor->core, dlv, "release protection of return from deliver");
+ }
}
@@ -208,16 +255,7 @@ static void on_startup(void *context)
uint64_t link_id;
qdr_terminus_t *dynamic_source = qdr_terminus(0);
qdr_terminus_set_dynamic(dynamic_source);
- qdr_terminus_t *target = qdr_terminus(0);
- qdr_terminus_set_address(target, "echo-service");
-
- adaptor->out_link = qdr_link_first_attach(adaptor->conn,
- QD_INCOMING,
- qdr_terminus(0), //qdr_terminus_t *source,
- target, //qdr_terminus_t *target,
- "ref.1", //const char *name,
- 0, //const char *terminus_addr,
- &link_id);
+
adaptor->in_link = qdr_link_first_attach(adaptor->conn,
QD_OUTGOING,
dynamic_source, //qdr_terminus_t *source,
@@ -246,6 +284,7 @@ static void on_activate(void *context)
void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
{
qdr_ref_adaptor_t *adaptor = NEW(qdr_ref_adaptor_t);
+ ZERO(adaptor);
adaptor->core = core;
adaptor->adaptor = qdr_protocol_adaptor(core,
"reference", // name
@@ -280,6 +319,7 @@ void qdr_ref_adaptor_final(void *adaptor_context)
qdr_protocol_adaptor_free(adaptor->core, adaptor->adaptor);
qd_timer_free(adaptor->startup_timer);
qd_timer_free(adaptor->activate_timer);
+ free(adaptor->reply_to);
free(adaptor);
}
diff --git a/src/message.c b/src/message.c
index 2ce6890..81ba863 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2179,6 +2179,33 @@ void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *field1, qd_com
}
+void qd_message_compose_5(qd_message_t *msg,
+ qd_composed_field_t *properties,
+ qd_composed_field_t *application_properties,
+ qd_buffer_list_t *body,
+ bool complete)
+{
+ qd_message_content_t *content = MSG_CONTENT(msg);
+ qd_buffer_list_t *properties_buffers = properties ? qd_compose_buffers(properties) : 0;
+ qd_buffer_list_t *application_properties_buffers = application_properties ? qd_compose_buffers(application_properties) : 0;
+
+ DEQ_INIT(content->buffers);
+ if (properties_buffers)
+ DEQ_APPEND(content->buffers, (*properties_buffers));
+ if (application_properties_buffers)
+ DEQ_APPEND(content->buffers, (*application_properties_buffers));
+
+ if (body) {
+ qd_composed_field_t *field = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
+ qd_compose_insert_binary_buffers(field, body);
+ DEQ_APPEND(content->buffers, (*qd_compose_buffers(field)));
+ qd_compose_free(field);
+ }
+
+ content->receive_complete = complete;
+}
+
+
qd_parsed_field_t *qd_message_get_ingress (qd_message_t *msg)
{
return ((qd_message_pvt_t*)msg)->content->ma_pf_ingress;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 10/32: Dataplane: Renamed tcp_adaptor to
reference_adaptor. Added more test content to the reference adaptor. It now
sends messages to a fixed address. Fixed qdr_terminus_format to show the
dynamically-assigned address for dynamis termini.
Posted by tr...@apache.org.
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 3b58b218cce9b66cc2f291710724ea377d7e9bbb
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 14:43:22 2020 -0400
Dataplane: Renamed tcp_adaptor to reference_adaptor. Added more test content to the reference adaptor. It now sends messages to a fixed address. Fixed qdr_terminus_format to show the dynamically-assigned address for dynamis termini.
---
src/CMakeLists.txt | 2 +-
src/adaptors/reference_adaptor.c | 280 +++++++++++++++++++++++++++++++++++++++
src/adaptors/tcp_adaptor.c | 145 --------------------
src/router_core/terminus.c | 12 +-
4 files changed, 289 insertions(+), 150 deletions(-)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 8c0c7f7..37a61d1 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -38,7 +38,7 @@ add_custom_command (
# Build the qpid-dispatch library.
set(qpid_dispatch_SOURCES
- adaptors/tcp_adaptor.c
+ adaptors/reference_adaptor.c
alloc_pool.c
amqp.c
bitmask.c
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
new file mode 100644
index 0000000..7bbc45d
--- /dev/null
+++ b/src/adaptors/reference_adaptor.c
@@ -0,0 +1,280 @@
+/*
+ * 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 "qpid/dispatch/timer.h"
+#include "qpid/dispatch/message.h"
+#include <stdio.h>
+#include <inttypes.h>
+
+typedef struct qdr_ref_adaptor_t {
+ qdr_core_t *core;
+ qdr_protocol_adaptor_t *adaptor;
+ qd_timer_t *timer;
+ int sequence;
+ qdr_connection_t *conn;
+ qdr_link_t *out_link;
+ qdr_link_t *in_link;
+} qdr_ref_adaptor_t;
+
+
+void qdr_ref_connection_activate_CT(void *context, qdr_connection_t *conn)
+{
+ //
+ // Don't do this here, use a zero-length timer to defer to an IO thread.
+ //
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+
+ while (qdr_connection_process(adaptor->conn)) {}
+}
+
+
+static void qdr_ref_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_ref_second_attach(void *context, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target)
+{
+ char ftarget[100];
+ char fsource[100];
+
+ ftarget[0] = '\0';
+ fsource[0] = '\0';
+
+ if (!!source) {
+ size_t size = 100;
+ qdr_terminus_format(source, fsource, &size);
+ }
+
+ if (!!target) {
+ size_t size = 100;
+ qdr_terminus_format(target, ftarget, &size);
+ }
+
+ printf("qdr_ref_second_attach: source=%s target=%s\n", fsource, ftarget);
+}
+
+
+static void qdr_ref_detach(void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close)
+{
+}
+
+
+static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
+{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+ qd_buffer_list_t buffers;
+ qd_buffer_t *buf;
+
+ printf("qdr_ref_flow: %d credits issued\n", credit);
+
+ qd_message_t *msg = qd_message();
+ DEQ_INIT(buffers);
+ buf = qd_buffer();
+ char *insert = (char*) qd_buffer_cursor(buf);
+ strcpy(insert, "Test Payload");
+ qd_buffer_insert(buf, 13);
+ DEQ_INSERT_HEAD(buffers, buf);
+ qd_message_compose_1(msg, "echo-service", &buffers);
+
+ qdr_link_deliver(adaptor->out_link, msg, 0, false, 0, 0);
+}
+
+
+static void qdr_ref_offer(void *context, qdr_link_t *link, int delivery_count)
+{
+}
+
+
+static void qdr_ref_drained(void *context, qdr_link_t *link)
+{
+}
+
+
+static void qdr_ref_drain(void *context, qdr_link_t *link, bool mode)
+{
+}
+
+
+static int qdr_ref_push(void *context, qdr_link_t *link, int limit)
+{
+ return 0;
+}
+
+
+static uint64_t qdr_ref_deliver(void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled)
+{
+ return 0;
+}
+
+
+static int qdr_ref_get_credit(void *context, qdr_link_t *link)
+{
+ return 0;
+}
+
+
+static void qdr_ref_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled)
+{
+ char *dispname;
+
+ switch (disp) {
+ case PN_ACCEPTED: dispname = "ACCEPTED"; break;
+ case PN_REJECTED: dispname = "REJECTED"; break;
+ case PN_RELEASED: dispname = "RELEASED"; break;
+ case PN_MODIFIED: dispname = "MODIFIED"; break;
+ default:
+ dispname = "<UNKNOWN>";
+ }
+ printf("qdr_ref_delivery_update: disp=%s settled=%s\n", dispname, settled ? "true" : "false");
+}
+
+
+static void qdr_ref_conn_close(void *context, qdr_connection_t *conn, qdr_error_t *error)
+{
+}
+
+
+static void qdr_ref_conn_trace(void *context, qdr_connection_t *conn, bool trace)
+{
+}
+
+
+static void on_timer(void *context)
+{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+
+ if (adaptor->sequence == 0) {
+ qdr_connection_info_t *info = qdr_connection_info(false, //bool is_encrypted,
+ false, //bool is_authenticated,
+ true, //bool opened,
+ "", //char *sasl_mechanisms,
+ QD_INCOMING, //qd_direction_t dir,
+ "127.0.0.1:47756", //const char *host,
+ "", //const char *ssl_proto,
+ "", //const char *ssl_cipher,
+ "", //const char *user,
+ "", //const char *container,
+ pn_data(0), //pn_data_t *connection_properties,
+ 0, //int ssl_ssf,
+ false, //bool ssl,
+ // set if remote is a qdrouter
+ 0); //const qdr_router_version_t *version)
+
+ adaptor->conn = qdr_connection_opened(adaptor->core,
+ adaptor->adaptor,
+ true,
+ QDR_ROLE_NORMAL,
+ 1,
+ 10000, // get this from qd_connection_t
+ 0,
+ 0,
+ false,
+ false,
+ false,
+ false,
+ 250,
+ 0,
+ info,
+ 0,
+ 0);
+
+ uint64_t link_id;
+ qdr_terminus_t *dynamic_source = qdr_terminus(0);
+ qdr_terminus_set_dynamic(dynamic_source);
+ qdr_terminus_t *target = qdr_terminus(0);
+ qdr_terminus_set_address(target, "echo-service");
+
+ adaptor->out_link = qdr_link_first_attach(adaptor->conn,
+ QD_INCOMING,
+ qdr_terminus(0), //qdr_terminus_t *source,
+ target, //qdr_terminus_t *target,
+ "ref.1", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+ adaptor->in_link = qdr_link_first_attach(adaptor->conn,
+ QD_OUTGOING,
+ dynamic_source, //qdr_terminus_t *source,
+ qdr_terminus(0), //qdr_terminus_t *target,
+ "ref.2", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+ adaptor->sequence++;
+ }
+
+ qd_timer_schedule(adaptor->timer, 1000);
+ qdr_connection_process(adaptor->conn);
+}
+
+
+/**
+ * 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_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
+{
+ qdr_ref_adaptor_t *adaptor = NEW(qdr_ref_adaptor_t);
+ adaptor->core = core;
+ adaptor->adaptor = qdr_protocol_adaptor(core,
+ "reference", // name
+ adaptor, // context
+ qdr_ref_connection_activate_CT,
+ qdr_ref_first_attach,
+ qdr_ref_second_attach,
+ qdr_ref_detach,
+ qdr_ref_flow,
+ qdr_ref_offer,
+ qdr_ref_drained,
+ qdr_ref_drain,
+ qdr_ref_push,
+ qdr_ref_deliver,
+ qdr_ref_get_credit,
+ qdr_ref_delivery_update,
+ qdr_ref_conn_close,
+ qdr_ref_conn_trace);
+ *adaptor_context = adaptor;
+
+ // TEMPORARY //
+ adaptor->timer = qd_timer(core->qd, on_timer, adaptor);
+ adaptor->sequence = 0;
+ qd_timer_schedule(adaptor->timer, 0);
+}
+
+
+static void qdr_ref_adaptor_final(void *adaptor_context)
+{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) adaptor_context;
+ qdr_protocol_adaptor_free(adaptor->core, adaptor->adaptor);
+ qd_timer_free(adaptor->timer);
+ free(adaptor);
+}
+
+/**
+ * Declare the adaptor so that it will self-register on process startup.
+ */
+QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
diff --git a/src/adaptors/tcp_adaptor.c b/src/adaptors/tcp_adaptor.c
deleted file mode 100644
index ab5dc48..0000000
--- a/src/adaptors/tcp_adaptor.c
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
- * 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", // name
- adaptor, // context
- 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/terminus.c b/src/router_core/terminus.c
index 9674987..921baf8 100644
--- a/src/router_core/terminus.c
+++ b/src/router_core/terminus.c
@@ -92,13 +92,17 @@ void qdr_terminus_format(qdr_terminus_t *term, char *output, size_t *size)
break;
}
- if (term->dynamic)
- len = safe_snprintf(output, *size, "<dynamic>");
- else if (term->address && term->address->iterator) {
+ if (term->dynamic) {
+ len = safe_snprintf(output, *size, "(dyn)");
+ output += len;
+ *size -= len;
+ }
+
+ if (term->address && term->address->iterator) {
qd_iterator_reset_view(term->address->iterator, ITER_VIEW_ALL);
len = qd_iterator_ncopy(term->address->iterator, (unsigned char*) output, *size - 1);
output[len] = 0;
- } else if (term->address == 0)
+ } else
len = safe_snprintf(output, *size, "<none>");
output += len;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 16/32: Dataplane: Added message method to set
send-complete. Added reference code to receive messages (non streamed).
Posted by tr...@apache.org.
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 9bf4ffd5ed2f391de05971106668047b1aa8c4c7
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jun 4 09:05:32 2020 -0400
Dataplane: Added message method to set send-complete. Added reference code to receive messages (non streamed).
---
include/qpid/dispatch/message.h | 6 ++++++
src/adaptors/reference_adaptor.c | 32 ++++++++++++++++++++++++++------
src/message.c | 15 +++++++++++----
3 files changed, 43 insertions(+), 10 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 030fc0c..29b2335 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -363,6 +363,12 @@ bool qd_message_receive_complete(qd_message_t *msg);
bool qd_message_send_complete(qd_message_t *msg);
/**
+ * Flag the message as being send-complete.
+ */
+void qd_message_set_send_complete(qd_message_t *msg);
+
+
+/**
* Returns true if the delivery tag has already been sent.
*/
bool qd_message_tag_sent(qd_message_t *msg);
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 95801f2..1680e8d 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -25,7 +25,7 @@
#include <stdio.h>
#include <inttypes.h>
-static char *address = "echo-service";
+static char *address = "examples";
typedef struct qdr_ref_adaptor_t {
qdr_core_t *core;
@@ -98,6 +98,8 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
0, //const char *terminus_addr,
&link_id);
+ qdr_link_flow(adaptor->core, adaptor->out_link, 10, false);
+
qd_iterator_t *reply_iter = qdr_terminus_get_address(source);
adaptor->reply_to = (char*) qd_iterator_copy(reply_iter);
printf("qdr_ref_second_attach: reply-to=%s\n", adaptor->reply_to);
@@ -142,8 +144,8 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
DEQ_INIT(buffers);
buf = qd_buffer();
char *insert = (char*) qd_buffer_cursor(buf);
- strcpy(insert, "Test Payload");
- qd_buffer_insert(buf, 13);
+ memcpy(insert, "\x00\x53\x77\xa1\x0cTest Payload", 17);
+ qd_buffer_insert(buf, 17);
DEQ_INSERT_HEAD(buffers, buf);
qd_message_compose_5(msg, props, &buffers, true);
@@ -172,19 +174,37 @@ static void qdr_ref_drain(void *context, qdr_link_t *link, bool mode)
static int qdr_ref_push(void *context, qdr_link_t *link, int limit)
{
- return 0;
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+ printf("qdr_ref_push: limit=%d\n", limit);
+ return qdr_link_process_deliveries(adaptor->core, link, limit);
}
static uint64_t qdr_ref_deliver(void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled)
{
- return 0;
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+ qd_message_t *msg = qdr_delivery_message(delivery);
+
+ qd_message_depth_status_t status = qd_message_check_depth(msg, QD_DEPTH_BODY);
+
+ if (status == QD_MESSAGE_DEPTH_OK) {
+ qd_iterator_t *body_iter = qd_message_field_iterator(msg, QD_FIELD_BODY);
+ char *body = (char*) qd_iterator_copy(body_iter);
+ printf("qdr_ref_deliver: message received, body=%s\n", body);
+ free(body);
+ qd_iterator_free(body_iter);
+ qd_message_set_send_complete(msg);
+ }
+
+ qdr_link_flow(adaptor->core, link, 1, false);
+
+ return PN_ACCEPTED; // This will cause the delivery to be settled
}
static int qdr_ref_get_credit(void *context, qdr_link_t *link)
{
- return 0;
+ return 8;
}
diff --git a/src/message.c b/src/message.c
index 06399cf..be0c7a8 100644
--- a/src/message.c
+++ b/src/message.c
@@ -1222,6 +1222,16 @@ bool qd_message_send_complete(qd_message_t *in_msg)
return msg->send_complete;
}
+
+void qd_message_set_send_complete(qd_message_t *in_msg)
+{
+ if (!!in_msg) {
+ qd_message_pvt_t *msg = (qd_message_pvt_t*) in_msg;
+ msg->send_complete = true;
+ }
+}
+
+
bool qd_message_tag_sent(qd_message_t *in_msg)
{
if (!in_msg)
@@ -2192,10 +2202,7 @@ void qd_message_compose_5(qd_message_t *msg,
DEQ_APPEND(content->buffers, (*headers_buffers));
if (body) {
- qd_composed_field_t *field = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
- qd_compose_insert_binary_buffers(field, body);
- DEQ_APPEND(content->buffers, (*qd_compose_buffers(field)));
- qd_compose_free(field);
+ DEQ_APPEND(content->buffers, (*body));
}
content->receive_complete = complete;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 20/32: Dataplane: Updates to the message-extend
(return buffer count for flow control). Added bidirectional streaming test
to ref adaptor.
Posted by tr...@apache.org.
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 1463337dcc13a73b357fbfe15b5d8f0f8dba34a8
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jun 8 09:49:37 2020 -0400
Dataplane: Updates to the message-extend (return buffer count for flow control). Added bidirectional streaming test to ref adaptor.
---
include/qpid/dispatch/message.h | 8 ++-
src/adaptors/reference_adaptor.c | 73 ++++++++++++++++------
src/message.c | 5 +-
.../streaming_link_scrubber.c | 2 +-
4 files changed, 67 insertions(+), 21 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 0f63b1a..1a7bce2 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -294,9 +294,15 @@ void qd_message_compose_5(qd_message_t *msg,
bool complete);
/**
+ * qd_message_extend
+ *
* Extend the content of a streaming message with more buffers.
+ *
+ * @param msg Pointer to a message
+ * @param buffers A list of buffers to be appended to the end of the message's stream
+ * @return The number of buffers stored in the message's content
*/
-void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers);
+int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers);
/** Put string representation of a message suitable for logging in buffer.
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index f227542..7a65c17 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -37,6 +37,7 @@ typedef struct qdr_ref_adaptor_t {
qdr_connection_t *conn;
qdr_link_t *out_link_1;
qdr_link_t *out_link_2;
+ qdr_link_t *in_link_2;
qdr_link_t *dynamic_in_link;
char *reply_to;
qd_message_t *streaming_message;
@@ -123,6 +124,16 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
"ref.2", //const char *name,
0, //const char *terminus_addr,
&link_id);
+
+ source = qdr_terminus(0);
+ qdr_terminus_set_address(source, address2);
+ adaptor->in_link_2 = qdr_link_first_attach(adaptor->conn,
+ QD_OUTGOING,
+ source, //qdr_terminus_t *source,
+ qdr_terminus(0), //qdr_terminus_t *target,
+ "ref.3", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
}
}
@@ -193,6 +204,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
qd_message_compose_5(adaptor->streaming_message, props, 0, false);
qd_compose_free(props);
+ printf("qdr_ref_flow: Starting a streaming delivery\n");
adaptor->streaming_delivery =
qdr_link_deliver(adaptor->out_link_2, adaptor->streaming_message, 0, false, 0, 0);
adaptor->stream_count = 0;
@@ -231,20 +243,34 @@ static uint64_t qdr_ref_deliver(void *context, qdr_link_t *link, qdr_delivery_t
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
qd_message_t *msg = qdr_delivery_message(delivery);
- qd_message_depth_status_t status = qd_message_check_depth(msg, QD_DEPTH_BODY);
-
- if (status == QD_MESSAGE_DEPTH_OK) {
- qd_iterator_t *body_iter = qd_message_field_iterator(msg, QD_FIELD_BODY);
- char *body = (char*) qd_iterator_copy(body_iter);
- printf("qdr_ref_deliver: message received, body=%s\n", body);
- free(body);
- qd_iterator_free(body_iter);
- qd_message_set_send_complete(msg);
+ qd_message_depth_status_t status = qd_message_check_depth(msg, QD_DEPTH_APPLICATION_PROPERTIES);
+
+ switch (status) {
+ case QD_MESSAGE_DEPTH_OK: {
+ if (qd_message_receive_complete(msg)) {
+ qd_iterator_t *body_iter = qd_message_field_iterator(msg, QD_FIELD_BODY);
+ char *body = (char*) qd_iterator_copy(body_iter);
+ printf("qdr_ref_deliver: complete message received, body=%s\n", body);
+ free(body);
+ qd_iterator_free(body_iter);
+ qd_message_set_send_complete(msg);
+ qdr_link_flow(adaptor->core, link, 1, false);
+ return PN_ACCEPTED; // This will cause the delivery to be settled
+ }
+ break;
}
- qdr_link_flow(adaptor->core, link, 1, false);
+ case QD_MESSAGE_DEPTH_INVALID:
+ printf("qdr_ref_deliver: message invalid\n");
+ qdr_link_flow(adaptor->core, link, 1, false);
+ break;
+
+ case QD_MESSAGE_DEPTH_INCOMPLETE:
+ printf("qdr_ref_deliver: message incomplete\n");
+ break;
+ }
- return PN_ACCEPTED; // This will cause the delivery to be settled
+ return 0;
}
@@ -269,7 +295,12 @@ static void qdr_ref_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t
}
printf("qdr_ref_delivery_update: disp=%s settled=%s\n", dispname, settled ? "true" : "false");
- if (settled && qdr_delivery_link(dlv) == adaptor->out_link_1)
+ if (qdr_delivery_link(dlv) == adaptor->out_link_2 && qdr_delivery_message(dlv) == adaptor->streaming_message) {
+ adaptor->streaming_message = 0;
+ adaptor->stream_count = 0;
+ }
+
+ if (settled)
qdr_delivery_decref(adaptor->core, dlv, "qdr_ref_delivery_update - settled delivery");
}
@@ -352,23 +383,30 @@ static void on_stream(void *context)
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
const char *content = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
const size_t content_length = strlen(content);
- qd_buffer_t *buf = qd_buffer();
- qd_buffer_list_t buffers;
+
+ if (!adaptor->streaming_message)
+ return;
+
+ qd_buffer_t *buf = qd_buffer();
+ qd_buffer_list_t buffers;
DEQ_INIT(buffers);
DEQ_INSERT_TAIL(buffers, buf);
memcpy(qd_buffer_cursor(buf), content, content_length);
qd_buffer_insert(buf, content_length);
- qd_message_extend(adaptor->streaming_message, &buffers);
+ int depth = qd_message_extend(adaptor->streaming_message, &buffers);
qdr_delivery_continue(adaptor->core, adaptor->streaming_delivery, false);
- if (adaptor->stream_count < 30) {
- qd_timer_schedule(adaptor->stream_timer, 1000);
+ if (adaptor->stream_count < 10) {
+ qd_timer_schedule(adaptor->stream_timer, 100);
adaptor->stream_count++;
+ printf("on_stream: sent streamed frame %d, depth=%d\n", adaptor->stream_count, depth);
} else {
qd_message_set_receive_complete(adaptor->streaming_message);
adaptor->streaming_message = 0;
+ adaptor->stream_count = 0;
+ printf("on_stream: completed streaming send, depth=%d\n", depth);
}
}
@@ -404,7 +442,6 @@ void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
qdr_ref_conn_trace);
*adaptor_context = adaptor;
- // TEMPORARY //
adaptor->startup_timer = qd_timer(core->qd, on_startup, adaptor);
qd_timer_schedule(adaptor->startup_timer, 0);
diff --git a/src/message.c b/src/message.c
index 8b833f3..97bcab2 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2218,10 +2218,11 @@ void qd_message_compose_5(qd_message_t *msg,
}
-void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
+int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
{
qd_message_content_t *content = MSG_CONTENT(msg);
qd_buffer_t *buf = DEQ_HEAD(*buffers);
+ int count;
LOCK(content->lock);
while (buf) {
@@ -2230,7 +2231,9 @@ void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
}
DEQ_APPEND(content->buffers, (*buffers));
+ count = DEQ_SIZE(content->buffers);
UNLOCK(content->lock);
+ return count;
}
diff --git a/src/router_core/modules/streaming_link_scrubber/streaming_link_scrubber.c b/src/router_core/modules/streaming_link_scrubber/streaming_link_scrubber.c
index 1c578ca..126a09b 100644
--- a/src/router_core/modules/streaming_link_scrubber/streaming_link_scrubber.c
+++ b/src/router_core/modules/streaming_link_scrubber/streaming_link_scrubber.c
@@ -184,4 +184,4 @@ static void qcm_streaming_link_scrubber_final_CT(void *module_context)
}
-QDR_CORE_MODULE_DECLARE("streaming_link_scruber", qcm_streaming_link_scrubber_enable_CT, qcm_streaming_link_scrubber_init_CT, qcm_streaming_link_scrubber_final_CT)
+QDR_CORE_MODULE_DECLARE("streaming_link_scrubber", qcm_streaming_link_scrubber_enable_CT, qcm_streaming_link_scrubber_init_CT, qcm_streaming_link_scrubber_final_CT)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 30/32: Dataplane: WIP
Posted by tr...@apache.org.
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 5bb50a1c6cd5571b81ab562ec6fe2360c46ce860
Author: Ted Ross <tr...@apache.org>
AuthorDate: Fri Jul 17 14:04:01 2020 -0400
Dataplane: WIP
---
include/qpid/dispatch/message.h | 17 +------------
src/adaptors/reference_adaptor.c | 22 +++++++----------
src/message.c | 28 ++++------------------
src/message_private.h | 1 +
src/python_embedded.c | 2 +-
src/router_core/core_client_api.c | 2 +-
.../modules/test_hooks/core_test_hooks.c | 2 +-
7 files changed, 18 insertions(+), 56 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 5a0b42a..f9e61d1 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -275,26 +275,11 @@ ssize_t qd_message_field_copy(qd_message_t *msg, qd_message_field_t field, char
// Convenience Functions
void qd_message_compose_1(qd_message_t *msg, const char *to, qd_buffer_list_t *buffers);
-void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *content);
+void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *content, bool complete);
void qd_message_compose_3(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2);
void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2, qd_composed_field_t *content3);
/**
- * Send a message with optional headers and an optional raw body with the option of starting
- * a streaming transfer.
- *
- * @param msg The message being composed
- * @param headers A composed field with 1 or more header sections (incl body performative) or NULL
- * @param body A buffer list of raw body content or NULL
- * @param complete True if the message is to be receive-complete.
- * False if more content will arrive later.
- */
-void qd_message_compose_5(qd_message_t *msg,
- qd_composed_field_t *headers,
- qd_buffer_list_t *body,
- bool complete);
-
-/**
* qd_message_extend
*
* Extend the content of a streaming message with more buffers.
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 2e1d98e..a0032a1 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -152,8 +152,6 @@ static void qdr_ref_detach(void *context, qdr_link_t *link, qdr_error_t *error,
static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
{
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
- qd_buffer_list_t buffers;
- qd_buffer_t *buf;
printf("qdr_ref_flow: %d credits issued\n", credit);
@@ -177,15 +175,12 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
*/
qd_compose_end_list(props);
+ props = qd_compose(QD_PERFORMATIVE_BODY_AMQP_VALUE, props);
+ qd_compose_insert_string(props, "Test Payload");
+
qd_message_t *msg = qd_message();
- DEQ_INIT(buffers);
- buf = qd_buffer();
- char *insert = (char*) qd_buffer_cursor(buf);
- memcpy(insert, "\x00\x53\x77\xa1\x0cTest Payload", 17);
- qd_buffer_insert(buf, 17);
- DEQ_INSERT_HEAD(buffers, buf);
-
- qd_message_compose_5(msg, props, &buffers, true);
+
+ qd_message_compose_2(msg, props, true);
qd_compose_free(props);
qdr_link_deliver(adaptor->out_link_1, msg, 0, false, 0, 0);
@@ -205,7 +200,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
adaptor->streaming_message = qd_message();
- qd_message_compose_5(adaptor->streaming_message, props, 0, false);
+ qd_message_compose_2(adaptor->streaming_message, props, false);
qd_compose_free(props);
printf("qdr_ref_flow: Starting a streaming delivery\n");
@@ -409,6 +404,7 @@ static void on_stream(void *context)
qd_buffer_list_t buffer_list;
DEQ_INIT(buffer_list);
qd_buffer_list_append(&buffer_list, (const uint8_t*) content, content_length);
+ qd_buffer_list_append(&buffer_list, (const uint8_t*) content, content_length);
//
// Compose a DATA performative for this section of the stream
@@ -423,7 +419,7 @@ static void on_stream(void *context)
qd_compose_free(field);
//
- // Notify the router that more data has arrived on the delivery
+ // Notify the router that more data is ready to be pushed out on the delivery
//
qdr_delivery_continue(adaptor->core, adaptor->streaming_delivery, false);
}
@@ -493,4 +489,4 @@ void qdr_ref_adaptor_final(void *adaptor_context)
/**
* Declare the adaptor so that it will self-register on process startup.
*/
-//QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
+QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
diff --git a/src/message.c b/src/message.c
index c00b909..2b758b0 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2224,14 +2224,14 @@ void qd_message_compose_1(qd_message_t *msg, const char *to, qd_buffer_list_t *b
}
-void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *field)
+void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *field, bool complete)
{
qd_message_content_t *content = MSG_CONTENT(msg);
- content->receive_complete = true;
-
qd_buffer_list_t *field_buffers = qd_compose_buffers(field);
- content->buffers = *field_buffers;
+ content->buffers = *field_buffers;
+ content->receive_complete = complete;
+
DEQ_INIT(*field_buffers); // Zero out the linkage to the now moved buffers.
}
@@ -2264,26 +2264,6 @@ void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *field1, qd_com
}
-void qd_message_compose_5(qd_message_t *msg,
- qd_composed_field_t *headers,
- qd_buffer_list_t *body,
- bool complete)
-{
- qd_message_content_t *content = MSG_CONTENT(msg);
- qd_buffer_list_t *headers_buffers = headers ? qd_compose_buffers(headers) : 0;
-
- DEQ_INIT(content->buffers);
- if (headers_buffers)
- DEQ_APPEND(content->buffers, (*headers_buffers));
-
- if (body) {
- DEQ_APPEND(content->buffers, (*body));
- }
-
- content->receive_complete = complete;
-}
-
-
int qd_message_extend(qd_message_t *msg, qd_composed_field_t *field)
{
qd_message_content_t *content = MSG_CONTENT(msg);
diff --git a/src/message_private.h b/src/message_private.h
index 47f8f3e..d2b62f7 100644
--- a/src/message_private.h
+++ b/src/message_private.h
@@ -135,6 +135,7 @@ typedef struct {
qd_buffer_list_t ma_trace; // trace list in outgoing message annotations
qd_buffer_list_t ma_ingress; // ingress field in outgoing message annotations
int ma_phase; // phase for the override address
+ qd_field_location_t body_section; // Location of the current parsed body section
bool strip_annotations_in;
bool send_complete; // Has the message been completely received and completely sent?
bool tag_sent; // Tags are sent
diff --git a/src/python_embedded.c b/src/python_embedded.c
index 7265eb2..9465929 100644
--- a/src/python_embedded.c
+++ b/src/python_embedded.c
@@ -764,7 +764,7 @@ static PyObject *qd_python_send(PyObject *self, PyObject *args)
if (compose_python_message(&field, message, ioa->qd) == QD_ERROR_NONE) {
qd_message_t *msg = qd_message();
- qd_message_compose_2(msg, field);
+ qd_message_compose_2(msg, field, true);
qd_composed_field_t *ingress = qd_compose_subfield(0);
qd_compose_insert_string(ingress, qd_router_id(ioa->qd));
diff --git a/src/router_core/core_client_api.c b/src/router_core/core_client_api.c
index 3b2b85c..5b52929 100644
--- a/src/router_core/core_client_api.c
+++ b/src/router_core/core_client_api.c
@@ -687,7 +687,7 @@ static qd_message_t *_create_message_CT(qdrc_client_t *client,
} else if (req->app_properties) {
qd_message_compose_3(message, fld, req->app_properties);
} else {
- qd_message_compose_2(message, fld);
+ qd_message_compose_2(message, fld, true);
}
qd_compose_free(fld);
qd_compose_free(req->body);
diff --git a/src/router_core/modules/test_hooks/core_test_hooks.c b/src/router_core/modules/test_hooks/core_test_hooks.c
index 263b9d9..a810d35 100644
--- a/src/router_core/modules/test_hooks/core_test_hooks.c
+++ b/src/router_core/modules/test_hooks/core_test_hooks.c
@@ -108,7 +108,7 @@ static void source_send(test_endpoint_t *ep, bool presettled)
qd_compose_insert_string(field, stringbuf);
dlv = qdrc_endpoint_delivery_CT(ep->node->core, ep->ep, msg);
- qd_message_compose_2(msg, field);
+ qd_message_compose_2(msg, field, true);
qd_compose_free(field);
qdrc_endpoint_send_CT(ep->node->core, ep->ep, dlv, presettled);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 26/32: Dataplane: WIP changes
Posted by tr...@apache.org.
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 4c11d640a73ab5255485feeae99538a0e876a017
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jul 6 11:23:46 2020 -0400
Dataplane: WIP changes
---
include/qpid/dispatch/buffer.h | 2 +-
include/qpid/dispatch/message.h | 4 ++--
src/adaptors/reference_adaptor.c | 48 +++++++++++++++++++++++++++++-----------
src/buffer.c | 2 +-
src/message.c | 18 +++++++++------
5 files changed, 50 insertions(+), 24 deletions(-)
diff --git a/include/qpid/dispatch/buffer.h b/include/qpid/dispatch/buffer.h
index 715c813..230eebc 100644
--- a/include/qpid/dispatch/buffer.h
+++ b/include/qpid/dispatch/buffer.h
@@ -194,7 +194,7 @@ static inline unsigned char *qd_buffer_at(const qd_buffer_t *buf, size_t len)
* @param data Pointer to raw binary data to be added to the buffer list
* @param len The number of bytes of data to append
*/
-void qd_buffer_list_append(qd_buffer_list_t *buflist, uint8_t *data, size_t len);
+void qd_buffer_list_append(qd_buffer_list_t *buflist, const uint8_t *data, size_t len);
///@}
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index e16d780..5a0b42a 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -300,10 +300,10 @@ void qd_message_compose_5(qd_message_t *msg,
* Extend the content of a streaming message with more buffers.
*
* @param msg Pointer to a message
- * @param buffers A list of buffers to be appended to the end of the message's stream
+ * @param field A composed field to be appended to the end of the message's stream
* @return The number of buffers stored in the message's content
*/
-int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers);
+int qd_message_extend(qd_message_t *msg, qd_composed_field_t *field);
/**
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 59975a9..5dffeca 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -203,8 +203,6 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
qd_compose_insert_string(props, adaptor->reply_to); // reply-to
qd_compose_end_list(props);
- props = qd_compose(QD_PERFORMATIVE_BODY_DATA, props);
-
adaptor->streaming_message = qd_message();
qd_message_compose_5(adaptor->streaming_message, props, 0, false);
@@ -239,7 +237,6 @@ static void qdr_ref_drain(void *context, qdr_link_t *link, bool mode)
static int qdr_ref_push(void *context, qdr_link_t *link, int limit)
{
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
- printf("qdr_ref_push: limit=%d\n", limit);
return qdr_link_process_deliveries(adaptor->core, link, limit);
}
@@ -249,10 +246,13 @@ static uint64_t qdr_ref_deliver(void *context, qdr_link_t *link, qdr_delivery_t
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
qd_message_t *msg = qdr_delivery_message(delivery);
- qd_message_depth_status_t status = qd_message_check_depth(msg, QD_DEPTH_APPLICATION_PROPERTIES);
+ printf("qdr_ref_deliver called\n");
+
+ qd_message_depth_status_t status = qd_message_check_depth(msg, QD_DEPTH_BODY);
switch (status) {
case QD_MESSAGE_DEPTH_OK: {
+ printf("qdr_ref_deliver: depth ok\n");
if (qd_message_receive_complete(msg)) {
qd_iterator_t *body_iter = qd_message_field_iterator(msg, QD_FIELD_BODY);
char *body = (char*) qd_iterator_copy(body_iter);
@@ -391,20 +391,42 @@ static void on_stream(void *context)
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
const char *content = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
const size_t content_length = strlen(content);
+ int depth;
if (!adaptor->streaming_message)
return;
- qd_buffer_t *buf = qd_buffer();
- qd_buffer_list_t buffers;
+ {
+ //
+ // This section shows the proper way to extend a streaming message with content.
+ // Note that the buffer list may be accumulated over the course of many asynchronous
+ // events before it is placed in the composed field and appended to the message stream.
+ //
- DEQ_INIT(buffers);
- DEQ_INSERT_TAIL(buffers, buf);
+ //
+ // Accumulated buffer list
+ //
+ qd_buffer_list_t buffer_list;
+ DEQ_INIT(buffer_list);
+ qd_buffer_list_append(&buffer_list, (const uint8_t*) content, content_length);
- memcpy(qd_buffer_cursor(buf), content, content_length);
- qd_buffer_insert(buf, content_length);
- int depth = qd_message_extend(adaptor->streaming_message, &buffers);
- qdr_delivery_continue(adaptor->core, adaptor->streaming_delivery, false);
+ //
+ // Compose a DATA performative for this section of the stream
+ //
+ qd_composed_field_t *field = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
+ qd_compose_insert_binary_buffers(field, &buffer_list);
+
+ //
+ // Extend the streaming message and free the composed field
+ //
+ depth = qd_message_extend(adaptor->streaming_message, field);
+ qd_compose_free(field);
+
+ //
+ // Notify the router that more data has arrived on the delivery
+ //
+ qdr_delivery_continue(adaptor->core, adaptor->streaming_delivery, false);
+ }
if (adaptor->stream_count < 10) {
qd_timer_schedule(adaptor->stream_timer, 100);
@@ -471,4 +493,4 @@ void qdr_ref_adaptor_final(void *adaptor_context)
/**
* Declare the adaptor so that it will self-register on process startup.
*/
-//QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
+QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
diff --git a/src/buffer.c b/src/buffer.c
index 66a4af2..94315b2 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -108,7 +108,7 @@ unsigned int qd_buffer_list_length(const qd_buffer_list_t *list)
}
-void qd_buffer_list_append(qd_buffer_list_t *buflist, uint8_t *data, size_t len)
+void qd_buffer_list_append(qd_buffer_list_t *buflist, const uint8_t *data, size_t len)
{
//
// If len is zero, there's no work to do.
diff --git a/src/message.c b/src/message.c
index 78f2888..8228cd8 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2218,11 +2218,12 @@ void qd_message_compose_5(qd_message_t *msg,
}
-int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
+int qd_message_extend(qd_message_t *msg, qd_composed_field_t *field)
{
qd_message_content_t *content = MSG_CONTENT(msg);
- qd_buffer_t *buf = DEQ_HEAD(*buffers);
int count;
+ qd_buffer_list_t *buffers = qd_compose_buffers(field);
+ qd_buffer_t *buf = DEQ_HEAD(*buffers);
LOCK(content->lock);
while (buf) {
@@ -2310,13 +2311,16 @@ void qd_message_release_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int bu
}
//
- // Free buffers at the head of the list that have zero refcounts.
+ // Free buffers not at the tail of the list that have zero refcounts.
//
buf = DEQ_HEAD(content->buffers);
- while (buf && qd_buffer_get_fanout(buf) == 0) {
- DEQ_REMOVE_HEAD(content->buffers);
- qd_buffer_free(buf);
- buf = DEQ_HEAD(content->buffers);
+ while (!!buf && buf != DEQ_TAIL(content->buffers)) {
+ qd_buffer_t *next = DEQ_NEXT(buf);
+ if (qd_buffer_get_fanout(buf) == 0) {
+ DEQ_REMOVE(content->buffers, buf);
+ qd_buffer_free(buf);
+ }
+ buf = next;
}
UNLOCK(content->lock);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 28/32: Dataplane: Fixed message parsing so it can
handle partial and streaming content.
Posted by tr...@apache.org.
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 df0568a8d4f03d04fe5046ee4159c330bb4291ca
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jul 15 17:25:35 2020 -0400
Dataplane: Fixed message parsing so it can handle partial and streaming content.
---
src/message.c | 256 +++++++++++++++++++++++++++++++-------------------
src/message_private.h | 8 +-
2 files changed, 165 insertions(+), 99 deletions(-)
diff --git a/src/message.c b/src/message.c
index 7f11e5e..c00b909 100644
--- a/src/message.c
+++ b/src/message.c
@@ -122,7 +122,6 @@ static void quote(char* bytes, int n, char **begin, char *end) {
/**
* Populates the buffer with formatted epoch_time
*/
-//static void format_time(pn_timestamp_t epoch_time, char *format, char *buffer, size_t len)
static void format_time(pn_timestamp_t epoch_time, char *format, char *buffer, size_t len)
{
struct timeval local_timeval;
@@ -369,36 +368,59 @@ char* qd_message_repr(qd_message_t *msg, char* buffer, size_t len, qd_log_bits f
return buffer;
}
+
+/**
+ * Return true if there is at least one consumable octet in the buffer chain
+ * starting at *cursor. If the cursor is beyond the end of the buffer, and there
+ * is another buffer in the chain, move the cursor and buffer pointers to reference
+ * the first octet in the next buffer. Note that this movement does NOT constitute
+ * advancement of the cursor in the buffer chain.
+ */
+static bool can_advance(unsigned char **cursor, qd_buffer_t **buffer)
+{
+ if (qd_buffer_cursor(*buffer) > *cursor)
+ return true;
+
+ if (DEQ_NEXT(*buffer)) {
+ *buffer = DEQ_NEXT(*buffer);
+ *cursor = qd_buffer_base(*buffer);
+ }
+
+ return qd_buffer_cursor(*buffer) > *cursor;
+}
+
+
/**
* Advance cursor through buffer chain by 'consume' bytes.
* Cursor and buffer args are advanced to point to new position in buffer chain.
* - if the number of bytes in the buffer chain is less than or equal to
- * the consume number then set *cursor and *buffer to NULL and
- * return the number of missing bytes
+ * the consume number then return false
* - the original buffer chain is not changed or freed.
*
* @param cursor Pointer into current buffer content
* @param buffer pointer to current buffer
* @param consume number of bytes to advance
- * @return 0 if all bytes consumed, != 0 if not enough bytes available
+ * @return true if all bytes consumed, false if not enough bytes available
*/
-static int advance(unsigned char **cursor, qd_buffer_t **buffer, int consume)
+static bool advance(unsigned char **cursor, qd_buffer_t **buffer, int consume)
{
+ if (!can_advance(cursor, buffer))
+ return false;
+
unsigned char *local_cursor = *cursor;
qd_buffer_t *local_buffer = *buffer;
int remaining = qd_buffer_cursor(local_buffer) - local_cursor;
while (consume > 0) {
- if (consume < remaining) {
+ if (consume <= remaining) {
local_cursor += consume;
consume = 0;
} else {
+ if (!local_buffer->next)
+ return false;
+
consume -= remaining;
local_buffer = local_buffer->next;
- if (local_buffer == 0){
- local_cursor = 0;
- break;
- }
local_cursor = qd_buffer_base(local_buffer);
remaining = qd_buffer_size(local_buffer);
}
@@ -407,7 +429,7 @@ static int advance(unsigned char **cursor, qd_buffer_t **buffer, int consume)
*cursor = local_cursor;
*buffer = local_buffer;
- return consume;
+ return true;
}
@@ -457,21 +479,29 @@ static void advance_guarded(unsigned char **cursor, qd_buffer_t **buffer, int co
}
-static unsigned char next_octet(unsigned char **cursor, qd_buffer_t **buffer)
+/**
+ * If there is an octet to be consumed, put it in octet and return true, else return false.
+ */
+static bool next_octet(unsigned char **cursor, qd_buffer_t **buffer, unsigned char *octet)
{
- unsigned char result = **cursor;
- advance(cursor, buffer, 1);
- return result;
+ if (can_advance(cursor, buffer)) {
+ *octet = **cursor;
+ advance(cursor, buffer, 1);
+ return true;
+ }
+ return false;
}
-static int traverse_field(unsigned char **cursor, qd_buffer_t **buffer, qd_field_location_t *field)
+static bool traverse_field(unsigned char **cursor, qd_buffer_t **buffer, qd_field_location_t *field)
{
qd_buffer_t *start_buffer = *buffer;
unsigned char *start_cursor = *cursor;
+ unsigned char tag;
+ unsigned char octet;
- unsigned char tag = next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
+ if (!next_octet(cursor, buffer, &tag))
+ return false;
int consume = 0;
size_t hdr_length = 1;
@@ -500,23 +530,33 @@ static int traverse_field(unsigned char **cursor, qd_buffer_t **buffer, qd_field
case 0xD0 :
case 0xF0 :
hdr_length += 3;
- consume |= ((int) next_octet(cursor, buffer)) << 24;
- if (!(*cursor)) return 0;
- consume |= ((int) next_octet(cursor, buffer)) << 16;
- if (!(*cursor)) return 0;
- consume |= ((int) next_octet(cursor, buffer)) << 8;
- if (!(*cursor)) return 0;
+ if (!next_octet(cursor, buffer, &octet))
+ return false;
+ consume |= ((int) octet) << 24;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return false;
+ consume |= ((int) octet) << 16;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return false;
+ consume |= ((int) octet) << 8;
+
// Fall through to the next case...
case 0xA0 :
case 0xC0 :
case 0xE0 :
hdr_length++;
- consume |= (int) next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
+ if (!next_octet(cursor, buffer, &octet))
+ return false;
+ consume |= (int) octet;
break;
}
+ if (!advance(cursor, buffer, consume))
+ return false;
+
if (field && !field->parsed) {
field->buffer = start_buffer;
field->offset = start_cursor - qd_buffer_base(start_buffer);
@@ -526,48 +566,58 @@ static int traverse_field(unsigned char **cursor, qd_buffer_t **buffer, qd_field
field->tag = tag;
}
- advance(cursor, buffer, consume);
- return 1;
+ return true;
}
-static int start_list(unsigned char **cursor, qd_buffer_t **buffer)
+static int get_list_count(unsigned char **cursor, qd_buffer_t **buffer)
{
- unsigned char tag = next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
- int length = 0;
- int count = 0;
+ unsigned char tag;
+ unsigned char octet;
+
+ if (!next_octet(cursor, buffer, &tag))
+ return 0;
+
+ int count = 0;
switch (tag) {
case 0x45 : // list0
break;
case 0xd0 : // list32
- length |= ((int) next_octet(cursor, buffer)) << 24;
- if (!(*cursor)) return 0;
- length |= ((int) next_octet(cursor, buffer)) << 16;
- if (!(*cursor)) return 0;
- length |= ((int) next_octet(cursor, buffer)) << 8;
- if (!(*cursor)) return 0;
- length |= (int) next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
-
- count |= ((int) next_octet(cursor, buffer)) << 24;
- if (!(*cursor)) return 0;
- count |= ((int) next_octet(cursor, buffer)) << 16;
- if (!(*cursor)) return 0;
- count |= ((int) next_octet(cursor, buffer)) << 8;
- if (!(*cursor)) return 0;
- count |= (int) next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
+ //
+ // Advance past the list length
+ //
+ if (!advance(cursor, buffer, 4))
+ return 0;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return 0;
+ count |= ((int) octet) << 24;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return 0;
+ count |= ((int) octet) << 16;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return 0;
+ count |= ((int) octet) << 8;
+
+ if (!next_octet(cursor, buffer, &octet))
+ return 0;
+ count |= (int) octet;
break;
case 0xc0 : // list8
- length |= (int) next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
+ //
+ // Advance past the list length
+ //
+ if (!advance(cursor, buffer, 1))
+ return 0;
- count |= (int) next_octet(cursor, buffer);
- if (!(*cursor)) return 0;
+ if (!next_octet(cursor, buffer, &octet))
+ return 0;
+ count |= (int) octet;
break;
}
@@ -609,14 +659,13 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
const unsigned char *expected_tags,
qd_field_location_t *location)
{
- qd_buffer_t *test_buffer = *buffer;
- unsigned char *test_cursor = *cursor;
-
- if (!test_cursor)
+ if (!*cursor || !can_advance(cursor, buffer))
return QD_SECTION_NEED_MORE;
+ qd_buffer_t *test_buffer = *buffer;
+ unsigned char *test_cursor = *cursor;
unsigned char *end_of_buffer = qd_buffer_cursor(test_buffer);
- int idx = 0;
+ int idx = 0;
while (idx < pattern_length && *test_cursor == pattern[idx]) {
idx++;
@@ -656,14 +705,19 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
// Check that the full section is present, if so advance the pointers to
// consume the whole section.
//
- int pre_consume = 1; // Count the already extracted tag
+ int pre_consume = 1; // Count the already extracted tag
uint32_t consume = 0;
- unsigned char tag = next_octet(&test_cursor, &test_buffer);
+ unsigned char tag;
+ unsigned char octet;
+
+ if (!next_octet(&test_cursor, &test_buffer, &tag))
+ return QD_SECTION_NEED_MORE;
+
unsigned char tag_subcat = tag & 0xF0;
// if there is no more data the only valid data type is a null type (0x40),
// size is implied as 0
- if (!test_cursor && tag_subcat != 0x40)
+ if (!can_advance(&test_cursor, &test_buffer) && tag_subcat != 0x40)
return QD_SECTION_NEED_MORE;
switch (tag_subcat) {
@@ -680,12 +734,18 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
case 0xF0:
// uint32_t size field:
pre_consume += 3;
- consume |= ((uint32_t) next_octet(&test_cursor, &test_buffer)) << 24;
- if (!test_cursor) return QD_SECTION_NEED_MORE;
- consume |= ((uint32_t) next_octet(&test_cursor, &test_buffer)) << 16;
- if (!test_cursor) return QD_SECTION_NEED_MORE;
- consume |= ((uint32_t) next_octet(&test_cursor, &test_buffer)) << 8;
- if (!test_cursor) return QD_SECTION_NEED_MORE;
+ if (!next_octet(&test_cursor, &test_buffer, &octet))
+ return QD_SECTION_NEED_MORE;
+ consume |= ((uint32_t) octet) << 24;
+
+ if (!next_octet(&test_cursor, &test_buffer, &octet))
+ return QD_SECTION_NEED_MORE;
+ consume |= ((uint32_t) octet) << 16;
+
+ if (!next_octet(&test_cursor, &test_buffer, &octet))
+ return QD_SECTION_NEED_MORE;
+ consume |= ((uint32_t) octet) << 8;
+
// Fall through to the next case...
case 0xA0:
@@ -693,14 +753,15 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
case 0xE0:
// uint8_t size field
pre_consume += 1;
- consume |= (uint32_t) next_octet(&test_cursor, &test_buffer);
- if (!test_cursor && consume > 0) return QD_SECTION_NEED_MORE;
+ if (!next_octet(&test_cursor, &test_buffer, &octet))
+ return QD_SECTION_NEED_MORE;
+ consume |= (uint32_t) octet;
break;
}
location->length = pre_consume + consume;
if (consume) {
- if (advance(&test_cursor, &test_buffer, consume) != 0) {
+ if (!advance(&test_cursor, &test_buffer, consume)) {
return QD_SECTION_NEED_MORE; // whole section not fully received
}
}
@@ -728,7 +789,7 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
start = DEQ_NEXT(start);
}
- location->parsed = 1;
+ location->parsed = 1;
*cursor = test_cursor;
*buffer = test_buffer;
@@ -786,19 +847,19 @@ static qd_field_location_t *qd_message_properties_field(qd_message_t *msg, qd_me
static const intptr_t offsets[] = {
// position of the field's qd_field_location_t in the message content
// object
- (intptr_t) &((qd_message_content_t *)0)->field_message_id,
- (intptr_t) &((qd_message_content_t *)0)->field_user_id,
- (intptr_t) &((qd_message_content_t *)0)->field_to,
- (intptr_t) &((qd_message_content_t *)0)->field_subject,
- (intptr_t) &((qd_message_content_t *)0)->field_reply_to,
- (intptr_t) &((qd_message_content_t *)0)->field_correlation_id,
- (intptr_t) &((qd_message_content_t *)0)->field_content_type,
- (intptr_t) &((qd_message_content_t *)0)->field_content_encoding,
- (intptr_t) &((qd_message_content_t *)0)->field_absolute_expiry_time,
- (intptr_t) &((qd_message_content_t *)0)->field_creation_time,
- (intptr_t) &((qd_message_content_t *)0)->field_group_id,
- (intptr_t) &((qd_message_content_t *)0)->field_group_sequence,
- (intptr_t) &((qd_message_content_t *)0)->field_reply_to_group_id
+ (intptr_t) &((qd_message_content_t*) 0)->field_message_id,
+ (intptr_t) &((qd_message_content_t*) 0)->field_user_id,
+ (intptr_t) &((qd_message_content_t*) 0)->field_to,
+ (intptr_t) &((qd_message_content_t*) 0)->field_subject,
+ (intptr_t) &((qd_message_content_t*) 0)->field_reply_to,
+ (intptr_t) &((qd_message_content_t*) 0)->field_correlation_id,
+ (intptr_t) &((qd_message_content_t*) 0)->field_content_type,
+ (intptr_t) &((qd_message_content_t*) 0)->field_content_encoding,
+ (intptr_t) &((qd_message_content_t*) 0)->field_absolute_expiry_time,
+ (intptr_t) &((qd_message_content_t*) 0)->field_creation_time,
+ (intptr_t) &((qd_message_content_t*) 0)->field_group_id,
+ (intptr_t) &((qd_message_content_t*) 0)->field_group_sequence,
+ (intptr_t) &((qd_message_content_t*) 0)->field_reply_to_group_id
};
// update table above if new fields need to be accessed:
assert(QD_FIELD_MESSAGE_ID <= field && field <= QD_FIELD_REPLY_TO_GROUP_ID);
@@ -810,23 +871,27 @@ static qd_field_location_t *qd_message_properties_field(qd_message_t *msg, qd_me
}
const int index = field - QD_FIELD_MESSAGE_ID;
- qd_field_location_t *const location = (qd_field_location_t *)((char *)content + offsets[index]);
+ qd_field_location_t *const location = (qd_field_location_t*) ((char*) content + offsets[index]);
if (location->parsed)
return location;
// requested field not parsed out. Need to parse out up to the requested field:
qd_buffer_t *buffer = content->section_message_properties.buffer;
unsigned char *cursor = qd_buffer_base(buffer) + content->section_message_properties.offset;
- advance(&cursor, &buffer, content->section_message_properties.hdr_length);
- if (index >= start_list(&cursor, &buffer)) return 0; // properties list too short
+ if (!advance(&cursor, &buffer, content->section_message_properties.hdr_length))
+ return 0;
+ if (index >= get_list_count(&cursor, &buffer))
+ return 0; // properties list too short
int position = 0;
while (position < index) {
- qd_field_location_t *f = (qd_field_location_t *)((char *)content + offsets[position]);
- if (f->parsed)
- advance(&cursor, &buffer, f->hdr_length + f->length);
- else // parse it out
- if (!traverse_field(&cursor, &buffer, f)) return 0;
+ qd_field_location_t *f = (qd_field_location_t*) ((char*) content + offsets[position]);
+ if (f->parsed) {
+ if (!advance(&cursor, &buffer, f->hdr_length + f->length))
+ return 0;
+ } else // parse it out
+ if (!traverse_field(&cursor, &buffer, f))
+ return 0;
position++;
}
@@ -1862,7 +1927,7 @@ static qd_message_depth_status_t message_check_depth_LH(qd_message_content_t *co
return QD_MESSAGE_DEPTH_INCOMPLETE;
// no more data is going to come. OK if at the end and optional:
- if (!content->parse_cursor && optional)
+ if (!can_advance(&content->parse_cursor, &content->parse_buffer) && optional)
return QD_MESSAGE_DEPTH_OK;
// otherwise we've got an invalid (truncated) header
@@ -2063,7 +2128,8 @@ qd_iterator_t *qd_message_field_iterator(qd_message_t *msg, qd_message_field_t f
qd_buffer_t *buffer = loc->buffer;
unsigned char *cursor = qd_buffer_base(loc->buffer) + loc->offset;
- advance(&cursor, &buffer, loc->hdr_length);
+ if (!advance(&cursor, &buffer, loc->hdr_length))
+ return 0;
return qd_iterator_buffer(buffer, cursor - qd_buffer_base(buffer), loc->length, ITER_VIEW_ALL);
}
diff --git a/src/message_private.h b/src/message_private.h
index 3d3c59e..47f8f3e 100644
--- a/src/message_private.h
+++ b/src/message_private.h
@@ -93,11 +93,11 @@ typedef struct {
qd_field_location_t field_group_id;
qd_field_location_t field_group_sequence;
qd_field_location_t field_reply_to_group_id;
-
qd_field_location_t body; // The body of the message
- qd_buffer_t *parse_buffer;
- unsigned char *parse_cursor;
- qd_message_depth_t parse_depth;
+
+ qd_buffer_t *parse_buffer; // Pointer to the buffer where parsing should resume, if needed
+ unsigned char *parse_cursor; // Pointer to octet in parse_buffer where parsing should resume, if needed
+ qd_message_depth_t parse_depth; // The depth to which this message content has been parsed
qd_iterator_t *ma_field_iter_in; // 'message field iterator' for msg.FIELD_MESSAGE_ANNOTATION
qd_iterator_pointer_t ma_user_annotation_blob; // Original user annotations
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 12/32: Dataplane: Updated the reference adaptor to
implement connection activation
Posted by tr...@apache.org.
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 9545f352aa13032c3433d704abe08a45d7bf5919
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 15:18:26 2020 -0400
Dataplane: Updated the reference adaptor to implement connection activation
---
src/adaptors/reference_adaptor.c | 150 +++++++++++++++++++++------------------
1 file changed, 79 insertions(+), 71 deletions(-)
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 7bbc45d..b4d4afa 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -28,8 +28,8 @@
typedef struct qdr_ref_adaptor_t {
qdr_core_t *core;
qdr_protocol_adaptor_t *adaptor;
- qd_timer_t *timer;
- int sequence;
+ qd_timer_t *startup_timer;
+ qd_timer_t *activate_timer;
qdr_connection_t *conn;
qdr_link_t *out_link;
qdr_link_t *in_link;
@@ -39,11 +39,15 @@ typedef struct qdr_ref_adaptor_t {
void qdr_ref_connection_activate_CT(void *context, qdr_connection_t *conn)
{
//
- // Don't do this here, use a zero-length timer to defer to an IO thread.
+ // Use a zero-delay timer to defer this call to an IO thread
+ //
+ // Note that this may not be generally safe to do. There's no guarantee that multiple
+ // activations won't schedule multiple IO threads running this code concurrently.
+ // Normally, we would rely on assurances provided by the IO scheduler (Proton) that no
+ // connection shall ever be served by more than one thread concurrently.
//
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
-
- while (qdr_connection_process(adaptor->conn)) {}
+ qd_timer_schedule(adaptor->activate_timer, 0);
}
@@ -162,70 +166,72 @@ static void qdr_ref_conn_trace(void *context, qdr_connection_t *conn, bool trace
}
-static void on_timer(void *context)
+static void on_startup(void *context)
{
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
- if (adaptor->sequence == 0) {
- qdr_connection_info_t *info = qdr_connection_info(false, //bool is_encrypted,
- false, //bool is_authenticated,
- true, //bool opened,
- "", //char *sasl_mechanisms,
- QD_INCOMING, //qd_direction_t dir,
- "127.0.0.1:47756", //const char *host,
- "", //const char *ssl_proto,
- "", //const char *ssl_cipher,
- "", //const char *user,
- "", //const char *container,
- pn_data(0), //pn_data_t *connection_properties,
- 0, //int ssl_ssf,
- false, //bool ssl,
- // set if remote is a qdrouter
- 0); //const qdr_router_version_t *version)
-
- adaptor->conn = qdr_connection_opened(adaptor->core,
- adaptor->adaptor,
- true,
- QDR_ROLE_NORMAL,
- 1,
- 10000, // get this from qd_connection_t
- 0,
- 0,
- false,
- false,
- false,
- false,
- 250,
- 0,
- info,
- 0,
- 0);
-
- uint64_t link_id;
- qdr_terminus_t *dynamic_source = qdr_terminus(0);
- qdr_terminus_set_dynamic(dynamic_source);
- qdr_terminus_t *target = qdr_terminus(0);
- qdr_terminus_set_address(target, "echo-service");
-
- adaptor->out_link = qdr_link_first_attach(adaptor->conn,
- QD_INCOMING,
- qdr_terminus(0), //qdr_terminus_t *source,
- target, //qdr_terminus_t *target,
- "ref.1", //const char *name,
- 0, //const char *terminus_addr,
- &link_id);
- adaptor->in_link = qdr_link_first_attach(adaptor->conn,
- QD_OUTGOING,
- dynamic_source, //qdr_terminus_t *source,
- qdr_terminus(0), //qdr_terminus_t *target,
- "ref.2", //const char *name,
- 0, //const char *terminus_addr,
- &link_id);
- adaptor->sequence++;
- }
+ qdr_connection_info_t *info = qdr_connection_info(false, //bool is_encrypted,
+ false, //bool is_authenticated,
+ true, //bool opened,
+ "", //char *sasl_mechanisms,
+ QD_INCOMING, //qd_direction_t dir,
+ "127.0.0.1:47756", //const char *host,
+ "", //const char *ssl_proto,
+ "", //const char *ssl_cipher,
+ "", //const char *user,
+ "", //const char *container,
+ pn_data(0), //pn_data_t *connection_properties,
+ 0, //int ssl_ssf,
+ false, //bool ssl,
+ // set if remote is a qdrouter
+ 0); //const qdr_router_version_t *version)
+
+ adaptor->conn = qdr_connection_opened(adaptor->core,
+ adaptor->adaptor,
+ true,
+ QDR_ROLE_NORMAL,
+ 1,
+ 10000, // get this from qd_connection_t
+ 0,
+ 0,
+ false,
+ false,
+ false,
+ false,
+ 250,
+ 0,
+ info,
+ 0,
+ 0);
+
+ uint64_t link_id;
+ qdr_terminus_t *dynamic_source = qdr_terminus(0);
+ qdr_terminus_set_dynamic(dynamic_source);
+ qdr_terminus_t *target = qdr_terminus(0);
+ qdr_terminus_set_address(target, "echo-service");
+
+ adaptor->out_link = qdr_link_first_attach(adaptor->conn,
+ QD_INCOMING,
+ qdr_terminus(0), //qdr_terminus_t *source,
+ target, //qdr_terminus_t *target,
+ "ref.1", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+ adaptor->in_link = qdr_link_first_attach(adaptor->conn,
+ QD_OUTGOING,
+ dynamic_source, //qdr_terminus_t *source,
+ qdr_terminus(0), //qdr_terminus_t *target,
+ "ref.2", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+}
+
- qd_timer_schedule(adaptor->timer, 1000);
- qdr_connection_process(adaptor->conn);
+static void on_activate(void *context)
+{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+
+ while (qdr_connection_process(adaptor->conn)) {}
}
@@ -236,7 +242,7 @@ static void on_timer(void *context)
* 1) Register the protocol adaptor with the router-core.
* 2) Prepare the protocol adaptor to be configured.
*/
-static void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
+void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
{
qdr_ref_adaptor_t *adaptor = NEW(qdr_ref_adaptor_t);
adaptor->core = core;
@@ -260,17 +266,19 @@ static void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
*adaptor_context = adaptor;
// TEMPORARY //
- adaptor->timer = qd_timer(core->qd, on_timer, adaptor);
- adaptor->sequence = 0;
- qd_timer_schedule(adaptor->timer, 0);
+ adaptor->startup_timer = qd_timer(core->qd, on_startup, adaptor);
+ qd_timer_schedule(adaptor->startup_timer, 0);
+
+ adaptor->activate_timer = qd_timer(core->qd, on_activate, adaptor);
}
-static void qdr_ref_adaptor_final(void *adaptor_context)
+void qdr_ref_adaptor_final(void *adaptor_context)
{
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) adaptor_context;
qdr_protocol_adaptor_free(adaptor->core, adaptor->adaptor);
- qd_timer_free(adaptor->timer);
+ qd_timer_free(adaptor->startup_timer);
+ qd_timer_free(adaptor->activate_timer);
free(adaptor);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 21/32: Dataplane: Added API for streaming data out
of messages. This commit adds the requirement for Proton raw-connection
support.
Posted by tr...@apache.org.
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 7cb6c07035216060ac48063a66a9e7c87bf6362a
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 9 13:18:45 2020 -0400
Dataplane: Added API for streaming data out of messages. This commit adds the requirement for Proton raw-connection support.
---
include/qpid/dispatch/message.h | 29 +++++++++++++++++++++++++++++
1 file changed, 29 insertions(+)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 1a7bce2..e16d780 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -26,6 +26,7 @@
#include <qpid/dispatch/parse.h>
#include <qpid/dispatch/container.h>
#include <qpid/dispatch/log.h>
+#include <proton/raw_connection.h>
/**@file
* Message representation.
@@ -305,6 +306,34 @@ void qd_message_compose_5(qd_message_t *msg,
int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers);
+/**
+ * qd_message_read_body
+ *
+ * Populate Proton raw buffers from the body section in a streaming fashion (i.e. repeated
+ * invocations yield new seqments of the content stream). The buffers will be left in place
+ * in the message until they are explicitly released.
+ *
+ * @param msg Pointer to a message
+ * @param buffers An array of raw-buffer descriptors to be written
+ * @param buffer_count The number of descriptors supplied in buffers
+ * @return The number of raw buffers written.
+ */
+int qd_message_read_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int buffer_count);
+
+
+/**
+ * qd_message_release_body
+ *
+ * Release buffers that were aliased by Proton raw buffers. The buffers in the message that
+ * have been fully read will have their reference counts decreased so they may be freed
+ *
+ * @param msg Pointer to a message
+ * @param buffers An array of raw-buffer descriptors previously returned by qd_message_read_body
+ * @param buffer_count The number of descriptors in the array that contained data
+ */
+void qd_message_release_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int buffer_count);
+
+
/** Put string representation of a message suitable for logging in buffer.
* @return buffer
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 11/32: Dataplane: connection-activate is now routed
through the protocol adapter that handles the connection.
Posted by tr...@apache.org.
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 f91328c3b80f9c908f93b44dfd14749ee7a209a1
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 15:16:56 2020 -0400
Dataplane: connection-activate is now routed through the protocol adapter that handles the connection.
---
src/router_core/router_core.c | 1 +
src/router_core/router_core_private.h | 1 +
src/router_core/router_core_thread.c | 4 +---
src/router_node.c | 4 ++++
4 files changed, 7 insertions(+), 3 deletions(-)
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index 31ae0ad..25c124a 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -945,6 +945,7 @@ qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
DEQ_ITEM_INIT(adaptor);
adaptor->name = name;
adaptor->user_context = context;
+ adaptor->activate_handler = activate;
adaptor->first_attach_handler = first_attach;
adaptor->second_attach_handler = second_attach;
adaptor->detach_handler = detach;
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 7d8e09e..ac123e5 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -794,6 +794,7 @@ struct qdr_protocol_adaptor_t {
// Callbacks
//
void *user_context;
+ qdr_connection_activate_t activate_handler;
qdr_link_first_attach_t first_attach_handler;
qdr_link_second_attach_t second_attach_handler;
qdr_link_detach_t detach_handler;
diff --git a/src/router_core/router_core_thread.c b/src/router_core/router_core_thread.c
index cc73d01..f55d67e 100644
--- a/src/router_core/router_core_thread.c
+++ b/src/router_core/router_core_thread.c
@@ -86,9 +86,7 @@ static void qdr_activate_connections_CT(qdr_core_t *core)
while (conn) {
DEQ_REMOVE_HEAD_N(ACTIVATE, core->connections_to_activate);
conn->in_activate_list = false;
- sys_mutex_lock(qd_server_get_activation_lock(core->qd->server));
- qd_server_activate((qd_connection_t*) qdr_connection_get_context(conn));
- sys_mutex_unlock(qd_server_get_activation_lock(core->qd->server));
+ conn->protocol_adaptor->activate_handler(conn->protocol_adaptor->user_context, conn);
conn = DEQ_HEAD(core->connections_to_activate);
}
}
diff --git a/src/router_node.c b/src/router_node.c
index 304fb4a..70622c6 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1499,11 +1499,15 @@ qd_router_t *qd_router(qd_dispatch_t *qd, qd_router_mode_t mode, const char *are
static void CORE_connection_activate(void *context, qdr_connection_t *conn)
{
+ qd_router_t *router = (qd_router_t*) context;
+
//
// IMPORTANT: This is the only core callback that is invoked on the core
// thread itself. It must not take locks that could deadlock the core.
//
+ sys_mutex_lock(qd_server_get_activation_lock(router->qd->server));
qd_server_activate((qd_connection_t*) qdr_connection_get_context(conn));
+ sys_mutex_unlock(qd_server_get_activation_lock(router->qd->server));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 23/32: Dataplane: (from gsim) Implementation of
qd_message_read_body.
Posted by tr...@apache.org.
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 20f841a982cab439cc1e902f0b4f961d21133b60
Author: Ted Ross <tr...@apache.org>
AuthorDate: Fri Jun 19 13:19:12 2020 -0400
Dataplane: (from gsim) Implementation of qd_message_read_body.
---
src/message.c | 55 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 55 insertions(+)
diff --git a/src/message.c b/src/message.c
index 97bcab2..55d134e 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2237,6 +2237,61 @@ int qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
}
+int qd_message_read_body(qd_message_t *in_msg, pn_raw_buffer_t* buffers, int length)
+{
+ qd_message_pvt_t *msg = (qd_message_pvt_t*) in_msg;
+ if (!(msg->cursor.buffer && msg->cursor.cursor)) {
+ qd_field_location_t *loc = qd_message_field_location(in_msg, QD_FIELD_BODY);
+ if (!loc || loc->tag == QD_AMQP_NULL)
+ return 0;
+ // TODO: need to actually determine this, could be different if vbin32 sent
+ int preamble = 5;
+ if (loc->offset + preamble < qd_buffer_size(loc->buffer)) {
+ msg->cursor.buffer = loc->buffer;
+ msg->cursor.cursor = qd_buffer_base(loc->buffer) + loc->offset + preamble;
+ } else {
+ msg->cursor.buffer = DEQ_NEXT(loc->buffer);
+ if (!msg->cursor.buffer) return 0;
+ msg->cursor.cursor = qd_buffer_base(msg->cursor.buffer) + ((loc->offset + preamble) - qd_buffer_size(loc->buffer));
+ }
+ }
+
+ qd_buffer_t *buf = msg->cursor.buffer;
+ unsigned char *cursor = msg->cursor.cursor;
+
+ // if we are at the end of the current buffer, try to move to the
+ // next buffer
+ if (cursor == qd_buffer_base(buf) + qd_buffer_size(buf)) {
+ buf = DEQ_NEXT(buf);
+ if (buf) {
+ cursor = qd_buffer_base(buf);
+ msg->cursor.buffer = buf;
+ msg->cursor.cursor = cursor;
+ } else {
+ return 0;
+ }
+ }
+
+ int count;
+ for (count = 0; count < length && buf; count++) {
+ buffers[count].bytes = (char*) qd_buffer_base(buf);
+ buffers[count].capacity = qd_buffer_size(buf);
+ buffers[count].size = qd_buffer_size(buf);
+ buffers[count].offset = cursor - qd_buffer_base(buf);
+ buffers[count].context = (uintptr_t) buf;
+ buf = DEQ_NEXT(buf);
+ if (buf) {
+ cursor = qd_buffer_base(buf);
+ msg->cursor.buffer = buf;
+ msg->cursor.cursor = cursor;
+ } else {
+ msg->cursor.cursor = qd_buffer_base(msg->cursor.buffer) + qd_buffer_size(msg->cursor.buffer);
+ }
+ }
+ return count;
+}
+
+
qd_parsed_field_t *qd_message_get_ingress (qd_message_t *msg)
{
return ((qd_message_pvt_t*)msg)->content->ma_pf_ingress;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 13/32: Dataplane: Fixed order problem in shutting
down the router. Disabled the reference adaptor by default (uncomment the
last line to re-enable). The reference adaptor causes test failures.
Posted by tr...@apache.org.
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 084daa1b8b468cbd1af05694bc775fd095ace3ea
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 16:16:44 2020 -0400
Dataplane: Fixed order problem in shutting down the router. Disabled the reference adaptor by default (uncomment the last line to re-enable). The reference adaptor causes test failures.
---
src/adaptors/reference_adaptor.c | 11 ++++++-----
src/router_node.c | 2 +-
2 files changed, 7 insertions(+), 6 deletions(-)
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index b4d4afa..ea0ca6f 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -61,19 +61,20 @@ static void qdr_ref_first_attach(void *context, qdr_connection_t *conn, qdr_link
static void qdr_ref_second_attach(void *context, qdr_link_t *link,
qdr_terminus_t *source, qdr_terminus_t *target)
{
- char ftarget[100];
- char fsource[100];
+#define TERM_SIZE 200
+ char ftarget[TERM_SIZE];
+ char fsource[TERM_SIZE];
ftarget[0] = '\0';
fsource[0] = '\0';
if (!!source) {
- size_t size = 100;
+ size_t size = TERM_SIZE;
qdr_terminus_format(source, fsource, &size);
}
if (!!target) {
- size_t size = 100;
+ size_t size = TERM_SIZE;
qdr_terminus_format(target, ftarget, &size);
}
@@ -285,4 +286,4 @@ void qdr_ref_adaptor_final(void *adaptor_context)
/**
* Declare the adaptor so that it will self-register on process startup.
*/
-QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
+//QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
diff --git a/src/router_node.c b/src/router_node.c
index 70622c6..06c44d9 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -2008,8 +2008,8 @@ void qd_router_free(qd_router_t *router)
qd_container_set_default_node_type(router->qd, 0, 0, QD_DIST_BOTH);
- qdr_protocol_adaptor_free(router->router_core, amqp_direct_adaptor);
qdr_core_free(router->router_core);
+ qdr_protocol_adaptor_free(router->router_core, amqp_direct_adaptor);
qd_tracemask_free(router->tracemask);
qd_timer_free(router->timer);
sys_mutex_free(router->lock);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 01/32: Dataplane: Moved protocol-adapter functions
from router_core.h into proto_adaptor.h
Posted by tr...@apache.org.
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 1e7d562a65314a5d141a62f27962a8b130a98a0f
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jun 1 12:23:59 2020 -0400
Dataplane: Moved protocol-adapter functions from router_core.h into proto_adaptor.h
---
.../dispatch/{router_core.h => protocol_adaptor.h} | 357 ++----------
include/qpid/dispatch/router_core.h | 626 +--------------------
src/http-libwebsockets.c | 2 +-
src/router_core/router_core_private.h | 2 +-
src/router_node.c | 2 +-
5 files changed, 46 insertions(+), 943 deletions(-)
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/protocol_adaptor.h
similarity index 65%
copy from include/qpid/dispatch/router_core.h
copy to include/qpid/dispatch/protocol_adaptor.h
index 14336d6..33916ab 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -1,5 +1,5 @@
-#ifndef __router_core_h__
-#define __router_core_h__ 1
+#ifndef __protocol_adaptor_h__
+#define __protocol_adaptor_h__ 1
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,137 +19,60 @@
* under the License.
*/
-#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/bitmask.h>
-#include <qpid/dispatch/compose.h>
-#include <qpid/dispatch/parse.h>
-#include <qpid/dispatch/router.h>
+#include <qpid/dispatch/router_core.h>
-
-/**
- * All callbacks in this module shall be invoked on a connection thread from the server thread pool.
- * If the callback needs to perform work on a connection, it will be invoked on a thread that has
- * exclusive access to that connection.
- */
-
-typedef struct qdr_subscription_t qdr_subscription_t;
-typedef struct qdr_connection_t qdr_connection_t;
-typedef struct qdr_link_t qdr_link_t;
-typedef struct qdr_delivery_t qdr_delivery_t;
-typedef struct qdr_terminus_t qdr_terminus_t;
-typedef struct qdr_error_t qdr_error_t;
-typedef struct qdr_connection_info_t qdr_connection_info_t;
-
-typedef enum {
- QD_ROUTER_MODE_STANDALONE, ///< Standalone router. No routing protocol participation
- QD_ROUTER_MODE_INTERIOR, ///< Interior router. Full participation in routing protocol.
- QD_ROUTER_MODE_EDGE, ///< Edge router. No transit-router capability.
- QD_ROUTER_MODE_ENDPOINT ///< No routing except for internal modules (agent, etc.).
-} qd_router_mode_t;
-ENUM_DECLARE(qd_router_mode);
-
-/**
- * Allocate and start an instance of the router core module.
- */
-qdr_core_t *qdr_core(qd_dispatch_t *qd, qd_router_mode_t mode, const char *area, const char *id);
-
-/**
- * Stop and deallocate an instance of the router core.
- */
-void qdr_core_free(qdr_core_t *core);
+typedef struct qdr_protocol_adaptor_t qdr_protocol_adaptor_t;
+typedef struct qdr_connection_t qdr_connection_t;
+typedef struct qdr_link_t qdr_link_t;
+typedef struct qdr_delivery_t qdr_delivery_t;
+typedef struct qdr_terminus_t qdr_terminus_t;
+typedef struct qdr_error_t qdr_error_t;
+typedef struct qdr_connection_info_t qdr_connection_info_t;
/**
******************************************************************************
- * Miscellaneous functions
+ * Callback function definitions
******************************************************************************
*/
/**
- * Drive the core-internal timer every one second.
+ * qdr_connection_activate_t callback
*
- * @param core Pointer to the core object returned by qd_core()
- */
-void qdr_process_tick(qdr_core_t *core);
-
-
-/**
- ******************************************************************************
- * Route table maintenance functions (Router Control)
- ******************************************************************************
+ * Activate a connection with pending work from the core to ensure it will be processed by
+ * the proactor: the core has deliveries on links, disposition updates on deliveries, or
+ * flow updates to be sent across the connection.
+ *
+ * IMPORTANT: This function will be invoked on the core thread. It must never block,
+ * delay, or do any lenghty computation.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param conn The connection object to be activated
*/
-void qdr_core_add_router(qdr_core_t *core, const char *address, int router_maskbit);
-void qdr_core_del_router(qdr_core_t *core, int router_maskbit);
-void qdr_core_set_link(qdr_core_t *core, int router_maskbit, int link_maskbit);
-void qdr_core_remove_link(qdr_core_t *core, int router_maskbit);
-void qdr_core_set_next_hop(qdr_core_t *core, int router_maskbit, int nh_router_maskbit);
-void qdr_core_remove_next_hop(qdr_core_t *core, int router_maskbit);
-void qdr_core_set_cost(qdr_core_t *core, int router_maskbit, int cost);
-void qdr_core_set_valid_origins(qdr_core_t *core, int router_maskbit, qd_bitmask_t *routers);
-void qdr_core_flush_destinations(qdr_core_t *core, int router_maskbit);
-void qdr_core_mobile_seq_advanced(qdr_core_t *core, int router_maskbit);
+typedef void (*qdr_connection_activate_t) (void *context, qdr_connection_t *conn);
-typedef void (*qdr_set_mobile_seq_t) (void *context, int router_maskbit, uint64_t mobile_seq);
-typedef void (*qdr_set_my_mobile_seq_t) (void *context, uint64_t mobile_seq);
-typedef void (*qdr_link_lost_t) (void *context, int link_maskbit);
+typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target,
+ qd_session_class_t);
+typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
+ qdr_terminus_t *source, qdr_terminus_t *target);
+typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
+typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
+typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
+typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
+typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
+typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
+typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
+typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
+typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
+typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
+typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
-void qdr_core_route_table_handlers(qdr_core_t *core,
- void *context,
- qdr_set_mobile_seq_t set_mobile_seq,
- qdr_set_my_mobile_seq_t set_my_mobile_seq,
- qdr_link_lost_t link_lost);
/**
******************************************************************************
- * In-process messaging functions
+ * Protocol adaptor plugin functions
******************************************************************************
*/
-typedef void (*qdr_receive_t) (void *context, qd_message_t *msg, int link_maskbit, int inter_router_cost,
- uint64_t conn_id);
-
-/**
- * qdr_core_subscribe
- *
- * Subscribe an in-process handler to receive messages to a particular address.
- *
- * @param core Pointer to the core module
- * @param address The address of messages to be received
- * @param aclass Address class character
- * @param phase Address phase character ('0' .. '9')
- * @param treatment Treatment for the address if it be being created as a side effect of this call
- * @param in_core True iff the handler is to be run in the context of the core thread
- * @param on_message The handler function
- * @param context The opaque context sent to the handler on all invocations
- * @return Pointer to the subscription object
- */
-qdr_subscription_t *qdr_core_subscribe(qdr_core_t *core,
- const char *address,
- char aclass,
- char phase,
- qd_address_treatment_t treatment,
- bool in_core,
- qdr_receive_t on_message,
- void *context);
-
-void qdr_core_unsubscribe(qdr_subscription_t *sub);
-
-/**
- * qdr_send_to
- *
- * Send a message to a destination. This function is used only by in-process components that
- * create messages to be sent. For these messages, there is no inbound link or delivery.
- * Note also that deliveries sent through this function will be pre-settled.
- *
- * @param core Pointer to the core module
- * @param msg Pointer to the message to be sent. The message will be copied during the call
- * and must be freed by the caller if the caller doesn't need to hold it for later use.
- * @param addr Field iterator describing the address to which the message should be delivered.
- * @param exclude_inprocess If true, the message will not be sent to in-process subscribers.
- * @param control If true, this message is to be treated as control traffic and flow on a control link.
- */
-void qdr_send_to1(qdr_core_t *core, qd_message_t *msg, qd_iterator_t *addr,
- bool exclude_inprocess, bool control);
-void qdr_send_to2(qdr_core_t *core, qd_message_t *msg, const char *addr,
- bool exclude_inprocess, bool control);
/**
@@ -269,21 +192,6 @@ const char *qdr_connection_get_tenant_space(const qdr_connection_t *conn, int *l
int qdr_connection_process(qdr_connection_t *conn);
/**
- * qdr_connection_activate_t callback
- *
- * Activate a connection with pending work from the core to ensure it will be processed by
- * the proactor: the core has deliveries on links, disposition updates on deliveries, or
- * flow updates to be sent across the connection.
- *
- * IMPORTANT: This function will be invoked on the core thread. It must never block,
- * delay, or do any lenghty computation.
- *
- * @param context The context supplied when the callback was registered
- * @param conn The connection object to be activated
- */
-typedef void (*qdr_connection_activate_t) (void *context, qdr_connection_t *conn);
-
-/**
******************************************************************************
* Terminus functions
******************************************************************************
@@ -689,10 +597,11 @@ qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_iterato
qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
qd_iterator_t *ingress, qd_iterator_t *addr,
bool settled, qd_bitmask_t *link_exclusion, int ingress_index,
- uint64_t remote_disposition,
+ int64_t remote_disposition,
pn_data_t *remote_extension_state);
qdr_delivery_t *qdr_link_deliver_to_routed_link(qdr_link_t *link, qd_message_t *msg, bool settled,
const uint8_t *tag, int tag_length,
+ uint64_t disposition, pn_data_t* disposition_state,
uint64_t remote_disposition,
pn_data_t *remote_extension_state);
int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit);
@@ -709,6 +618,7 @@ void qdr_link_flow(qdr_core_t *core, qdr_link_t *link, int credit, bool drain_mo
*/
void qdr_link_set_drained(qdr_core_t *core, qdr_link_t *link);
+
/**
* Write the disposition and state data that has arrived from the remote endpoint to the delivery
*/
@@ -719,23 +629,6 @@ 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);
-typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target,
- qd_session_class_t);
-typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target);
-typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
-typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
-typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
-typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
-typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
-typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
-typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
-typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
-typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
-
void qdr_connection_handlers(qdr_core_t *core,
void *context,
@@ -754,144 +647,6 @@ void qdr_connection_handlers(qdr_core_t *core,
qdr_connection_close_t conn_close,
qdr_connection_trace_t conn_trace);
-/**
- ******************************************************************************
- * Management functions
- ******************************************************************************
- */
-typedef enum {
- QD_ROUTER_CONFIG_ADDRESS,
- QD_ROUTER_CONFIG_LINK_ROUTE,
- QD_ROUTER_CONFIG_AUTO_LINK,
- QD_ROUTER_CONNECTION,
- QD_ROUTER_ROUTER,
- QD_ROUTER_LINK,
- QD_ROUTER_ADDRESS,
- QD_ROUTER_EXCHANGE,
- QD_ROUTER_BINDING,
- QD_ROUTER_FORBIDDEN,
- QD_ROUTER_CONN_LINK_ROUTE
-} qd_router_entity_type_t;
-
-typedef struct qdr_query_t qdr_query_t;
-
-/**
- * qdr_manage_create
- *
- * Request a managed entity to be created in the router core.
- *
- * @param core Pointer to the core object returned by qd_core()
- * @param context An opaque context that will be passed back in the invocation of the response callback
- * @param type The entity type for the create request
- * @param name The name supplied for the entity
- * @param in_body The body of the request message
- * @param out_body A composed field for the body of the response message
- * @param in_conn The identity of the connection over which the mgmt message arrived (0 if config file)
- */
-void qdr_manage_create(qdr_core_t *core, void *context, qd_router_entity_type_t type,
- qd_iterator_t *name, qd_parsed_field_t *in_body, qd_composed_field_t *out_body,
- qd_buffer_list_t body_buffers, uint64_t in_conn);
-
-/**
- * qdr_manage_delete
- *
- * Request the deletion of a managed entity in the router core.
- *
- * @param core Pointer to the core object returned by qd_core()
- * @param context An opaque context that will be passed back in the invocation of the response callback
- * @param type The entity type for the create request
- * @param name The name supplied with the request (or 0 if the identity was supplied)
- * @param identity The identity supplied with the request (or 0 if the name was supplied)
- * @param in_conn The identity of the connection over which the mgmt message arrived (0 if config file)
- */
-void qdr_manage_delete(qdr_core_t *core, void *context, qd_router_entity_type_t type,
- qd_iterator_t *name, qd_iterator_t *identity, uint64_t in_conn);
-
-/**
- * qdr_manage_read
- *
- * Request a read of a managed entity in the router core.
- *
- * @param core Pointer to the core object returned by qd_core()
- * @param context An opaque context that will be passed back in the invocation of the response callback
- * @param type The entity type for the create request
- * @param name The name supplied with the request (or 0 if the identity was supplied)
- * @param identity The identity supplied with the request (or 0 if the name was supplied)
- * @param body A composed field for the body of the response message
- * @param in_conn The identity of the connection over which the mgmt message arrived (0 if config file)
- */
-void qdr_manage_read(qdr_core_t *core, void *context, qd_router_entity_type_t type,
- qd_iterator_t *name, qd_iterator_t *identity, qd_composed_field_t *body,
- uint64_t in_conn);
-
-
-/**
- * qdr_manage_update
- *
- * Request the update of a managed entity in the router core.
- *
- * @param core Pointer to the core object returned by qd_core()
- * @param context An opaque context that will be passed back in the invocation of the response callback
- * @param type The entity type for the update request
- * @param name The name supplied with the request (or 0 if the identity was supplied)
- * @param identity The identity supplied with the request (or 0 if the name was supplied)
- * @param in_body The body of the request message
- * @param out_body A composed field for the body of the response message
- * @param in_conn The identity of the connection over which the mgmt message arrived (0 if config file)
- */
-void qdr_manage_update(qdr_core_t *core, void *context, qd_router_entity_type_t type,
- qd_iterator_t *name, qd_iterator_t *identity,
- qd_parsed_field_t *in_body, qd_composed_field_t *out_body,
- uint64_t in_conn);
-
-/**
- * Sequence for running a query:
- *
- * 1) Locate the attributeNames field in the body of the QUERY request
- * 2) Create a composed field for the body of the reply message
- * 3) Call qdr_manage_query with the attributeNames field and the response body
- * 4) Start the body map, add the "attributeNames" key
- * 5) Call qdr_query_add_attribute_names. This will add the attribute names list
- * 6) Add the "results" key, start the outer list
- * 7) Call qdr_query_get_first. This will asynchronously add the first inner list.
- * 8) When the qdr_manage_response_t callback is invoked:
- * a) if more is true and count is not exceeded, call qdr_query_get_next
- * b) if more is false or count is exceeded, call qdr_query_free, close the outer list, close the map
- */
-
-qdr_query_t *qdr_manage_query(qdr_core_t *core, void *context, qd_router_entity_type_t type,
- qd_parsed_field_t *attribute_names, qd_composed_field_t *body,
- uint64_t in_conn);
-void qdr_query_add_attribute_names(qdr_query_t *query);
-void qdr_query_get_first(qdr_query_t *query, int offset);
-void qdr_query_get_next(qdr_query_t *query);
-void qdr_query_free(qdr_query_t *query);
-
-typedef void (*qdr_manage_response_t) (void *context, const qd_amqp_error_t *status, bool more);
-void qdr_manage_handler(qdr_core_t *core, qdr_manage_response_t response_handler);
-
-typedef struct {
- uint16_t major;
- uint16_t minor;
- uint16_t patch;
- uint16_t flags;
-#define QDR_ROUTER_VERSION_SNAPSHOT 0x0100
-#define QDR_ROUTER_VERSION_RC 0x0200 // lower byte == RC #
-#define QDR_ROUTER_VERSION_RC_MASK 0x00FF
-} qdr_router_version_t;
-
-// version >= (Major, Minor, Patch)
-#define QDR_ROUTER_VERSION_AT_LEAST(V, MAJOR, MINOR, PATCH) \
- ((V).major > (MAJOR) || ((V).major == (MAJOR) \
- && ((V).minor > (MINOR) || ((V).minor == (MINOR) \
- && (V).patch >= (PATCH)) \
- ) \
- ) \
- )
-
-// version < (Major, Minor, Patch)
-#define QDR_ROUTER_VERSION_LESS_THAN(V, MAJOR, MINOR, PATCH) \
- (!QDR_ROUTER_VERSION_AT_LEAST(V, MAJOR, MINOR, PATCH))
qdr_connection_info_t *qdr_connection_info(bool is_encrypted,
@@ -911,32 +666,4 @@ qdr_connection_info_t *qdr_connection_info(bool is_encrypted,
const qdr_router_version_t *version);
-typedef struct {
- size_t connections;
- size_t links;
- size_t addrs;
- size_t routers;
- size_t link_routes;
- size_t auto_links;
- size_t presettled_deliveries;
- size_t dropped_presettled_deliveries;
- size_t accepted_deliveries;
- size_t rejected_deliveries;
- size_t released_deliveries;
- size_t modified_deliveries;
- size_t deliveries_ingress;
- size_t deliveries_egress;
- size_t deliveries_transit;
- size_t deliveries_ingress_route_container;
- size_t deliveries_egress_route_container;
- size_t deliveries_delayed_1sec;
- size_t deliveries_delayed_10sec;
- size_t deliveries_stuck;
- size_t links_blocked;
- size_t deliveries_redirected_to_fallback;
-} qdr_global_stats_t;
-ALLOC_DECLARE(qdr_global_stats_t);
-typedef void (*qdr_global_stats_handler_t) (void *context);
-void qdr_request_global_stats(qdr_core_t *core, qdr_global_stats_t *stats, qdr_global_stats_handler_t callback, void *context);
-
#endif
diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index 14336d6..0a0c7c5 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -33,12 +33,6 @@
*/
typedef struct qdr_subscription_t qdr_subscription_t;
-typedef struct qdr_connection_t qdr_connection_t;
-typedef struct qdr_link_t qdr_link_t;
-typedef struct qdr_delivery_t qdr_delivery_t;
-typedef struct qdr_terminus_t qdr_terminus_t;
-typedef struct qdr_error_t qdr_error_t;
-typedef struct qdr_connection_info_t qdr_connection_info_t;
typedef enum {
QD_ROUTER_MODE_STANDALONE, ///< Standalone router. No routing protocol participation
@@ -154,608 +148,6 @@ void qdr_send_to2(qdr_core_t *core, qd_message_t *msg, const char *addr,
/**
******************************************************************************
- * Connection functions
- ******************************************************************************
- */
-
-typedef enum {
- QD_LINK_ENDPOINT, ///< A link to a connected endpoint
- QD_LINK_CONTROL, ///< A link to a peer router for control messages
- QD_LINK_ROUTER, ///< A link to a peer router for routed messages
- QD_LINK_EDGE_DOWNLINK ///< Default link from an interior router to an edge router
-} qd_link_type_t;
-
-typedef enum {
- QDR_ROLE_NORMAL,
- QDR_ROLE_INTER_ROUTER,
- QDR_ROLE_ROUTE_CONTAINER,
- QDR_ROLE_EDGE_CONNECTION
-} qdr_connection_role_t;
-
-typedef void (*qdr_connection_bind_context_t) (qdr_connection_t *context, void* token);
-
-/**
- * qdr_connection_opened
- *
- * This function must be called once for every connection that is opened in the router.
- * Once a new connection has been both remotely and locally opened, the core must be notified.
- *
- * @param core Pointer to the core object
- * @param incoming True iff this connection is associated with a listener, False if a connector
- * @param role The configured role of this connection
- * @param cost If the role is inter_router, this is the configured cost for the connection.
- * @param management_id - A unique identifier that is used in management and logging operations.
- * @param label Optional label provided in the connection's configuration. This is used to
- * correlate the connection with waypoints and link-route destinations that use the connection.
- * @param strip_annotations_in True if configured to remove annotations on inbound messages.
- * @param strip_annotations_out True if configured to remove annotations on outbound messages.
- * @param policy_allow_dynamic_link_routes True if this connection is allowed by policy to create link route destinations.
- * @param policy_allow_admin_status_update True if this connection is allowed to modify admin_status on other connections.
- * @param link_capacity The capacity, in deliveries, for links in this connection.
- * @param vhost If non-null, this is the vhost of the connection to be used for multi-tenancy.
- * @return Pointer to a connection object that can be used to refer to this connection over its lifetime.
- */
-qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
- bool incoming,
- qdr_connection_role_t role,
- int cost,
- uint64_t management_id,
- const char *label,
- const char *remote_container_id,
- bool strip_annotations_in,
- bool strip_annotations_out,
- bool policy_allow_dynamic_link_routes,
- bool policy_allow_admin_status_update,
- int link_capacity,
- const char *vhost,
- qdr_connection_info_t *connection_info,
- qdr_connection_bind_context_t context_binder,
- void* bind_token);
-
-/**
- * qdr_connection_closed
- *
- * This function must be called when a connection is closed, either cleanly by protocol
- * or uncleanly by lost connectivity. Once this functino is called, the caller must never
- * again refer to or use the connection pointer.
- *
- * @param conn The pointer returned by qdr_connection_opened
- */
-void qdr_connection_closed(qdr_connection_t *conn);
-
-bool qdr_connection_route_container(qdr_connection_t *conn);
-
-/**
- * qdr_connection_set_context
- *
- * Store an arbitrary void pointer in the connection object.
- */
-void qdr_connection_set_context(qdr_connection_t *conn, void *context);
-
-/**
- * qdr_connection_get_context
- *
- * Retrieve the stored void pointer from the connection object.
- */
-void *qdr_connection_get_context(const qdr_connection_t *conn);
-
-
-/**
- * qdr_connection_role
- *
- * Retrieve the role of the connection object.
- */
-qdr_connection_role_t qdr_connection_role(const qdr_connection_t *conn);
-
-
-/**
- * qdr_connection_get_tenant_space
- *
- * Retrieve the multi-tenant space for a connection. Returns 0 if there is
- * no multi-tenancy on this connection.
- */
-const char *qdr_connection_get_tenant_space(const qdr_connection_t *conn, int *len);
-
-/**
- * qdr_connection_process
- *
- * Allow the core to process work associated with this connection.
- * This function MUST be called on a thread that exclusively owns
- * this connection.
- *
- * @param conn The pointer returned by qdr_connection_opened
- * @return The number of actions processed.
- */
-int qdr_connection_process(qdr_connection_t *conn);
-
-/**
- * qdr_connection_activate_t callback
- *
- * Activate a connection with pending work from the core to ensure it will be processed by
- * the proactor: the core has deliveries on links, disposition updates on deliveries, or
- * flow updates to be sent across the connection.
- *
- * IMPORTANT: This function will be invoked on the core thread. It must never block,
- * delay, or do any lenghty computation.
- *
- * @param context The context supplied when the callback was registered
- * @param conn The connection object to be activated
- */
-typedef void (*qdr_connection_activate_t) (void *context, qdr_connection_t *conn);
-
-/**
- ******************************************************************************
- * Terminus functions
- ******************************************************************************
- */
-
-/**
- * qdr_terminus
- *
- * Create a qdr_terminus_t that contains all the content of the
- * pn_terminus_t. Note that the pointer to the pn_terminus_t
- * _will not_ be held or referenced further after this function
- * returns.
- *
- * @param pn Pointer to a proton terminus object that will be copied into
- * the qdr_terminus object
- * @return Pointer to a newly allocated qdr_terminus object
- */
-qdr_terminus_t *qdr_terminus(pn_terminus_t *pn);
-
-/**
- * qdr_terminus_free
- *
- * Free a qdr_terminus object once it is no longer needed.
- *
- * @param terminus The pointer returned by qdr_terminus()
- */
-void qdr_terminus_free(qdr_terminus_t *terminus);
-
-/**
- * qdr_terminus_format
- *
- * Write a human-readable representation of the terminus content to the string
- * in 'output'.
- *
- * @param terminus The pointer returned by qdr_terminus()
- * @param output The string buffer where the result shall be written
- * @param size Input: the number of bytes availabie in output for writing. Output: the
- * number of bytes remaining after the operation.
- */
-void qdr_terminus_format(qdr_terminus_t *terminus, char *output, size_t *size);
-
-/**
- * qdr_terminus_copy
- *
- * Copy the contents of the qdr_terminus into a proton terminus
- *
- * @param from A qdr_terminus pointer returned by qdr_terminus()
- * @param to A proton terminus to be overwritten with the contents
- * of 'from'
- */
-void qdr_terminus_copy(qdr_terminus_t *from, pn_terminus_t *to);
-
-/**
- * qdr_terminus_add_capability
- *
- * Add a capability symbol to the terminus.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param capability A string to be added as a symbol to the capability list
- */
-void qdr_terminus_add_capability(qdr_terminus_t *term, const char *capability);
-
-/**
- * qdr_terminus_has_capability
- *
- * Check to see if a terminus has a particular capability.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param capability A string describing a capability to be checked
- * @return true iff the capability is advertised for this terminus
- */
-bool qdr_terminus_has_capability(qdr_terminus_t *term, const char *capability);
-
-/**
- * qdr_terminus_waypoint_capability
- *
- * If the terminus has a waypoint capability, return the ordinal of the
- * waypoint. If not, return zero.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return 1..9 if the terminus has waypoint capability, 0 otherwise
- */
-int qdr_terminus_waypoint_capability(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_is_anonymous
- *
- * Indicate whether this terminus represents an anonymous endpoint.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return true iff the terminus is anonymous
- */
-bool qdr_terminus_is_anonymous(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_is_coordinator
- *
- * Indicates if the terminus is a coordinator.
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return true iff the terminus is a coordinator
- */
-bool qdr_terminus_is_coordinator(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_is_dynamic
- *
- * Indicate whether this terminus represents a dynamic endpoint.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return true iff the terminus is dynamic
- */
-bool qdr_terminus_is_dynamic(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_survives_disconnect
- *
- * Indicate whether this terminus will survive disconnection (i.e. if
- * state is expected to be kept).
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return true iff the terminus has a timeout greater than 0 or an
- * expiry-policy of never
- */
-bool qdr_terminus_survives_disconnect(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_set_address
- *
- * Set the terminus address
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param addr An AMQP address (null-terminated string)
- */
-void qdr_terminus_set_address(qdr_terminus_t *term, const char *addr);
-void qdr_terminus_set_address_iterator(qdr_terminus_t *term, qd_iterator_t *addr);
-
-/**
- * qdr_terminus_get_address
- *
- * Return the address of the terminus in the form of an iterator.
- * The iterator is borrowed, the caller must not free the iterator.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return A pointer to an iterator or 0 if the terminus is anonymous.
- */
-qd_iterator_t *qdr_terminus_get_address(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_insert_address_prefix
- *
- * Insert the given prefix into the terminus address
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param prefix null-terminated string
- */
-void qdr_terminus_insert_address_prefix(qdr_terminus_t *term, const char *prefix);
-
-/**
- * qdr_terminus_strip_address_prefix
- *
- * Remove the given prefix from the terminus address
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param prefix null-terminated string
- */
-void qdr_terminus_strip_address_prefix(qdr_terminus_t *term, const char *prefix);
-
-/**
- * qdr_terminus_dnp_address
- *
- * Return the address field in the dynamic-node-properties if it is there.
- * This iterator is given, the caller must free it when it is no longer needed.
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @return A pointer to an iterator or 0 if there is no such field.
- */
-qd_iterator_t *qdr_terminus_dnp_address(qdr_terminus_t *term);
-
-/**
- * qdr_terminus_set_dnp_address_iterator
- *
- * Overwrite the dynamic-node-properties.address in the terminus
- *
- * @param term A qdr_terminus pointer returned by qdr_terminus()
- * @param iter An iterator whos view shall be placed in the dnp.address
- */
-void qdr_terminus_set_dnp_address_iterator(qdr_terminus_t *term, qd_iterator_t *iter);
-
-
-/**
- ******************************************************************************
- * Error functions
- ******************************************************************************
- */
-
-qdr_error_t *qdr_error_from_pn(pn_condition_t *pn);
-qdr_error_t *qdr_error(const char *name, const char *description);
-void qdr_error_free(qdr_error_t *error);
-void qdr_error_copy(qdr_error_t *from, pn_condition_t *to);
-char *qdr_error_description(const qdr_error_t *err);
-char *qdr_error_name(const qdr_error_t *err);
-pn_data_t *qdr_error_info(const qdr_error_t *err);
-
-/**
- ******************************************************************************
- * Link functions
- ******************************************************************************
- */
-
-/**
- * qdr_link_set_context
- *
- * Store an arbitrary void pointer in the link object.
- */
-void qdr_link_set_context(qdr_link_t *link, void *context);
-
-/**
- * qdr_link_get_context
- *
- * Retrieve the stored void pointer from the link object.
- */
-void *qdr_link_get_context(const qdr_link_t *link);
-
-/**
- * qdr_link_type
- *
- * Retrieve the link-type from the link object.
- *
- * @param link Link object
- * @return Link-type
- */
-qd_link_type_t qdr_link_type(const qdr_link_t *link);
-
-/**
- * qdr_link_direction
- *
- * Retrieve the link-direction from the link object.
- *
- * @param link Link object
- * @return Link-direction
- */
-qd_direction_t qdr_link_direction(const qdr_link_t *link);
-
-/**
- * qdr_link_phase
- *
- * If this link is associated with an auto_link, return the address phase. Otherwise
- * return zero.
- *
- * @param link Link object
- * @return 0 or the phase of the link's auto_link.
- */
-int qdr_link_phase(const qdr_link_t *link);
-
-/**
- * qdr_link_internal_address
- *
- * If this link is associated with an auto_link and the auto_link has different
- * internal and external addresses, return the internal (routing) address.
- *
- * @param link Link object
- * @return 0 or the auto_link's internal address.
- */
-const char *qdr_link_internal_address(const qdr_link_t *link);
-
-/**
- * qdr_link_is_anonymous
- *
- * Indicate whether the link is anonymous. Note that this is determined inside
- * the core thread. In the time between first creating the link and when the
- * core thread determines its status, a link will indicate "true" for being anonymous.
- * The reason for this is to be conservative. The anonymous check is an optimization
- * used by the caller to skip parsing the "to" field for messages on non-anonymous links.
- *
- * @param link Link object
- * @return True if the link is anonymous or the link hasn't been processed yet.
- */
-bool qdr_link_is_anonymous(const qdr_link_t *link);
-
-/**
- * qdr_link_is_routed
- *
- * Indicate whether the link is link-routed.
- *
- * @param link Link object
- * @return True if the link is link-routed.
- */
-bool qdr_link_is_routed(const qdr_link_t *link);
-
-/**
- * qdr_link_strip_annotations_in
- *
- * Indicate whether the link's connection is configured to strip message annotations on inbound messages.
- */
-bool qdr_link_strip_annotations_in(const qdr_link_t *link);
-
-/**
- * qdr_link_strip_annotations_oout
- *
- * Indicate whether the link's connection is configured to strip message annotations on outbound messages.
- */
-bool qdr_link_strip_annotations_out(const qdr_link_t *link);
-
-/**
- * qdr_link_stalled_outbound
- *
- * Tell the link that it has been stalled outbound due to back-pressure from the
- * transport buffers. Stalling is undone during link-flow processing.
- */
-void qdr_link_stalled_outbound(qdr_link_t *link);
-
-/**
- * qdr_link_name
- *
- * Retrieve the name of the link.
- *
- * @param link Link object
- * @return The link's name
- */
-const char *qdr_link_name(const qdr_link_t *link);
-
-/**
- * qdr_link_first_attach
- *
- * This function is invoked when a first-attach (not a response to an earlier attach)
- * arrives for a connection.
- *
- * @param conn Connection pointer returned by qdr_connection_opened
- * @param dir Direction of the new link, incoming or outgoing
- * @param source Source terminus of the attach
- * @param target Target terminus of the attach
- * @param name - name of the link
- * @param terminus_addr - terminus address if any
- * @param link_id - set to the management id of the new link
- * @return A pointer to a new qdr_link_t object to track the link
- */
-qdr_link_t *qdr_link_first_attach(qdr_connection_t *conn,
- qd_direction_t dir,
- qdr_terminus_t *source,
- qdr_terminus_t *target,
- const char *name,
- const char *terminus_addr,
- uint64_t *link_id);
-
-/**
- * qdr_link_second_attach
- *
- * This function is invoked when a second-attach (a response to an attach we sent)
- * arrives for a connection.
- *
- * @param link The link pointer returned by qdr_link_first_attach or in a FIRST_ATTACH event.
- * @param source Source terminus of the attach
- * @param target Target terminus of the attach
- */
-void qdr_link_second_attach(qdr_link_t *link, qdr_terminus_t *source, qdr_terminus_t *target);
-
-/**
- * qdr_link_detach
- *
- * This function is invoked when a link detach arrives.
- *
- * @param link The link pointer returned by qdr_link_first_attach or in a FIRST_ATTACH event.
- * @param dt The type of detach that occurred.
- * @param error The link error from the detach frame or 0 if none.
- */
-void qdr_link_detach(qdr_link_t *link, qd_detach_type_t dt, qdr_error_t *error);
-
-/**
- * qdr_link_delete
- *
- * Request that the router-core delete this link and free all its associated resources.
- *
- * @param link The link pointer returned by qdr_link_first_attach or in a FIRST_ATTACH event.
- */
-void qdr_link_delete(qdr_link_t *link);
-
-/**
- * qdr_link_deliver
- *
- * Deliver a message to the router core for forwarding. This function is used in cases where
- * the link contains all the information needed for proper message routing (i.e. non-anonymous
- * inbound links).
- *
- * @param link Pointer to the link over which the message arrived.
- * @param msg Pointer to the delivered message. The sender is giving this reference to the router
- * core. The sender _must not_ free or otherwise use the message after invoking this function.
- * @param ingress Field iterator referencing the value of the ingress-router header. NOTE: This
- * iterator is assumed to reference content in the message that will stay valid
- * through the lifetime of the message.
- * @param settled True iff the delivery is pre-settled.
- * @param link_exclusion If present, this is a bitmask of inter-router links that should not be used
- * to send this message. This bitmask is created by the trace_mask module and
- * it built on the trace header from a received message.
- * @param ingress_index The bitmask index of the router that this delivery entered the network through.
- * @param remote_disposition as set by sender on the transfer
- * @param remote_disposition_state as set by sender on the transfer
- * @return Pointer to the qdr_delivery that will track the lifecycle of this delivery on this link.
- */
-qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_iterator_t *ingress,
- bool settled, qd_bitmask_t *link_exclusion, int ingress_index,
- uint64_t remote_disposition,
- pn_data_t *remote_extension_state);
-qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
- qd_iterator_t *ingress, qd_iterator_t *addr,
- bool settled, qd_bitmask_t *link_exclusion, int ingress_index,
- uint64_t remote_disposition,
- pn_data_t *remote_extension_state);
-qdr_delivery_t *qdr_link_deliver_to_routed_link(qdr_link_t *link, qd_message_t *msg, bool settled,
- const uint8_t *tag, int tag_length,
- uint64_t remote_disposition,
- pn_data_t *remote_extension_state);
-int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit);
-
-void qdr_link_flow(qdr_core_t *core, qdr_link_t *link, int credit, bool drain_mode);
-
-/**
- * Sets the link's drain flag to false and sets credit to core to zero.
- * The passed in link has been drained and hence no longer in drain mode.
- * Call this right after calling pn_link_drained
- *
- * @param core - router core
- * @param link - the link that has been drained
- */
-void qdr_link_set_drained(qdr_core_t *core, qdr_link_t *link);
-
-/**
- * Write the disposition and state data that has arrived from the remote endpoint to the delivery
- */
-void qdr_delivery_set_remote_extension_state(qdr_delivery_t *dlv, uint64_t remote_dispo, pn_data_t *remote_ext_state);
-
-/**
- * Extract the disposition and state data that is to be sent to the remote endpoint via the delivery
- */
-pn_data_t *qdr_delivery_take_local_extension_state(qdr_delivery_t *dlv, uint64_t *dispo);
-
-typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target,
- qd_session_class_t);
-typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target);
-typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
-typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
-typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
-typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
-typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
-typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
-typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
-typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
-typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
-typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
-
-
-void qdr_connection_handlers(qdr_core_t *core,
- void *context,
- qdr_connection_activate_t activate,
- qdr_link_first_attach_t first_attach,
- qdr_link_second_attach_t second_attach,
- qdr_link_detach_t detach,
- qdr_link_flow_t flow,
- qdr_link_offer_t offer,
- qdr_link_drained_t drained,
- qdr_link_drain_t drain,
- qdr_link_push_t push,
- qdr_link_deliver_t deliver,
- qdr_link_get_credit_t get_credit,
- qdr_delivery_update_t delivery_update,
- qdr_connection_close_t conn_close,
- qdr_connection_trace_t conn_trace);
-
-/**
- ******************************************************************************
* Management functions
******************************************************************************
*/
@@ -894,23 +286,6 @@ typedef struct {
(!QDR_ROUTER_VERSION_AT_LEAST(V, MAJOR, MINOR, PATCH))
-qdr_connection_info_t *qdr_connection_info(bool is_encrypted,
- bool is_authenticated,
- bool opened,
- char *sasl_mechanisms,
- qd_direction_t dir,
- const char *host,
- const char *ssl_proto,
- const char *ssl_cipher,
- const char *user,
- const char *container,
- pn_data_t *connection_properties,
- int ssl_ssf,
- bool ssl,
- // set if remote is a qdrouter
- const qdr_router_version_t *version);
-
-
typedef struct {
size_t connections;
size_t links;
@@ -939,4 +314,5 @@ ALLOC_DECLARE(qdr_global_stats_t);
typedef void (*qdr_global_stats_handler_t) (void *context);
void qdr_request_global_stats(qdr_core_t *core, qdr_global_stats_t *stats, qdr_global_stats_handler_t callback, void *context);
+
#endif
diff --git a/src/http-libwebsockets.c b/src/http-libwebsockets.c
index 4958dfd..de4aea7 100644
--- a/src/http-libwebsockets.c
+++ b/src/http-libwebsockets.c
@@ -19,7 +19,7 @@
#include <qpid/dispatch/atomic.h>
#include <qpid/dispatch/amqp.h>
-#include <qpid/dispatch/router_core.h>
+#include <qpid/dispatch/protocol_adaptor.h>
#include <qpid/dispatch/threading.h>
#include <qpid/dispatch/timer.h>
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index 707d5f2..49d9650 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -21,7 +21,7 @@
#include "dispatch_private.h"
#include "message_private.h"
-#include <qpid/dispatch/router_core.h>
+#include <qpid/dispatch/protocol_adaptor.h>
#include <qpid/dispatch/threading.h>
#include <qpid/dispatch/atomic.h>
#include <qpid/dispatch/log.h>
diff --git a/src/router_node.c b/src/router_node.c
index 0597ba0..ce59329 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -28,7 +28,7 @@
#include "router_private.h"
#include "delivery.h"
#include "policy.h"
-#include <qpid/dispatch/router_core.h>
+#include <qpid/dispatch/protocol_adaptor.h>
#include <qpid/dispatch/proton_utils.h>
#include <proton/sasl.h>
#include <inttypes.h>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 31/32: Dataplane - Added the body_data data
structure for reading streaming messages. WIP - The following functions (in
message.c) need to be implemented: find_last_buffer
qd_message_body_data_iterator qd_message_body_data_buffer_count
qd_message_body_data_buffers qd_message_body_data_release
Posted by tr...@apache.org.
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 ee033ef6ef931e7fb9c3757bc220bc4ecc42632f
Author: Ted Ross <tr...@apache.org>
AuthorDate: Sun Jul 26 21:02:36 2020 -0400
Dataplane - Added the body_data data structure for reading streaming messages. WIP - The following functions (in message.c) need to be implemented: find_last_buffer qd_message_body_data_iterator qd_message_body_data_buffer_count qd_message_body_data_buffers qd_message_body_data_release
---
include/qpid/dispatch/message.h | 80 ++++++++++++++++++-----
src/adaptors/reference_adaptor.c | 33 ++++++++--
src/message.c | 135 +++++++++++++++++++++++++++++++++++++++
src/message_private.h | 46 ++++++++-----
4 files changed, 258 insertions(+), 36 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index f9e61d1..f409302 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -59,7 +59,8 @@
// Callback for status change (confirmed persistent, loaded-in-memory, etc.)
-typedef struct qd_message_t qd_message_t;
+typedef struct qd_message_t qd_message_t;
+typedef struct qd_message_body_data_t qd_message_body_data_t;
/** Amount of message to be parsed. */
typedef enum {
@@ -292,31 +293,78 @@ int qd_message_extend(qd_message_t *msg, qd_composed_field_t *field);
/**
- * qd_message_read_body
+ * qd_message_body_data_iterator
*
- * Populate Proton raw buffers from the body section in a streaming fashion (i.e. repeated
- * invocations yield new seqments of the content stream). The buffers will be left in place
- * in the message until they are explicitly released.
+ * Return an iterator that references the content (not the performative headers)
+ * of the entire body-data section.
*
- * @param msg Pointer to a message
- * @param buffers An array of raw-buffer descriptors to be written
- * @param buffer_count The number of descriptors supplied in buffers
- * @return The number of raw buffers written.
+ * The returned iterator must eventually be freed by the caller.
+ *
+ * @param body_data Pointer to a body_data object produced by qd_message_next_body_data
+ * @return Pointer to an iterator referencing the body_data content
+ */
+qd_iterator_t *qd_message_body_data_iterator(const qd_message_body_data_t *body_data);
+
+
+/**
+ * qd_message_body_data_buffer_count
+ *
+ * Return the number of buffers that are needed to hold this body-data's content.
+ *
+ * @param body_data Pointer to a body_data object produced by qd_message_next_body_data
+ * @return Number of pn_raw_buffers needed to contain the entire content of this body_data.
+ */
+int qd_message_body_data_buffer_count(const qd_message_body_data_t *body_data);
+
+
+/**
+ * qd_message_body_data_buffers
+ *
+ * Populate an array of pn_raw_buffer_t objects with references to the body_data's content.
+ *
+ * @param body_data Pointer to a body_data object produced by qd_message_next_body_data
+ * @param buffers Pointer to an array of pn_raw_buffer_t objects
+ * @param offset The offset (in the body_data's buffer set) from which copying should begin
+ * @param count The number of pn_raw_buffer_t objects in the buffers array
+ * @return The number of pn_raw_buffer_t objects that were overwritten
+ */
+int qd_message_body_data_buffers(qd_message_body_data_t *body_data, pn_raw_buffer_t *buffers, int offset, int count);
+
+
+/**
+ * qd_message_body_data_release
+ *
+ * Release buffers that were associated with a body-data section. It is not required that body-data
+ * objects be released in the same order in which they were offered.
+ *
+ * Once this function is called, the caller must drop its reference to the body_data object
+ * and not use it again.
+ *
+ * @param body_data Pointer to a body data object returned by qd_message_next_body_data
*/
-int qd_message_read_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int buffer_count);
+void qd_message_body_data_release(qd_message_body_data_t *body_data);
+
+
+typedef enum {
+ QD_MESSAGE_BODY_DATA_OK, // A valid body data object have been returned
+ QD_MESSAGE_BODY_DATA_INCOMPLETE, // The next body data is incomplete, try again later
+ QD_MESSAGE_BODY_DATA_NO_MORE, // There are no more body data objects in this stream
+ QD_MESSAGE_BODY_DATA_INVALID, // The next body data is invalid, the stream is corrupted
+ QD_MESSAGE_BODY_DATA_NOT_DATA // The body of the message is not a DATA segment
+} qd_message_body_data_result_t;
/**
- * qd_message_release_body
+ * qd_message_next_body_data
*
- * Release buffers that were aliased by Proton raw buffers. The buffers in the message that
- * have been fully read will have their reference counts decreased so they may be freed
+ * Get the next body-data section from this streaming message return the result and
+ * possibly the valid, completed body_data object.
*
* @param msg Pointer to a message
- * @param buffers An array of raw-buffer descriptors previously returned by qd_message_read_body
- * @param buffer_count The number of descriptors in the array that contained data
+ * @param body_data Output pointer to a body_data object (or 0 if not OK)
+ * @return The body_data_result describing the result of this operation
*/
-void qd_message_release_body(qd_message_t *msg, pn_raw_buffer_t *buffers, int buffer_count);
+qd_message_body_data_result_t qd_message_next_body_data(qd_message_t *msg, qd_message_body_data_t **body_data);
/** Put string representation of a message suitable for logging in buffer.
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index a0032a1..2a3a975 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -248,22 +248,47 @@ static uint64_t qdr_ref_deliver(void *context, qdr_link_t *link, qdr_delivery_t
switch (status) {
case QD_MESSAGE_DEPTH_OK: {
printf("qdr_ref_deliver: depth ok\n");
- if (qd_message_receive_complete(msg)) {
- qd_iterator_t *body_iter = qd_message_field_iterator(msg, QD_FIELD_BODY);
+ qd_message_body_data_t *body_data;
+ qd_message_body_data_result_t body_data_result;
+ body_data_result = qd_message_next_body_data(msg, &body_data);
+
+ switch (body_data_result) {
+ case QD_MESSAGE_BODY_DATA_OK: {
+ qd_iterator_t *body_iter = qd_message_body_data_iterator(body_data);
char *body = (char*) qd_iterator_copy(body_iter);
- printf("qdr_ref_deliver: complete message received, body=%s\n", body);
+ printf("qdr_ref_deliver: message body-data received: %s\n", body);
free(body);
qd_iterator_free(body_iter);
+ break;
+ }
+
+ case QD_MESSAGE_BODY_DATA_INCOMPLETE:
+ printf("qdr_ref_deliver: body-data incomplete\n");
+ break;
+
+ case QD_MESSAGE_BODY_DATA_NO_MORE:
qd_message_set_send_complete(msg);
qdr_link_flow(adaptor->core, link, 1, false);
return PN_ACCEPTED; // This will cause the delivery to be settled
+
+ case QD_MESSAGE_BODY_DATA_INVALID:
+ printf("qdr_ref_deliver: body-data invalid\n");
+ qdr_link_flow(adaptor->core, link, 1, false);
+ return PN_REJECTED;
+
+ case QD_MESSAGE_BODY_DATA_NOT_DATA:
+ printf("qdr_ref_deliver: body not data\n");
+ qdr_link_flow(adaptor->core, link, 1, false);
+ return PN_REJECTED;
}
+
break;
}
case QD_MESSAGE_DEPTH_INVALID:
printf("qdr_ref_deliver: message invalid\n");
qdr_link_flow(adaptor->core, link, 1, false);
+ return PN_REJECTED;
break;
case QD_MESSAGE_DEPTH_INCOMPLETE:
@@ -489,4 +514,4 @@ void qdr_ref_adaptor_final(void *adaptor_context)
/**
* Declare the adaptor so that it will self-register on process startup.
*/
-QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
+//QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
diff --git a/src/message.c b/src/message.c
index 2b758b0..364ec21 100644
--- a/src/message.c
+++ b/src/message.c
@@ -86,6 +86,7 @@ PN_HANDLE(PN_DELIVERY_CTX)
ALLOC_DEFINE_CONFIG(qd_message_t, sizeof(qd_message_pvt_t), 0, 0);
ALLOC_DEFINE(qd_message_content_t);
+ALLOC_DEFINE(qd_message_body_data_t);
typedef void (*buffer_process_t) (void *context, const unsigned char *base, int length);
@@ -1092,6 +1093,8 @@ qd_message_t *qd_message_copy(qd_message_t *in_msg)
if (!copy)
return 0;
+ ZERO(copy);
+
qd_buffer_list_clone(©->ma_to_override, &msg->ma_to_override);
qd_buffer_list_clone(©->ma_trace, &msg->ma_trace);
qd_buffer_list_clone(©->ma_ingress, &msg->ma_ingress);
@@ -2284,6 +2287,138 @@ int qd_message_extend(qd_message_t *msg, qd_composed_field_t *field)
}
+/**
+ * find_last_buffer
+ *
+ * Given a field location, find the following:
+ *
+ * - *cursor - The pointer to the octet _past_ the last octet in the field. If this is the last octet in
+ * the buffer, the cursor must point one octet past the buffer.
+ * - *buffer - The last buffer that contains content for this field.
+ *
+ * Important: If the last octet of the field is the last octet of a buffer and there are more buffers in the
+ * buffer list, *buffer _must_ refer to the buffer that contains the last octet of the field and *cursor must
+ * point at the octet following that octet, even if it points past the end of the buffer.
+ */
+static void find_last_buffer(qd_field_location_t *location, unsigned char **cursor, qd_buffer_t **buffer)
+{
+ //qd_buffer_t *buf = location->buffer;
+
+}
+
+
+/**
+ * qd_message_body_data_iterator
+ *
+ * Given a body_data object, return an iterator that refers to the content of that body data. This iterator
+ * shall not refer to the 3-byte performative header or the header for the vbin{8,32} field.
+ *
+ * The iterator must be freed eventually by the caller.
+ */
+qd_iterator_t *qd_message_body_data_iterator(const qd_message_body_data_t *body_data)
+{
+ return 0;
+}
+
+
+/**
+ * qd_message_body_data_buffer_count
+ *
+ * Return the number of buffers contained in the body_data object.
+ */
+int qd_message_body_data_buffer_count(const qd_message_body_data_t *body_data)
+{
+ return 0;
+}
+
+
+/**
+ * qd_message_body_data_buffers
+ *
+ * Populate the provided array of pn_raw_buffers with the addresses and lengths of the buffers in the body_data
+ * object. Don't fill more than count raw_buffers with data. Start at offset from the zero-th buffer in the
+ * body_data.
+ */
+int qd_message_body_data_buffers(qd_message_body_data_t *body_data, pn_raw_buffer_t *buffers, int offset, int count)
+{
+ return 0;
+}
+
+
+/**
+ * qd_message_body_data_release
+ *
+ * Decrement the fanout ref-counts for all of the buffers referred to in the body_data. If any have reached zero,
+ * remove them from the buffer list and free them. Never dec-ref the last buffer in the content's buffer list.
+ */
+void qd_message_body_data_release(qd_message_body_data_t *body_data)
+{
+}
+
+
+qd_message_body_data_result_t qd_message_next_body_data(qd_message_t *in_msg, qd_message_body_data_t **out_body_data)
+{
+ qd_message_pvt_t *msg = (qd_message_pvt_t*) in_msg;
+ qd_message_body_data_t *body_data = 0;
+
+ if (!msg->body_cursor) {
+ //
+ // We haven't returned a body-data record for this message yet.
+ //
+ qd_message_depth_status_t status = qd_message_check_depth(in_msg, QD_DEPTH_BODY);
+ if (status == QD_MESSAGE_DEPTH_OK) {
+ body_data = new_qd_message_body_data_t();
+ ZERO(body_data);
+ body_data->owning_message = msg;
+ body_data->section = msg->content->section_body;
+
+ find_last_buffer(&body_data->section, &msg->body_cursor, &msg->body_buffer);
+ body_data->last_buffer = msg->body_buffer;
+
+ assert(DEQ_SIZE(msg->body_data_list) == 0);
+ DEQ_INSERT_TAIL(msg->body_data_list, body_data);
+ *out_body_data = body_data;
+ return QD_MESSAGE_BODY_DATA_OK;
+ } else if (status == QD_MESSAGE_DEPTH_INCOMPLETE)
+ return QD_MESSAGE_BODY_DATA_INCOMPLETE;
+ else if (status == QD_MESSAGE_DEPTH_INVALID)
+ return QD_MESSAGE_BODY_DATA_INVALID;
+ }
+
+ qd_section_status_t section_status;
+ qd_field_location_t location;
+
+ section_status = message_section_check(&msg->body_buffer, &msg->body_cursor,
+ BODY_DATA_SHORT, 3, TAGS_BINARY,
+ &location);
+
+ switch (section_status) {
+ case QD_SECTION_INVALID:
+ case QD_SECTION_NO_MATCH:
+ return QD_MESSAGE_BODY_DATA_INVALID;
+
+ case QD_SECTION_MATCH:
+ body_data = new_qd_message_body_data_t();
+ ZERO(body_data);
+ body_data->owning_message = msg;
+ body_data->section = location;
+ find_last_buffer(&body_data->section, &msg->body_cursor, &msg->body_buffer);
+ body_data->last_buffer = msg->body_buffer;
+ DEQ_INSERT_TAIL(msg->body_data_list, body_data);
+ *out_body_data = body_data;
+ return QD_MESSAGE_BODY_DATA_OK;
+
+ case QD_SECTION_NEED_MORE:
+ if (msg->content->receive_complete)
+ return QD_MESSAGE_BODY_DATA_NO_MORE;
+ else
+ return QD_MESSAGE_BODY_DATA_INCOMPLETE;
+ }
+
+ return QD_MESSAGE_BODY_DATA_NO_MORE;
+}
+
+
int qd_message_read_body(qd_message_t *in_msg, pn_raw_buffer_t* buffers, int length)
{
qd_message_pvt_t *msg = (qd_message_pvt_t*) in_msg;
diff --git a/src/message_private.h b/src/message_private.h
index d2b62f7..6dc901a 100644
--- a/src/message_private.h
+++ b/src/message_private.h
@@ -24,6 +24,8 @@
#include <qpid/dispatch/threading.h>
#include <qpid/dispatch/atomic.h>
+typedef struct qd_message_pvt_t qd_message_pvt_t;
+
/** @file
* Message representation.
*
@@ -59,6 +61,16 @@ typedef struct {
} qd_field_location_t;
+struct qd_message_body_data_t {
+ DEQ_LINKS(qd_message_body_data_t); // Linkage to form a DEQ
+ qd_message_pvt_t *owning_message; // Pointer to the owning message
+ qd_field_location_t section; // Section descriptor for the field
+ qd_buffer_t *last_buffer; // Pointer to the last buffer in the field
+};
+
+ALLOC_DECLARE(qd_message_body_data_t);
+DEQ_DECLARE(qd_message_body_data_t, qd_message_body_data_list_t);
+
// TODO - consider using pointers to qd_field_location_t below to save memory
// TODO - provide a way to allocate a message without a lock for the link-routing case.
// It's likely that link-routing will cause no contention for the message content.
@@ -93,7 +105,6 @@ typedef struct {
qd_field_location_t field_group_id;
qd_field_location_t field_group_sequence;
qd_field_location_t field_reply_to_group_id;
- qd_field_location_t body; // The body of the message
qd_buffer_t *parse_buffer; // Pointer to the buffer where parsing should resume, if needed
unsigned char *parse_cursor; // Pointer to octet in parse_buffer where parsing should resume, if needed
@@ -126,21 +137,24 @@ typedef struct {
uint8_t priority; // The priority of this message
} qd_message_content_t;
-typedef struct {
- qd_iterator_pointer_t cursor; // A pointer to the current location of the outgoing byte stream.
- qd_message_depth_t message_depth; // What is the depth of the message that has been received so far
- qd_message_depth_t sent_depth; // How much of the message has been sent? QD_DEPTH_NONE means nothing has been sent so far, QD_DEPTH_HEADER means the header has already been sent, dont send it again and so on.
- qd_message_content_t *content; // The actual content of the message. The content is never copied
- qd_buffer_list_t ma_to_override; // to field in outgoing message annotations.
- qd_buffer_list_t ma_trace; // trace list in outgoing message annotations
- qd_buffer_list_t ma_ingress; // ingress field in outgoing message annotations
- int ma_phase; // phase for the override address
- qd_field_location_t body_section; // Location of the current parsed body section
- bool strip_annotations_in;
- bool send_complete; // Has the message been completely received and completely sent?
- bool tag_sent; // Tags are sent
- bool is_fanout; // If msg is an outgoing fanout
-} qd_message_pvt_t;
+struct qd_message_pvt_t {
+ qd_iterator_pointer_t cursor; // A pointer to the current location of the outgoing byte stream.
+ qd_message_depth_t message_depth; // What is the depth of the message that has been received so far
+ qd_message_depth_t sent_depth; // How much of the message has been sent? QD_DEPTH_NONE means nothing has been sent so far, QD_DEPTH_HEADER means the header has already been sent, dont send it again and so on.
+ qd_message_content_t *content; // The actual content of the message. The content is never copied
+ qd_buffer_list_t ma_to_override; // to field in outgoing message annotations.
+ qd_buffer_list_t ma_trace; // trace list in outgoing message annotations
+ qd_buffer_list_t ma_ingress; // ingress field in outgoing message annotations
+ int ma_phase; // phase for the override address
+ qd_message_body_data_list_t body_data_list; // TODO - move this to the content for one-time parsing (TLR)
+ qd_message_body_data_t *next_body_data;
+ unsigned char *body_cursor;
+ qd_buffer_t *body_buffer;
+ bool strip_annotations_in;
+ bool send_complete; // Has the message been completely received and completely sent?
+ bool tag_sent; // Tags are sent
+ bool is_fanout; // If msg is an outgoing fanout
+};
ALLOC_DECLARE(qd_message_t);
ALLOC_DECLARE(qd_message_content_t);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 08/32: Dataplane: minor cleanup
Posted by tr...@apache.org.
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 1ab9b7fcd6ac8303c75d8a652643966f5296af84
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 13:46:54 2020 -0400
Dataplane: minor cleanup
---
include/qpid/dispatch/protocol_adaptor.h | 2 +-
src/adaptors/tcp_adaptor.c | 6 +++---
src/router_node.c | 2 +-
3 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index cea625b..b96dc41 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -334,7 +334,7 @@ typedef enum {
QDR_ROLE_EDGE_CONNECTION
} qdr_connection_role_t;
-typedef void (*qdr_connection_bind_context_t) (qdr_connection_t *context, void* token);
+typedef void (*qdr_connection_bind_context_t) (qdr_connection_t *context, void *token);
/**
* qdr_connection_opened
diff --git a/src/adaptors/tcp_adaptor.c b/src/adaptors/tcp_adaptor.c
index 4ec9884..ab5dc48 100644
--- a/src/adaptors/tcp_adaptor.c
+++ b/src/adaptors/tcp_adaptor.c
@@ -112,9 +112,9 @@ 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
+ "tcp", // name
+ adaptor, // context
+ 0, // activate
qdr_tcp_first_attach,
qdr_tcp_second_attach,
qdr_tcp_detach,
diff --git a/src/router_node.c b/src/router_node.c
index 90a8f1b..304fb4a 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1977,7 +1977,7 @@ void qd_router_setup_late(qd_dispatch_t *qd)
qd->router->router_core = qdr_core(qd, qd->router->router_mode, qd->router->router_area, qd->router->router_id);
amqp_direct_adaptor = qdr_protocol_adaptor(qd->router->router_core,
- "amqp-direct",
+ "amqp",
(void*) qd->router,
CORE_connection_activate,
CORE_link_first_attach,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 17/32: Dataplane: Exposed access to connection-ids
from server. Moved the generation of the "connection opened" log from
router_node.c to the core module. This causes the log to be raised for all
protocol adaptors.
Posted by tr...@apache.org.
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 c2810643cd8f11258c268b2df09c9ee21a63ade1
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jun 4 11:58:46 2020 -0400
Dataplane: Exposed access to connection-ids from server. Moved the generation of the "connection opened" log from router_node.c to the core module. This causes the log to be raised for all protocol adaptors.
---
src/adaptors/reference_adaptor.c | 65 ++++++++++++++++++++--------------------
src/router_core/connections.c | 12 ++++++++
src/router_node.c | 11 -------
src/server.c | 11 +++++++
src/server_private.h | 2 ++
5 files changed, 57 insertions(+), 44 deletions(-)
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 1680e8d..c7ca07c 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -238,39 +238,38 @@ static void on_startup(void *context)
{
qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
- qdr_connection_info_t *info = qdr_connection_info(false, //bool is_encrypted,
- false, //bool is_authenticated,
- true, //bool opened,
- "", //char *sasl_mechanisms,
- QD_INCOMING, //qd_direction_t dir,
- "127.0.0.1:47756", //const char *host,
- "", //const char *ssl_proto,
- "", //const char *ssl_cipher,
- "", //const char *user,
- "", //const char *container,
- pn_data(0), //pn_data_t *connection_properties,
- 0, //int ssl_ssf,
- false, //bool ssl,
- // set if remote is a qdrouter
- 0); //const qdr_router_version_t *version)
-
- adaptor->conn = qdr_connection_opened(adaptor->core,
- adaptor->adaptor,
- true,
- QDR_ROLE_NORMAL,
- 1,
- 10000, // get this from qd_connection_t
- 0,
- 0,
- false,
- false,
- false,
- false,
- 250,
- 0,
- info,
- 0,
- 0);
+ qdr_connection_info_t *info = qdr_connection_info(false, //bool is_encrypted,
+ false, //bool is_authenticated,
+ true, //bool opened,
+ "", //char *sasl_mechanisms,
+ QD_INCOMING, //qd_direction_t dir,
+ "", //const char *host,
+ "", //const char *ssl_proto,
+ "", //const char *ssl_cipher,
+ "", //const char *user,
+ "reference-adaptor", //const char *container,
+ pn_data(0), //pn_data_t *connection_properties,
+ 0, //int ssl_ssf,
+ false, //bool ssl,
+ 0); //const qdr_router_version_t *version)
+
+ adaptor->conn = qdr_connection_opened(adaptor->core, // core
+ adaptor->adaptor, // protocol_adaptor
+ true, // incoming
+ QDR_ROLE_NORMAL, // role
+ 1, // cost
+ qd_server_allocate_connection_id(adaptor->core->qd->server),
+ 0, // label
+ 0, // remote_container_id
+ false, // strip_annotations_in
+ false, // strip_annotations_out
+ false, // policy_allow_dynamic_link_routes
+ false, // policy_allow_admin_status_update
+ 250, // link_capacity
+ 0, // vhost
+ info, // connection_info
+ 0, // context_binder
+ 0); // bind_token
uint64_t link_id;
qdr_terminus_t *dynamic_source = qdr_terminus(0);
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 67ee1cb..d6c7e40 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -128,6 +128,18 @@ qdr_connection_t *qdr_connection_opened(qdr_core_t *core,
action->args.connection.container_id = qdr_field(remote_container_id);
qdr_action_enqueue(core, action);
+ char props_str[1000];
+ size_t props_len = 1000;
+
+ pn_data_format(connection_info->connection_properties, props_str, &props_len);
+
+ qd_log(core->log, QD_LOG_INFO, "[C%"PRIu64"] Connection Opened: dir=%s host=%s vhost=%s encrypted=%s"
+ " auth=%s user=%s container_id=%s props=%s",
+ management_id, incoming ? "in" : "out",
+ connection_info->host, vhost ? vhost : "", connection_info->is_encrypted ? connection_info->ssl_proto : "no",
+ connection_info->is_authenticated ? connection_info->sasl_mechanisms : "no",
+ connection_info->user, connection_info->container, props_str);
+
return conn;
}
diff --git a/src/router_node.c b/src/router_node.c
index 06c44d9..95f9619 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1219,11 +1219,6 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
bind_connection_context,
conn);
- char props_str[1000];
- size_t props_len = 1000;
-
- pn_data_format(props, props_str, &props_len);
-
if (conn->connector) {
char conn_msg[300];
qd_format_string(conn_msg, 300, "[C%"PRIu64"] Connection Opened: dir=%s host=%s vhost=%s encrypted=%s"
@@ -1232,12 +1227,6 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
authenticated ? mech : "no", (char*) user, container);
strcpy(conn->connector->conn_msg, conn_msg);
}
-
-
- qd_log(router->log_source, QD_LOG_INFO, "[C%"PRIu64"] Connection Opened: dir=%s host=%s vhost=%s encrypted=%s"
- " auth=%s user=%s container_id=%s props=%s",
- connection_id, inbound ? "in" : "out", host, vhost ? vhost : "", encrypted ? proto : "no",
- authenticated ? mech : "no", (char*) user, container, props_str);
}
diff --git a/src/server.c b/src/server.c
index 27505b7..83b0802 100644
--- a/src/server.c
+++ b/src/server.c
@@ -1685,6 +1685,17 @@ bool qd_connection_handle(qd_connection_t *c, pn_event_t *e) {
return true;
}
+
+uint64_t qd_server_allocate_connection_id(qd_server_t *server)
+{
+ uint64_t id;
+ sys_mutex_lock(server->lock);
+ id = server->next_connection_id++;
+ sys_mutex_unlock(server->lock);
+ return id;
+}
+
+
bool qd_connection_strip_annotations_in(const qd_connection_t *c) {
return c->strip_annotations_in;
}
diff --git a/src/server_private.h b/src/server_private.h
index 005f7ad..48a08bb 100644
--- a/src/server_private.h
+++ b/src/server_private.h
@@ -45,6 +45,8 @@ qd_connector_t* qd_connection_connector(const qd_connection_t *c);
bool qd_connection_handle(qd_connection_t *c, pn_event_t *e);
+uint64_t qd_server_allocate_connection_id(qd_server_t *server);
+
const qd_server_config_t *qd_connector_config(const qd_connector_t *c);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 18/32: Dataplane: Added calls in message.h for
streaming putput from adaptors. Renamed qdr_deliver_continue* to
qdr_delivery_continue*
Posted by tr...@apache.org.
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 a307cf433ee35c1f6be53ab9eae0b57225b2125a
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jun 4 15:22:29 2020 -0400
Dataplane: Added calls in message.h for streaming putput from adaptors. Renamed qdr_deliver_continue* to qdr_delivery_continue*
---
include/qpid/dispatch/message.h | 22 ++++++
src/adaptors/reference_adaptor.c | 141 ++++++++++++++++++++++++++++++---------
src/message.c | 16 +++++
src/router_core/connections.c | 4 +-
src/router_core/delivery.c | 20 +++---
src/router_core/delivery.h | 4 +-
src/router_node.c | 2 +-
7 files changed, 164 insertions(+), 45 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 29b2335..0f63b1a 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -278,11 +278,27 @@ void qd_message_compose_2(qd_message_t *msg, qd_composed_field_t *content);
void qd_message_compose_3(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2);
void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2, qd_composed_field_t *content3);
+/**
+ * Send a message with optional headers and an optional raw body with the option of starting
+ * a streaming transfer.
+ *
+ * @param msg The message being composed
+ * @param headers A composed field with 1 or more header sections (incl body performative) or NULL
+ * @param body A buffer list of raw body content or NULL
+ * @param complete True if the message is to be receive-complete.
+ * False if more content will arrive later.
+ */
void qd_message_compose_5(qd_message_t *msg,
qd_composed_field_t *headers,
qd_buffer_list_t *body,
bool complete);
+/**
+ * Extend the content of a streaming message with more buffers.
+ */
+void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers);
+
+
/** Put string representation of a message suitable for logging in buffer.
* @return buffer
*/
@@ -369,6 +385,12 @@ void qd_message_set_send_complete(qd_message_t *msg);
/**
+ * Flag the message as being receive-complete.
+ */
+void qd_message_set_receive_complete(qd_message_t *msg);
+
+
+/**
* Returns true if the delivery tag has already been sent.
*/
bool qd_message_tag_sent(qd_message_t *msg);
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index c7ca07c..f227542 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -25,17 +25,23 @@
#include <stdio.h>
#include <inttypes.h>
-static char *address = "examples";
+static char *address1 = "examples";
+static char *address2 = "stream";
typedef struct qdr_ref_adaptor_t {
qdr_core_t *core;
qdr_protocol_adaptor_t *adaptor;
qd_timer_t *startup_timer;
qd_timer_t *activate_timer;
+ qd_timer_t *stream_timer;
qdr_connection_t *conn;
- qdr_link_t *out_link;
- qdr_link_t *in_link;
+ qdr_link_t *out_link_1;
+ qdr_link_t *out_link_2;
+ qdr_link_t *dynamic_in_link;
char *reply_to;
+ qd_message_t *streaming_message;
+ qdr_delivery_t *streaming_delivery;
+ int stream_count;
} qdr_ref_adaptor_t;
@@ -84,25 +90,39 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
printf("qdr_ref_second_attach: source=%s target=%s\n", fsource, ftarget);
- if (link == adaptor->in_link) {
- uint64_t link_id;
- qdr_terminus_t *target = qdr_terminus(0);
-
- qdr_terminus_set_address(target, address);
-
- adaptor->out_link = qdr_link_first_attach(adaptor->conn,
- QD_INCOMING,
- qdr_terminus(0), //qdr_terminus_t *source,
- target, //qdr_terminus_t *target,
- "ref.1", //const char *name,
- 0, //const char *terminus_addr,
- &link_id);
-
- qdr_link_flow(adaptor->core, adaptor->out_link, 10, false);
-
+ if (link == adaptor->dynamic_in_link) {
+ //
+ // The dynamic in-link has been attached. Get the reply-to address and open
+ // a couple of out-links.
+ //
qd_iterator_t *reply_iter = qdr_terminus_get_address(source);
adaptor->reply_to = (char*) qd_iterator_copy(reply_iter);
printf("qdr_ref_second_attach: reply-to=%s\n", adaptor->reply_to);
+
+ //
+ // Open an out-link for each address
+ //
+ uint64_t link_id;
+ qdr_terminus_t *target = qdr_terminus(0);
+
+ qdr_terminus_set_address(target, address1);
+ adaptor->out_link_1 = qdr_link_first_attach(adaptor->conn,
+ QD_INCOMING,
+ qdr_terminus(0), //qdr_terminus_t *source,
+ target, //qdr_terminus_t *target,
+ "ref.1", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
+
+ target = qdr_terminus(0);
+ qdr_terminus_set_address(target, address2);
+ adaptor->out_link_2 = qdr_link_first_attach(adaptor->conn,
+ QD_INCOMING,
+ qdr_terminus(0), //qdr_terminus_t *source,
+ target, //qdr_terminus_t *target,
+ "ref.2", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
}
}
@@ -120,7 +140,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
printf("qdr_ref_flow: %d credits issued\n", credit);
- if (link == adaptor->out_link) {
+ if (link == adaptor->out_link_1) {
qd_composed_field_t *props = qd_compose(QD_PERFORMATIVE_PROPERTIES, 0);
qd_compose_start_list(props);
qd_compose_insert_null(props); // message-id
@@ -151,8 +171,34 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
qd_message_compose_5(msg, props, &buffers, true);
qd_compose_free(props);
- qdr_delivery_t *dlv = qdr_link_deliver(adaptor->out_link, msg, 0, false, 0, 0);
- qdr_delivery_decref(adaptor->core, dlv, "release protection of return from deliver");
+ qdr_link_deliver(adaptor->out_link_1, msg, 0, false, 0, 0);
+ // Keep return-protection delivery reference as the adaptor's reference
+ } else if (link == adaptor->out_link_2) {
+ //
+ // Begin streaming a long message on the link.
+ //
+ qd_composed_field_t *props = qd_compose(QD_PERFORMATIVE_PROPERTIES, 0);
+ qd_compose_start_list(props);
+ qd_compose_insert_null(props); // message-id
+ qd_compose_insert_null(props); // user-id
+ qd_compose_insert_null(props); // to
+ qd_compose_insert_null(props); // subject
+ qd_compose_insert_string(props, adaptor->reply_to); // reply-to
+ qd_compose_end_list(props);
+
+ props = qd_compose(QD_PERFORMATIVE_BODY_DATA, props);
+
+ adaptor->streaming_message = qd_message();
+
+ qd_message_compose_5(adaptor->streaming_message, props, 0, false);
+ qd_compose_free(props);
+
+ adaptor->streaming_delivery =
+ qdr_link_deliver(adaptor->out_link_2, adaptor->streaming_message, 0, false, 0, 0);
+ adaptor->stream_count = 0;
+ // Keep return-protection delivery reference as the adaptor's reference
+
+ qd_timer_schedule(adaptor->stream_timer, 1000);
}
}
@@ -210,7 +256,8 @@ static int qdr_ref_get_credit(void *context, qdr_link_t *link)
static void qdr_ref_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled)
{
- char *dispname;
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+ char *dispname;
switch (disp) {
case PN_ACCEPTED: dispname = "ACCEPTED"; break;
@@ -221,6 +268,9 @@ static void qdr_ref_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t
dispname = "<UNKNOWN>";
}
printf("qdr_ref_delivery_update: disp=%s settled=%s\n", dispname, settled ? "true" : "false");
+
+ if (settled && qdr_delivery_link(dlv) == adaptor->out_link_1)
+ qdr_delivery_decref(adaptor->core, dlv, "qdr_ref_delivery_update - settled delivery");
}
@@ -272,16 +322,20 @@ static void on_startup(void *context)
0); // bind_token
uint64_t link_id;
+
+ //
+ // Create a dynamic receiver
+ //
qdr_terminus_t *dynamic_source = qdr_terminus(0);
qdr_terminus_set_dynamic(dynamic_source);
- adaptor->in_link = qdr_link_first_attach(adaptor->conn,
- QD_OUTGOING,
- dynamic_source, //qdr_terminus_t *source,
- qdr_terminus(0), //qdr_terminus_t *target,
- "ref.2", //const char *name,
- 0, //const char *terminus_addr,
- &link_id);
+ adaptor->dynamic_in_link = qdr_link_first_attach(adaptor->conn,
+ QD_OUTGOING,
+ dynamic_source, //qdr_terminus_t *source,
+ qdr_terminus(0), //qdr_terminus_t *target,
+ "ref.0", //const char *name,
+ 0, //const char *terminus_addr,
+ &link_id);
}
@@ -293,6 +347,32 @@ static void on_activate(void *context)
}
+static void on_stream(void *context)
+{
+ qdr_ref_adaptor_t *adaptor = (qdr_ref_adaptor_t*) context;
+ const char *content = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
+ const size_t content_length = strlen(content);
+ qd_buffer_t *buf = qd_buffer();
+ qd_buffer_list_t buffers;
+
+ DEQ_INIT(buffers);
+ DEQ_INSERT_TAIL(buffers, buf);
+
+ memcpy(qd_buffer_cursor(buf), content, content_length);
+ qd_buffer_insert(buf, content_length);
+ qd_message_extend(adaptor->streaming_message, &buffers);
+ qdr_delivery_continue(adaptor->core, adaptor->streaming_delivery, false);
+
+ if (adaptor->stream_count < 30) {
+ qd_timer_schedule(adaptor->stream_timer, 1000);
+ adaptor->stream_count++;
+ } else {
+ qd_message_set_receive_complete(adaptor->streaming_message);
+ adaptor->streaming_message = 0;
+ }
+}
+
+
/**
* This initialization function will be invoked when the router core is ready for the protocol
* adaptor to be created. This function must:
@@ -329,6 +409,7 @@ void qdr_ref_adaptor_init(qdr_core_t *core, void **adaptor_context)
qd_timer_schedule(adaptor->startup_timer, 0);
adaptor->activate_timer = qd_timer(core->qd, on_activate, adaptor);
+ adaptor->stream_timer = qd_timer(core->qd, on_stream, adaptor);
}
diff --git a/src/message.c b/src/message.c
index be0c7a8..87c559e 100644
--- a/src/message.c
+++ b/src/message.c
@@ -1232,6 +1232,15 @@ void qd_message_set_send_complete(qd_message_t *in_msg)
}
+void qd_message_set_receive_complete(qd_message_t *in_msg)
+{
+ if (!!in_msg) {
+ qd_message_content_t *content = MSG_CONTENT(in_msg);
+ content->receive_complete = true;
+ }
+}
+
+
bool qd_message_tag_sent(qd_message_t *in_msg)
{
if (!in_msg)
@@ -2209,6 +2218,13 @@ void qd_message_compose_5(qd_message_t *msg,
}
+void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
+{
+ qd_message_content_t *content = MSG_CONTENT(msg);
+ DEQ_APPEND(content->buffers, (*buffers));
+}
+
+
qd_parsed_field_t *qd_message_get_ingress (qd_message_t *msg)
{
return ((qd_message_pvt_t*)msg)->content->ma_pf_ingress;
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index d6c7e40..88c4737 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -824,7 +824,7 @@ static void qdr_link_cleanup_deliveries_CT(qdr_core_t *core, qdr_connection_t *c
if (!qdr_delivery_receive_complete(dlv)) {
qdr_delivery_set_aborted(dlv, true);
- qdr_deliver_continue_peers_CT(core, dlv, false);
+ qdr_delivery_continue_peers_CT(core, dlv, false);
}
if (dlv->multicast) {
@@ -873,7 +873,7 @@ static void qdr_link_cleanup_deliveries_CT(qdr_core_t *core, qdr_connection_t *c
if (!qdr_delivery_receive_complete(dlv)) {
qdr_delivery_set_aborted(dlv, true);
- qdr_deliver_continue_peers_CT(core, dlv, false);
+ qdr_delivery_continue_peers_CT(core, dlv, false);
}
peer = qdr_delivery_first_peer_CT(dlv);
diff --git a/src/router_core/delivery.c b/src/router_core/delivery.c
index 662ac2d..412683d 100644
--- a/src/router_core/delivery.c
+++ b/src/router_core/delivery.c
@@ -25,7 +25,7 @@ ALLOC_DEFINE(qdr_delivery_t);
static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_delete_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
-static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
+static void qdr_delivery_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_delete_delivery_internal_CT(qdr_core_t *core, qdr_delivery_t *delivery);
static bool qdr_delivery_anycast_update_CT(qdr_core_t *core, qdr_delivery_t *dlv,
qdr_delivery_t *peer, uint64_t new_disp, bool settled,
@@ -211,10 +211,10 @@ void qdr_delivery_remote_state_updated(qdr_core_t *core, qdr_delivery_t *deliver
}
-qdr_delivery_t *qdr_deliver_continue(qdr_core_t *core,qdr_delivery_t *in_dlv, bool settled)
+qdr_delivery_t *qdr_delivery_continue(qdr_core_t *core,qdr_delivery_t *in_dlv, bool settled)
{
- qdr_action_t *action = qdr_action(qdr_deliver_continue_CT, "deliver_continue");
+ qdr_action_t *action = qdr_action(qdr_delivery_continue_CT, "delivery_continue");
action->args.delivery.delivery = in_dlv;
qd_message_t *msg = qdr_delivery_message(in_dlv);
@@ -222,7 +222,7 @@ qdr_delivery_t *qdr_deliver_continue(qdr_core_t *core,qdr_delivery_t *in_dlv, bo
action->args.delivery.presettled = settled;
// This incref is for the action reference
- qdr_delivery_incref(in_dlv, "qdr_deliver_continue - add to action list");
+ qdr_delivery_incref(in_dlv, "qdr_delivery_continue - add to action list");
qdr_action_enqueue(core, action);
return in_dlv;
}
@@ -1050,7 +1050,7 @@ static void qdr_delete_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool
}
-void qdr_deliver_continue_peers_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, bool more)
+void qdr_delivery_continue_peers_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, bool more)
{
qdr_delivery_t *peer = qdr_delivery_first_peer_CT(in_dlv);
@@ -1100,7 +1100,7 @@ void qdr_deliver_continue_peers_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, boo
}
-static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
+static void qdr_delivery_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
if (discard)
return;
@@ -1123,7 +1123,7 @@ static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool
// If it is already in the undelivered list, don't try to deliver this again.
//
if (!!link && in_dlv->where != QDR_DELIVERY_IN_UNDELIVERED) {
- qdr_deliver_continue_peers_CT(core, in_dlv, more);
+ qdr_delivery_continue_peers_CT(core, in_dlv, more);
qd_message_t *msg = qdr_delivery_message(in_dlv);
@@ -1147,7 +1147,7 @@ static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool
// We dont want to deal with such deliveries.
//
if (in_dlv->settled && in_dlv->where == QDR_DELIVERY_NOWHERE) {
- qdr_delivery_decref_CT(core, in_dlv, "qdr_deliver_continue_CT - remove from action 1");
+ qdr_delivery_decref_CT(core, in_dlv, "qdr_delivery_continue_CT - remove from action 1");
return;
}
@@ -1179,13 +1179,13 @@ static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool
DEQ_REMOVE(link->settled, in_dlv);
// expect: action holds a ref to in_dlv, so it should not be freed here
assert(sys_atomic_get(&in_dlv->ref_count) > 1);
- qdr_delivery_decref_CT(core, in_dlv, "qdr_deliver_continue_CT - remove from settled list");
+ qdr_delivery_decref_CT(core, in_dlv, "qdr_delivery_continue_CT - remove from settled list");
}
}
}
// This decref is for the action reference
- qdr_delivery_decref_CT(core, in_dlv, "qdr_deliver_continue_CT - remove from action 2");
+ qdr_delivery_decref_CT(core, in_dlv, "qdr_delivery_continue_CT - remove from action 2");
}
diff --git a/src/router_core/delivery.h b/src/router_core/delivery.h
index ce8fd60..7e50efb 100644
--- a/src/router_core/delivery.h
+++ b/src/router_core/delivery.h
@@ -124,7 +124,7 @@ void qdr_delivery_remote_state_updated(qdr_core_t *core, qdr_delivery_t *deliver
bool settled, qdr_error_t *error, pn_data_t *ext_state, bool ref_given);
/* invoked when incoming message data arrives - schedule core thread */
-qdr_delivery_t *qdr_deliver_continue(qdr_core_t *core, qdr_delivery_t *delivery, bool settled);
+qdr_delivery_t *qdr_delivery_continue(qdr_core_t *core, qdr_delivery_t *delivery, bool settled);
//
@@ -153,7 +153,7 @@ qdr_delivery_t *qdr_delivery_first_peer_CT(qdr_delivery_t *dlv);
qdr_delivery_t *qdr_delivery_next_peer_CT(qdr_delivery_t *dlv);
/* schedules all peer deliveries with work for I/O processing */
-void qdr_deliver_continue_peers_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, bool more);
+void qdr_delivery_continue_peers_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, bool more);
/* update the links counters with respect to its delivery */
void qdr_delivery_increment_counters_CT(qdr_core_t *core, qdr_delivery_t *delivery);
diff --git a/src/router_node.c b/src/router_node.c
index 95f9619..bd60542 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -436,7 +436,7 @@ static bool AMQP_rx_handler(void* context, qd_link_t *link)
//
if (delivery) {
- qdr_deliver_continue(router->router_core, delivery, pn_delivery_settled(pnd));
+ qdr_delivery_continue(router->router_core, delivery, pn_delivery_settled(pnd));
return next_delivery;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 09/32: Dataplane: Added setter for dynamic in
qdr_terminus_t
Posted by tr...@apache.org.
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 6ea13c06c6d4921d9454f4f6fff3318b8c1a361f
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 16:19:40 2020 -0400
Dataplane: Added setter for dynamic in qdr_terminus_t
---
include/qpid/dispatch/protocol_adaptor.h | 9 +++++++++
src/router_core/terminus.c | 6 ++++++
2 files changed, 15 insertions(+)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index b96dc41..9920aba 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -537,6 +537,15 @@ bool qdr_terminus_is_anonymous(qdr_terminus_t *term);
bool qdr_terminus_is_coordinator(qdr_terminus_t *term);
/**
+ * qdr_terminus_set_dynamic
+ *
+ * Set this terminus to be dynamic.
+ *
+ * @param term A qdr_terminus pointer returned by qdr_terminus()
+ */
+void qdr_terminus_set_dynamic(qdr_terminus_t *term);
+
+/**
* qdr_terminus_is_dynamic
*
* Indicate whether this terminus represents a dynamic endpoint.
diff --git a/src/router_core/terminus.c b/src/router_core/terminus.c
index b6bb209..9674987 100644
--- a/src/router_core/terminus.c
+++ b/src/router_core/terminus.c
@@ -299,6 +299,12 @@ bool qdr_terminus_is_coordinator(qdr_terminus_t *term)
}
+void qdr_terminus_set_dynamic(qdr_terminus_t *term)
+{
+ term->dynamic = true;
+}
+
+
bool qdr_terminus_is_dynamic(qdr_terminus_t *term)
{
return term->dynamic;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 32/32: DISPATCH-1742 Dataplane: Fixups from rebase
to master.
Posted by tr...@apache.org.
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 bcc2ddff7e20c7c8981d91a8ebe388a6de15eb2f
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Aug 4 13:45:12 2020 -0400
DISPATCH-1742 Dataplane: Fixups from rebase to master.
---
include/qpid/dispatch/protocol_adaptor.h | 3 +--
src/adaptors/reference_adaptor.c | 4 ++--
2 files changed, 3 insertions(+), 4 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index bbf2f27..865f454 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -851,11 +851,10 @@ qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_iterato
qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
qd_iterator_t *ingress, qd_iterator_t *addr,
bool settled, qd_bitmask_t *link_exclusion, int ingress_index,
- int64_t remote_disposition,
+ uint64_t remote_disposition,
pn_data_t *remote_extension_state);
qdr_delivery_t *qdr_link_deliver_to_routed_link(qdr_link_t *link, qd_message_t *msg, bool settled,
const uint8_t *tag, int tag_length,
- uint64_t disposition, pn_data_t* disposition_state,
uint64_t remote_disposition,
pn_data_t *remote_extension_state);
int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit);
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 2a3a975..b6b0c1b 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -183,7 +183,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
qd_message_compose_2(msg, props, true);
qd_compose_free(props);
- qdr_link_deliver(adaptor->out_link_1, msg, 0, false, 0, 0);
+ qdr_link_deliver(adaptor->out_link_1, msg, 0, false, 0, 0, 0, 0);
// Keep return-protection delivery reference as the adaptor's reference
} else if (link == adaptor->out_link_2) {
//
@@ -205,7 +205,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
printf("qdr_ref_flow: Starting a streaming delivery\n");
adaptor->streaming_delivery =
- qdr_link_deliver(adaptor->out_link_2, adaptor->streaming_message, 0, false, 0, 0);
+ qdr_link_deliver(adaptor->out_link_2, adaptor->streaming_message, 0, false, 0, 0, 0, 0);
adaptor->stream_count = 0;
// Keep return-protection delivery reference as the adaptor's reference
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 04/32: Dataplane: Improved sasl-plain test by using
symbolic attribute names, not positional indices.
Posted by tr...@apache.org.
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 4c7b94dca62e7e2a78ad71eda93fa3cf9b611895
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 08:08:54 2020 -0400
Dataplane: Improved sasl-plain test by using symbolic attribute names, not positional indices.
---
tests/system_tests_sasl_plain.py | 37 +++++++++++++++++++------------------
1 file changed, 19 insertions(+), 18 deletions(-)
diff --git a/tests/system_tests_sasl_plain.py b/tests/system_tests_sasl_plain.py
index ff6c5d0..8283250 100644
--- a/tests/system_tests_sasl_plain.py
+++ b/tests/system_tests_sasl_plain.py
@@ -505,19 +505,19 @@ class RouterTestPlainSaslOverSsl(RouterTestPlainSaslCommon):
"""
local_node = Node.connect(self.routers[0].addresses[1], timeout=TIMEOUT)
- results = local_node.query(type='org.apache.qpid.dispatch.connection').results
+ results = local_node.query(type='org.apache.qpid.dispatch.connection').get_entities()
# sslProto should be TLSv1.x
- self.assertTrue(u'TLSv1' in results[0][10])
+ self.assertTrue(u'TLSv1' in results[0].sslProto)
# role should be inter-router
- self.assertEqual(u'inter-router', results[0][3])
+ self.assertEqual(u'inter-router', results[0].role)
# sasl must be plain
- self.assertEqual(u'PLAIN', results[0][6])
+ self.assertEqual(u'PLAIN', results[0].sasl)
# user must be test@domain.com
- self.assertEqual(u'test@domain.com', results[0][8])
+ self.assertEqual(u'test@domain.com', results[0].user)
class RouterTestVerifyHostNameYes(RouterTestPlainSaslCommon):
@@ -604,15 +604,16 @@ class RouterTestVerifyHostNameYes(RouterTestPlainSaslCommon):
due to setting 'verifyHostname': 'yes'
"""
local_node = Node.connect(self.routers[1].addresses[0], timeout=TIMEOUT)
- results = local_node.query(type='org.apache.qpid.dispatch.connection').results
+ results = local_node.query(type='org.apache.qpid.dispatch.connection').get_entities()
+
# There should be only two connections.
# There will be no inter-router connection
self.assertEqual(2, len(results))
- self.assertEqual('in', results[0][4])
- self.assertEqual('normal', results[0][3])
- self.assertEqual('anonymous', results[0][8])
- self.assertEqual('normal', results[1][3])
- self.assertEqual('anonymous', results[1][8])
+ self.assertEqual('in', results[0].dir)
+ self.assertEqual('normal', results[0].role)
+ self.assertEqual('anonymous', results[0].user)
+ self.assertEqual('normal', results[1].role)
+ self.assertEqual('anonymous', results[1].user)
class RouterTestVerifyHostNameNo(RouterTestPlainSaslCommon):
@@ -700,23 +701,23 @@ class RouterTestVerifyHostNameNo(RouterTestPlainSaslCommon):
found = False
for N in range(0, len(results)):
- if results[N][5] == search:
+ if results[N].container == search:
found = True
break
self.assertTrue(found, "Connection to %s not found" % search)
# sslProto should be TLSv1.x
- self.assertTrue(u'TLSv1' in results[N][10])
+ self.assertTrue(u'TLSv1' in results[N].sslProto)
# role should be inter-router
- self.assertEqual(u'inter-router', results[N][3])
+ self.assertEqual(u'inter-router', results[N].role)
# sasl must be plain
- self.assertEqual(u'PLAIN', results[N][6])
+ self.assertEqual(u'PLAIN', results[N].sasl)
# user must be test@domain.com
- self.assertEqual(u'test@domain.com', results[N][8])
+ self.assertEqual(u'test@domain.com', results[N].user)
@SkipIfNeeded(not SASL.extended(), "Cyrus library not available. skipping test")
def test_inter_router_plain_over_ssl_exists(self):
@@ -725,7 +726,7 @@ class RouterTestVerifyHostNameNo(RouterTestPlainSaslCommon):
"""
local_node = Node.connect(self.routers[1].addresses[0], timeout=TIMEOUT)
- results = local_node.query(type='org.apache.qpid.dispatch.connection').results
+ results = local_node.query(type='org.apache.qpid.dispatch.connection').get_entities()
self.common_asserts(results)
@@ -763,7 +764,7 @@ class RouterTestVerifyHostNameNo(RouterTestPlainSaslCommon):
'saslUsername': 'test@domain.com',
'saslPassword': 'password'})
self.routers[1].wait_connectors()
- results = local_node.query(type='org.apache.qpid.dispatch.connection').results
+ results = local_node.query(type='org.apache.qpid.dispatch.connection').get_entities()
self.common_asserts(results)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 27/32: Dataplane: Message parsing bug fixed: now
properly handles empty var-length fields.
Posted by tr...@apache.org.
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 3e2418e6712a38584cd787f0163d5cf120cbb077
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jul 9 15:49:52 2020 -0400
Dataplane: Message parsing bug fixed: now properly handles empty var-length fields.
---
src/message.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/message.c b/src/message.c
index 8228cd8..7f11e5e 100644
--- a/src/message.c
+++ b/src/message.c
@@ -694,7 +694,7 @@ static qd_section_status_t message_section_check(qd_buffer_t **buffer,
// uint8_t size field
pre_consume += 1;
consume |= (uint32_t) next_octet(&test_cursor, &test_buffer);
- if (!test_cursor) return QD_SECTION_NEED_MORE;
+ if (!test_cursor && consume > 0) return QD_SECTION_NEED_MORE;
break;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 19/32: Dataplane: Set proper buffer refcount in
messages during buffer-extend. This ensures that the streaming buffers are
properly freed when no longer needed.
Posted by tr...@apache.org.
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 777f0b7f0709143717291f6819dd20f795dab701
Author: Ted Ross <tr...@apache.org>
AuthorDate: Fri Jun 5 10:16:54 2020 -0400
Dataplane: Set proper buffer refcount in messages during buffer-extend. This ensures that the streaming buffers are properly freed when no longer needed.
---
src/message.c | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/src/message.c b/src/message.c
index 87c559e..8b833f3 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2221,7 +2221,16 @@ void qd_message_compose_5(qd_message_t *msg,
void qd_message_extend(qd_message_t *msg, qd_buffer_list_t *buffers)
{
qd_message_content_t *content = MSG_CONTENT(msg);
+ qd_buffer_t *buf = DEQ_HEAD(*buffers);
+
+ LOCK(content->lock);
+ while (buf) {
+ qd_buffer_set_fanout(buf, content->fanout);
+ buf = DEQ_NEXT(buf);
+ }
+
DEQ_APPEND(content->buffers, (*buffers));
+ UNLOCK(content->lock);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 15/32: Dataplane: Changed the new compose function
to have only one field for headers. This field can have both properties and
application properties. It's more efficient put together like this.
Posted by tr...@apache.org.
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 ac5aa774bb19282831e5e1840270d9c46676fe95
Author: Ted Ross <tr...@apache.org>
AuthorDate: Wed Jun 3 18:14:40 2020 -0400
Dataplane: Changed the new compose function to have only one field for headers. This field can have both properties and application properties. It's more efficient put together like this.
---
include/qpid/dispatch/message.h | 3 +--
src/adaptors/reference_adaptor.c | 2 +-
src/message.c | 14 +++++---------
3 files changed, 7 insertions(+), 12 deletions(-)
diff --git a/include/qpid/dispatch/message.h b/include/qpid/dispatch/message.h
index 55c3dba..030fc0c 100644
--- a/include/qpid/dispatch/message.h
+++ b/include/qpid/dispatch/message.h
@@ -279,8 +279,7 @@ void qd_message_compose_3(qd_message_t *msg, qd_composed_field_t *content1, qd_c
void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *content1, qd_composed_field_t *content2, qd_composed_field_t *content3);
void qd_message_compose_5(qd_message_t *msg,
- qd_composed_field_t *properties,
- qd_composed_field_t *application_properties,
+ qd_composed_field_t *headers,
qd_buffer_list_t *body,
bool complete);
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index b6ea5b4..95801f2 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -146,7 +146,7 @@ static void qdr_ref_flow(void *context, qdr_link_t *link, int credit)
qd_buffer_insert(buf, 13);
DEQ_INSERT_HEAD(buffers, buf);
- qd_message_compose_5(msg, props, 0, &buffers, true);
+ qd_message_compose_5(msg, props, &buffers, true);
qd_compose_free(props);
qdr_delivery_t *dlv = qdr_link_deliver(adaptor->out_link, msg, 0, false, 0, 0);
diff --git a/src/message.c b/src/message.c
index 81ba863..06399cf 100644
--- a/src/message.c
+++ b/src/message.c
@@ -2180,20 +2180,16 @@ void qd_message_compose_4(qd_message_t *msg, qd_composed_field_t *field1, qd_com
void qd_message_compose_5(qd_message_t *msg,
- qd_composed_field_t *properties,
- qd_composed_field_t *application_properties,
+ qd_composed_field_t *headers,
qd_buffer_list_t *body,
bool complete)
{
- qd_message_content_t *content = MSG_CONTENT(msg);
- qd_buffer_list_t *properties_buffers = properties ? qd_compose_buffers(properties) : 0;
- qd_buffer_list_t *application_properties_buffers = application_properties ? qd_compose_buffers(application_properties) : 0;
+ qd_message_content_t *content = MSG_CONTENT(msg);
+ qd_buffer_list_t *headers_buffers = headers ? qd_compose_buffers(headers) : 0;
DEQ_INIT(content->buffers);
- if (properties_buffers)
- DEQ_APPEND(content->buffers, (*properties_buffers));
- if (application_properties_buffers)
- DEQ_APPEND(content->buffers, (*application_properties_buffers));
+ if (headers_buffers)
+ DEQ_APPEND(content->buffers, (*headers_buffers));
if (body) {
qd_composed_field_t *field = qd_compose(QD_PERFORMATIVE_BODY_DATA, 0);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 22/32: Dataplane: Added no_route and
initial_delivery on link-first-attach.
Posted by tr...@apache.org.
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 b3af7de4ce33660aff4db1c9b7dc419d8b733738
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jun 18 14:25:44 2020 -0400
Dataplane: Added no_route and initial_delivery on link-first-attach.
---
include/qpid/dispatch/protocol_adaptor.h | 4 ++
src/adaptors/reference_adaptor.c | 8 ++++
src/router_core/connections.c | 64 ++++++++++++++++++++++++++++++--
src/router_core/router_core.c | 14 +++++++
src/router_core/router_core_private.h | 2 +
src/router_core/transfer.c | 51 ++++++++++++++-----------
src/router_node.c | 4 ++
7 files changed, 122 insertions(+), 25 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index 9920aba..bbf2f27 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -775,6 +775,8 @@ const char *qdr_link_name(const qdr_link_t *link);
* @param target Target terminus of the attach
* @param name - name of the link
* @param terminus_addr - terminus address if any
+ * @param no_route If true, new deliveries are not to be routed to this link
+ * @param initial_delivery (optional) Move this delivery from its existing link to the head of this link's buffer
* @param link_id - set to the management id of the new link
* @return A pointer to a new qdr_link_t object to track the link
*/
@@ -784,6 +786,8 @@ qdr_link_t *qdr_link_first_attach(qdr_connection_t *conn,
qdr_terminus_t *target,
const char *name,
const char *terminus_addr,
+ bool no_route,
+ qdr_delivery_t *initial_delivery,
uint64_t *link_id);
/**
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 7a65c17..59975a9 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -113,6 +113,8 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
target, //qdr_terminus_t *target,
"ref.1", //const char *name,
0, //const char *terminus_addr,
+ false, //bool no_route
+ 0, //qdr_delivery_t *initial_delivery
&link_id);
target = qdr_terminus(0);
@@ -123,6 +125,8 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
target, //qdr_terminus_t *target,
"ref.2", //const char *name,
0, //const char *terminus_addr,
+ false, //bool no_route
+ 0, //qdr_delivery_t *initial_delivery
&link_id);
source = qdr_terminus(0);
@@ -133,6 +137,8 @@ static void qdr_ref_second_attach(void *context, qdr_link_t *link,
qdr_terminus(0), //qdr_terminus_t *target,
"ref.3", //const char *name,
0, //const char *terminus_addr,
+ false, //bool no_route
+ 0, //qdr_delivery_t *initial_delivery
&link_id);
}
}
@@ -366,6 +372,8 @@ static void on_startup(void *context)
qdr_terminus(0), //qdr_terminus_t *target,
"ref.0", //const char *name,
0, //const char *terminus_addr,
+ false, //bool no_route
+ 0, //qdr_delivery_t *initial_delivery
&link_id);
}
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 88c4737..d76f01c 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -543,6 +543,8 @@ qdr_link_t *qdr_link_first_attach(qdr_connection_t *conn,
qdr_terminus_t *target,
const char *name,
const char *terminus_addr,
+ bool no_route,
+ qdr_delivery_t *initial_delivery,
uint64_t *link_id)
{
qdr_action_t *action = qdr_action(qdr_link_inbound_first_attach_CT, "link_first_attach");
@@ -573,6 +575,7 @@ qdr_link_t *qdr_link_first_attach(qdr_connection_t *conn,
link->core_ticks = conn->core->uptime_ticks;
link->zero_credit_time = conn->core->uptime_ticks;
link->terminus_survives_disconnect = qdr_terminus_survives_disconnect(local_terminus);
+ link->no_route = no_route;
link->strip_annotations_in = conn->strip_annotations_in;
link->strip_annotations_out = conn->strip_annotations_out;
@@ -595,6 +598,7 @@ qdr_link_t *qdr_link_first_attach(qdr_connection_t *conn,
action->args.connection.dir = dir;
action->args.connection.source = source;
action->args.connection.target = target;
+ action->args.connection.initial_delivery = initial_delivery;
qdr_action_enqueue(conn->core, action);
return link;
@@ -1563,6 +1567,56 @@ static void qdr_attach_link_downlink_CT(qdr_core_t *core, qdr_connection_t *conn
}
+static void qdr_link_process_initial_delivery_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *dlv)
+{
+ qdr_link_t *old_link = safe_deref_qdr_link_t(dlv->link_sp);
+ int ref_delta = 0;
+
+ //
+ // Remove the delivery from its current link if needed
+ //
+ if (!!old_link) {
+ switch (dlv->where) {
+ case QDR_DELIVERY_NOWHERE:
+ break;
+
+ case QDR_DELIVERY_IN_UNDELIVERED:
+ DEQ_REMOVE(old_link->undelivered, dlv);
+ dlv->where = QDR_DELIVERY_NOWHERE;
+ ref_delta--;
+ break;
+
+ case QDR_DELIVERY_IN_UNSETTLED:
+ DEQ_REMOVE(old_link->unsettled, dlv);
+ dlv->where = QDR_DELIVERY_NOWHERE;
+ ref_delta--;
+ break;
+
+ case QDR_DELIVERY_IN_SETTLED:
+ DEQ_REMOVE(old_link->settled, dlv);
+ dlv->where = QDR_DELIVERY_NOWHERE;
+ ref_delta--;
+ break;
+ }
+ }
+
+ //
+ // Enqueue the delivery onto the new link's undelivered list
+ //
+ set_safe_ptr_qdr_link_t(link, &dlv->link_sp);
+ qdr_forward_deliver_CT(core, link, dlv);
+
+ //
+ // Adjust the delivery's reference count
+ //
+ assert(ref_delta <= 0);
+ while (ref_delta < 0) {
+ qdr_delivery_decref(core, dlv, "qdr_link_process_initial_delivery_CT");
+ ref_delta++;
+ }
+}
+
+
static void qdr_link_inbound_first_attach_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
qdr_connection_t *conn = safe_deref_qdr_connection_t(action->args.connection.conn);
@@ -1570,9 +1624,10 @@ static void qdr_link_inbound_first_attach_CT(qdr_core_t *core, qdr_action_t *act
if (discard || !conn || !link)
return;
- qd_direction_t dir = action->args.connection.dir;
- qdr_terminus_t *source = action->args.connection.source;
- qdr_terminus_t *target = action->args.connection.target;
+ qd_direction_t dir = action->args.connection.dir;
+ qdr_terminus_t *source = action->args.connection.source;
+ qdr_terminus_t *target = action->args.connection.target;
+ qdr_delivery_t *initial_dlv = action->args.connection.initial_delivery;
//
// Start the attach count.
@@ -1675,6 +1730,9 @@ static void qdr_link_inbound_first_attach_CT(qdr_core_t *core, qdr_action_t *act
//
// Handle outgoing link cases
//
+ if (initial_dlv)
+ qdr_link_process_initial_delivery_CT(core, link, initial_dlv);
+
switch (link->link_type) {
case QD_LINK_ENDPOINT: {
if (core->addr_lookup_handler)
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index 25c124a..f61ebfd 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -567,6 +567,13 @@ void qdr_core_bind_address_link_CT(qdr_core_t *core, qdr_address_t *addr, qdr_li
if (key && (*key == QD_ITER_HASH_PREFIX_MOBILE))
link->phase = (int) (key[1] - '0');
+ //
+ // If this link is configured as no-route, don't create any functional linkage between the
+ // link and the address beyond the owning_addr.
+ //
+ if (link->no_route)
+ return;
+
if (link->link_direction == QD_OUTGOING) {
qdr_add_link_ref(&addr->rlinks, link, QDR_LINK_LIST_CLASS_ADDRESS);
if (DEQ_SIZE(addr->rlinks) == 1) {
@@ -599,6 +606,13 @@ void qdr_core_unbind_address_link_CT(qdr_core_t *core, qdr_address_t *addr, qdr_
{
link->owning_addr = 0;
+ //
+ // If the link is configured as no_route, there will be no further link/address
+ // linkage to disconnect.
+ //
+ if (link->no_route)
+ return;
+
if (link->link_direction == QD_OUTGOING) {
qdr_del_link_ref(&addr->rlinks, link, QDR_LINK_LIST_CLASS_ADDRESS);
if (DEQ_SIZE(addr->rlinks) == 0) {
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index ac123e5..1709442 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -136,6 +136,7 @@ struct qdr_action_t {
qd_detach_type_t dt;
int credit;
bool drain;
+ qdr_delivery_t *initial_delivery;
} connection;
//
@@ -464,6 +465,7 @@ struct qdr_link_t {
bool streaming; ///< True if this link can be reused for streaming msgs
bool in_streaming_pool; ///< True if this link is in the connections standby pool STREAMING_POOL
bool terminus_survives_disconnect;
+ bool no_route; ///< True if this link is to not receive routed deliveries
char *strip_prefix;
char *insert_prefix;
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index add176a..6ca2b2b 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -151,7 +151,8 @@ int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
dlv = DEQ_HEAD(link->undelivered);
if (dlv) {
qdr_delivery_incref(dlv, "qdr_link_process_deliveries - holding the undelivered delivery locally");
- uint64_t new_disp = 0;
+ uint64_t new_disp = 0;
+ bool to_new_link = false; ///< Delivery got moved to a new link by the handler
// DISPATCH-1302 race hack fix: There is a race between the CORE thread
// and the outbound (this) thread over settlement. It occurs when the CORE
@@ -165,9 +166,13 @@ int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
sys_mutex_unlock(conn->work_lock);
new_disp = conn->protocol_adaptor->deliver_handler(conn->protocol_adaptor->user_context, link, dlv, settled);
sys_mutex_lock(conn->work_lock);
- } while (settled != dlv->settled); // oops missed the settlement
+ if (safe_deref_qdr_link_t(dlv->link_sp) != link) {
+ to_new_link = true;
+ break;
+ }
+ } while (settled != dlv->settled && !to_new_link); // oops missed the settlement
send_complete = qdr_delivery_send_complete(dlv);
- if (send_complete) {
+ if (send_complete || to_new_link) {
//
// The entire message has been sent. It is now the appropriate time to have the delivery removed
// from the head of the undelivered list and move it to the unsettled list if it is not settled.
@@ -178,26 +183,28 @@ int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
link->credit_to_core--;
link->total_deliveries++;
- // DISPATCH-1153:
- // If the undelivered list is cleared the link may have detached. Stop processing.
- offer = DEQ_SIZE(link->undelivered);
- if (offer == 0) {
- qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - release local reference - closed link");
- sys_mutex_unlock(conn->work_lock);
- return num_deliveries_completed;
- }
+ if (!to_new_link) {
+ // DISPATCH-1153:
+ // If the undelivered list is cleared the link may have detached. Stop processing.
+ offer = DEQ_SIZE(link->undelivered);
+ if (offer == 0) {
+ qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - release local reference - closed link");
+ sys_mutex_unlock(conn->work_lock);
+ return num_deliveries_completed;
+ }
- assert(dlv == DEQ_HEAD(link->undelivered));
- DEQ_REMOVE_HEAD(link->undelivered);
- dlv->link_work = 0;
-
- if (settled || qdr_delivery_oversize(dlv) || qdr_delivery_is_aborted(dlv)) {
- dlv->where = QDR_DELIVERY_NOWHERE;
- qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - remove from undelivered list");
- } else {
- DEQ_INSERT_TAIL(link->unsettled, dlv);
- dlv->where = QDR_DELIVERY_IN_UNSETTLED;
- qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer: dlv:%lx qdr_link_process_deliveries: undelivered-list -> unsettled-list", (long) dlv);
+ assert(dlv == DEQ_HEAD(link->undelivered));
+ DEQ_REMOVE_HEAD(link->undelivered);
+ dlv->link_work = 0;
+
+ if (settled || qdr_delivery_oversize(dlv) || qdr_delivery_is_aborted(dlv)) {
+ dlv->where = QDR_DELIVERY_NOWHERE;
+ qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - remove from undelivered list");
+ } else {
+ DEQ_INSERT_TAIL(link->unsettled, dlv);
+ dlv->where = QDR_DELIVERY_IN_UNSETTLED;
+ qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer: dlv:%lx qdr_link_process_deliveries: undelivered-list -> unsettled-list", (long) dlv);
+ }
}
}
else {
diff --git a/src/router_node.c b/src/router_node.c
index bd60542..993a235 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -827,6 +827,8 @@ static int AMQP_incoming_link_handler(void* context, qd_link_t *link)
qdr_terminus(qd_link_remote_target(link)),
pn_link_name(qd_link_pn(link)),
terminus_addr,
+ false,
+ 0,
&link_id);
qd_link_set_link_id(link, link_id);
qdr_link_set_context(qdr_link, link);
@@ -856,6 +858,8 @@ static int AMQP_outgoing_link_handler(void* context, qd_link_t *link)
qdr_terminus(qd_link_remote_target(link)),
pn_link_name(qd_link_pn(link)),
terminus_addr,
+ false,
+ 0,
&link_id);
qd_link_set_link_id(link, link_id);
qdr_link_set_context(qdr_link, link);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 05/32: Dataplane: Added adaptor plugin capability.
Started first reference adaptor (TCP).
Posted by tr...@apache.org.
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
[qpid-dispatch] 29/32: Dataplane: disabled reference adaptor
Posted by tr...@apache.org.
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 2898359dd882d0fcaeec3c8dbb6361c4dfc91e32
Author: Ted Ross <tr...@apache.org>
AuthorDate: Thu Jul 16 08:52:24 2020 -0400
Dataplane: disabled reference adaptor
---
src/adaptors/reference_adaptor.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/adaptors/reference_adaptor.c b/src/adaptors/reference_adaptor.c
index 5dffeca..2e1d98e 100644
--- a/src/adaptors/reference_adaptor.c
+++ b/src/adaptors/reference_adaptor.c
@@ -493,4 +493,4 @@ void qdr_ref_adaptor_final(void *adaptor_context)
/**
* Declare the adaptor so that it will self-register on process startup.
*/
-QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
+//QDR_CORE_ADAPTOR_DECLARE("ref-adaptor", qdr_ref_adaptor_init, qdr_ref_adaptor_final)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 06/32: Dataplane: Added documentation for the
protocol adaptor callbacks.
Posted by tr...@apache.org.
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 b75f05fd37e0035f549aa5e3c96097e91fc68684
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 11:42:27 2020 -0400
Dataplane: Added documentation for the protocol adaptor callbacks.
---
include/qpid/dispatch/protocol_adaptor.h | 142 +++++++++++++++++++++++++++++--
src/router_node.c | 5 +-
2 files changed, 138 insertions(+), 9 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index 570b867..37fb850 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -54,7 +54,7 @@ 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.
+ * A protocol adaptor may 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
@@ -67,6 +67,7 @@ void qdr_register_adaptor(const char *name,
qdr_adaptor_init_t on_init,
qdr_adaptor_final_t on_final);
+
/**
******************************************************************************
* Callback function definitions
@@ -90,69 +91,177 @@ typedef void (*qdr_connection_activate_t) (void *context, qdr_connection_t *conn
/**
* qdr_link_first_attach_t callback
+ *
+ * This function is invoked when the core requires that a new link be attached over a
+ * connection. Such a link is either being initiated from the core or is the propagation
+ * of a link route from an originator somewhere in the network.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param conn The connection over which the first attach is to be sent
+ * @param link The link object for the new link
+ * @param source The source terminus for the attach
+ * @param target The target terminus for the attach
+ * @param ssn_class The session class to be used to allocate this link to a session
*/
-typedef void (*qdr_link_first_attach_t) (void *context, qdr_connection_t *conn, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target,
- qd_session_class_t);
+typedef void (*qdr_link_first_attach_t) (void *context,
+ qdr_connection_t *conn,
+ qdr_link_t *link,
+ qdr_terminus_t *source,
+ qdr_terminus_t *target,
+ qd_session_class_t ssn_class);
/**
* qdr_link_second_attach_t callback
+ *
+ * This function is invoked when the core is responding to an incoming attach from an
+ * external container. The function must send the responding (second) attach to the
+ * remote container to complete the attachment of the link.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being attached
+ * @param source The source terminus for the attach
+ * @param target The target terminus for the attach
*/
-typedef void (*qdr_link_second_attach_t) (void *context, qdr_link_t *link,
- qdr_terminus_t *source, qdr_terminus_t *target);
+typedef void (*qdr_link_second_attach_t) (void *context,
+ qdr_link_t *link,
+ qdr_terminus_t *source,
+ qdr_terminus_t *target);
/**
* qdr_link_detach_t callback
+ *
+ * A DETACH performative must be sent for a link that is being closed or detached.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being detached
+ * @param error Error record if the detach is the result of an error condition, null otherwise
+ * @param first True if this is the first detach (i.e. initiated outbound), False if this is the
+ * the response to a remotely initiated detach
+ * @param close True if this is a link close, False if this is a link detach
*/
typedef void (*qdr_link_detach_t) (void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close);
/**
* qdr_link_flow_t callback
+ *
+ * Credit is being issued for an incoming link. Credit is issued incrementally, being added
+ * to credit may have been issued in the past.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link for which credit is being issued
+ * @param credit The number of new credits being issued to the link
*/
typedef void (*qdr_link_flow_t) (void *context, qdr_link_t *link, int credit);
/**
* qdr_link_offer_t callback
+ *
+ * This function is invoked when the core wishes to inform the remote terminus of an outoing link
+ * that it is willing and ready to transfer a certain number of deliveries over that link.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being affected
+ * @param delivery_count The number of deliveries available to be sent over this link
*/
typedef void (*qdr_link_offer_t) (void *context, qdr_link_t *link, int delivery_count);
/**
* qdr_link_drained_t callback
+ *
+ * This function is invoked when the core wishes to inform the remote terminus of an outgoing link
+ * that it has drained its outgoing deliveries and removed any residual credit.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being affected
*/
typedef void (*qdr_link_drained_t) (void *context, qdr_link_t *link);
/**
* qdr_link_drain_t callback
+ *
+ * This functino is invoked when the core wishes to inform the remote terminus of a link
+ * that the drain mode of the link has changed.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being affected
+ * @param mode True for enabling drain mode, False for disabling drain mode
*/
typedef void (*qdr_link_drain_t) (void *context, qdr_link_t *link, bool mode);
/**
* qdr_link_push_t callback
+ *
+ * The core invokes this function when it wishes to transfer deliveries on an outgoing link.
+ * This function, in turn, calls qdr_link_process_deliveries with the desired number of
+ * deliveries (up to limit) that should be transferred from the core. Typically, this
+ * function will call qdr_link_process_deliveries with MIN(limit, available-credit).
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link over which deliveries should be transfered
+ * @param limit The maximum number of deliveries that should be transferred
+ * @return The number of deliveries transferred
*/
typedef int (*qdr_link_push_t) (void *context, qdr_link_t *link, int limit);
/**
* qdr_link_deliver_t callback
+ *
+ * This function is invoked by the core during the execution of qdr_link_process_deliveries. There
+ * is one invocation for each delivery to be transferred. If this function returns a non-zero
+ * disposition, the core will settle the delivery with that disposition back to the sender.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link over which deliveries should be transfered
+ * @param delivery The delivery (which contains the message) to be transferred
+ * @param settled True iff the delivery is already settled
+ * @return The disposition of the delivery to be sent back to the sender, or 0 if no disposition
*/
typedef uint64_t (*qdr_link_deliver_t) (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
/**
* qdr_link_get_credit_t callback
+ *
+ * Query a link for the current amount of available credit.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param link The link being queried
+ * @return The number of credits available on this link
*/
typedef int (*qdr_link_get_credit_t) (void *context, qdr_link_t *link);
/**
* qdr_delivery_update_t callback
+ *
+ * This function is invoked by the core when a delivery's disposition and settlement are being
+ * changed. This fuction must send the updated delivery state to the remote terminus.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param dlv The delivery being updated
+ * @param disp The new disposition for the delivery
+ * @param settled True iff the delivery is being settled
*/
typedef void (*qdr_delivery_update_t) (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
/**
* qdr_connection_close_t callback
+ *
+ * The core invokes this function when a connection to a remote container is to be closed.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param conn The connection being closed
+ * @param error If the close is a result of an error, this is the error record to be used, else it's null
*/
typedef void (*qdr_connection_close_t) (void *context, qdr_connection_t *conn, qdr_error_t *error);
/**
* qdr_connection_trace_t callback
+ *
+ * This callback is invoked when per-connection tracing is being turned on of off. The new tracing
+ * state must be propagated down into the tracing capabilities of the lower layers of connection processing.
+ *
+ * @param context The context supplied when the callback was registered
+ * @param conn The connection being affected
+ * @param trace True to enable tracing for this connection, False to disable tracing for this connection
*/
typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, bool trace);
@@ -163,6 +272,17 @@ typedef void (*qdr_connection_trace_t) (void *context, qdr_connection_t *conn, b
******************************************************************************
*/
+/**
+ * qdr_protocol_adaptor
+ *
+ * Register a new protocol adaptor with the router core.
+ *
+ * @param core Pointer to the core object
+ * @param name The name of this adaptor's protocol
+ * @param context The context to be used in all of the callbacks
+ * @param callbacks Pointers to all of the callback functions used in the adaptor
+ * @return Pointer to a protocol adaptor object
+ */
qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
const char *name,
void *context,
@@ -181,6 +301,16 @@ qdr_protocol_adaptor_t *qdr_protocol_adaptor(qdr_core_t *core,
qdr_connection_close_t conn_close,
qdr_connection_trace_t conn_trace);
+
+/**
+ * qdr_protocol_adaptor_free
+ *
+ * Free the resources used for a protocol adaptor. This should be called during adaptor
+ * finalization.
+ *
+ * @param core Pointer to the core object
+ * @param adaptor Pointer to a protocol adaptor object returned by qdr_protocol_adaptor
+ */
void qdr_protocol_adaptor_free(qdr_core_t *core, qdr_protocol_adaptor_t *adaptor);
diff --git a/src/router_node.c b/src/router_node.c
index a5278dc..90a8f1b 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1576,12 +1576,12 @@ static void CORE_link_second_attach(void *context, qdr_link_t *link, qdr_terminu
static void CORE_conn_trace(void *context, qdr_connection_t *qdr_conn, bool trace)
{
- qd_connection_t *qconn = (qd_connection_t*) qdr_connection_get_context(qdr_conn);
+ qd_connection_t *qconn = (qd_connection_t*) qdr_connection_get_context(qdr_conn);
if (!qconn)
return;
- pn_transport_t *tport = pn_connection_transport(qconn->pn_conn);
+ pn_transport_t *tport = pn_connection_transport(qconn->pn_conn);
if (!tport)
return;
@@ -1589,7 +1589,6 @@ static void CORE_conn_trace(void *context, qdr_connection_t *qdr_conn, bool trac
if (trace) {
pn_transport_trace(tport, PN_TRACE_FRM);
pn_transport_set_tracer(tport, connection_transport_tracer);
-
}
else {
pn_transport_trace(tport, PN_TRACE_OFF);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 07/32: Dataplane: Removed old handler call which is
not used anymore.
Posted by tr...@apache.org.
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 72753508bd320cbb2fde649fc748bfea5d65b58c
Author: Ted Ross <tr...@apache.org>
AuthorDate: Tue Jun 2 13:24:29 2020 -0400
Dataplane: Removed old handler call which is not used anymore.
---
include/qpid/dispatch/protocol_adaptor.h | 19 -------------------
1 file changed, 19 deletions(-)
diff --git a/include/qpid/dispatch/protocol_adaptor.h b/include/qpid/dispatch/protocol_adaptor.h
index 37fb850..cea625b 100644
--- a/include/qpid/dispatch/protocol_adaptor.h
+++ b/include/qpid/dispatch/protocol_adaptor.h
@@ -871,25 +871,6 @@ 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 *context,
- qdr_connection_activate_t activate,
- qdr_link_first_attach_t first_attach,
- qdr_link_second_attach_t second_attach,
- qdr_link_detach_t detach,
- qdr_link_flow_t flow,
- qdr_link_offer_t offer,
- qdr_link_drained_t drained,
- qdr_link_drain_t drain,
- qdr_link_push_t push,
- qdr_link_deliver_t deliver,
- qdr_link_get_credit_t get_credit,
- qdr_delivery_update_t delivery_update,
- qdr_connection_close_t conn_close,
- qdr_connection_trace_t conn_trace);
-
-
-
qdr_connection_info_t *qdr_connection_info(bool is_encrypted,
bool is_authenticated,
bool opened,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-dispatch] 03/32: Dataplane: Exposed the protocol name with
the connection entity.
Posted by tr...@apache.org.
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 f4a41568eb8eaa5a4498b779967e89db81fc4a9c
Author: Ted Ross <tr...@apache.org>
AuthorDate: Mon Jun 1 18:31:32 2020 -0400
Dataplane: Exposed the protocol name with the connection entity.
---
python/qpid_dispatch/management/qdrouter.json | 4 +++
src/router_core/agent_connection.c | 46 +++++++++++++++------------
src/router_core/agent_connection.h | 2 +-
tools/qdstat.in | 4 ++-
4 files changed, 34 insertions(+), 22 deletions(-)
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index 616cf64..4dee14a 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -1742,6 +1742,10 @@
"operStatus": {
"type": ["up", "closing"]
},
+ "protocol": {
+ "type": "string",
+ "description": "The protocol adaptor that is handling this connection"
+ },
"container": {
"description": "The container for this connection",
"type": "string"
diff --git a/src/router_core/agent_connection.c b/src/router_core/agent_connection.c
index 93faec6..f1fef54 100644
--- a/src/router_core/agent_connection.c
+++ b/src/router_core/agent_connection.c
@@ -26,26 +26,27 @@
#define QDR_CONNECTION_IDENTITY 1
#define QDR_CONNECTION_HOST 2
#define QDR_CONNECTION_ROLE 3
-#define QDR_CONNECTION_DIR 4
-#define QDR_CONNECTION_CONTAINER_ID 5
-#define QDR_CONNECTION_SASL_MECHANISMS 6
-#define QDR_CONNECTION_IS_AUTHENTICATED 7
-#define QDR_CONNECTION_USER 8
-#define QDR_CONNECTION_IS_ENCRYPTED 9
-#define QDR_CONNECTION_SSLPROTO 10
-#define QDR_CONNECTION_SSLCIPHER 11
-#define QDR_CONNECTION_PROPERTIES 12
-#define QDR_CONNECTION_SSLSSF 13
-#define QDR_CONNECTION_TENANT 14
-#define QDR_CONNECTION_TYPE 15
-#define QDR_CONNECTION_SSL 16
-#define QDR_CONNECTION_OPENED 17
-#define QDR_CONNECTION_ACTIVE 18
-#define QDR_CONNECTION_ADMIN_STATUS 19
-#define QDR_CONNECTION_OPER_STATUS 20
-#define QDR_CONNECTION_UPTIME_SECONDS 21
-#define QDR_CONNECTION_LAST_DLV_SECONDS 22
-#define QDR_CONNECTION_ENABLE_PROTOCOL_TRACE 23
+#define QDR_CONNECTION_PROTOCOL 4
+#define QDR_CONNECTION_DIR 5
+#define QDR_CONNECTION_CONTAINER_ID 6
+#define QDR_CONNECTION_SASL_MECHANISMS 7
+#define QDR_CONNECTION_IS_AUTHENTICATED 8
+#define QDR_CONNECTION_USER 9
+#define QDR_CONNECTION_IS_ENCRYPTED 10
+#define QDR_CONNECTION_SSLPROTO 11
+#define QDR_CONNECTION_SSLCIPHER 12
+#define QDR_CONNECTION_PROPERTIES 13
+#define QDR_CONNECTION_SSLSSF 14
+#define QDR_CONNECTION_TENANT 15
+#define QDR_CONNECTION_TYPE 16
+#define QDR_CONNECTION_SSL 17
+#define QDR_CONNECTION_OPENED 18
+#define QDR_CONNECTION_ACTIVE 19
+#define QDR_CONNECTION_ADMIN_STATUS 20
+#define QDR_CONNECTION_OPER_STATUS 21
+#define QDR_CONNECTION_UPTIME_SECONDS 22
+#define QDR_CONNECTION_LAST_DLV_SECONDS 23
+#define QDR_CONNECTION_ENABLE_PROTOCOL_TRACE 24
const char * const QDR_CONNECTION_DIR_IN = "in";
@@ -70,6 +71,7 @@ const char *qdr_connection_columns[] =
"identity",
"host",
"role",
+ "protocol",
"dir",
"container",
"sasl",
@@ -147,6 +149,10 @@ static void qdr_connection_insert_column_CT(qdr_core_t *core, qdr_connection_t *
qd_compose_insert_string(body, qdr_connection_roles[conn->connection_info->role]);
break;
+ case QDR_CONNECTION_PROTOCOL:
+ qd_compose_insert_string(body, conn->protocol_adaptor->name);
+ break;
+
case QDR_CONNECTION_DIR:
if (conn->connection_info->dir == QD_INCOMING)
qd_compose_insert_string(body, QDR_CONNECTION_DIR_IN);
diff --git a/src/router_core/agent_connection.h b/src/router_core/agent_connection.h
index f9ce002..dc57dc5 100644
--- a/src/router_core/agent_connection.h
+++ b/src/router_core/agent_connection.h
@@ -35,7 +35,7 @@ void qdra_connection_update_CT(qdr_core_t *core,
qdr_query_t *query,
qd_parsed_field_t *in_body);
-#define QDR_CONNECTION_COLUMN_COUNT 24
+#define QDR_CONNECTION_COLUMN_COUNT 25
extern const char *qdr_connection_columns[QDR_CONNECTION_COLUMN_COUNT + 1];
#endif
diff --git a/tools/qdstat.in b/tools/qdstat.in
index 5b423c4..13be1d0 100755
--- a/tools/qdstat.in
+++ b/tools/qdstat.in
@@ -192,6 +192,7 @@ class BusManager(Node):
heads.append(Header("host"))
heads.append(Header("container"))
heads.append(Header("role"))
+ heads.append(Header("proto"))
heads.append(Header("dir"))
heads.append(Header("security"))
heads.append(Header("authentication"))
@@ -230,7 +231,8 @@ class BusManager(Node):
row.append(conn.identity)
row.append(conn.host)
row.append(conn.container)
- row.append(conn.role)
+ row.append(get(conn, 'role'))
+ row.append(conn.protocol)
row.append(conn.dir)
row.append(self.connSecurity(conn))
row.append(self.connAuth(conn))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org