[qpid-dispatch] 16/32: Dataplane: Added message method to set send-complete. Added reference code to receive messages (non streamed).

[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