You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by jd...@apache.org on 2021/04/22 09:33:13 UTC
[qpid-proton] branch main updated: PROTON-2370: [cpp] An accessor
for the delivery tag (#309)
This is an automated email from the ASF dual-hosted git repository.
jdanek pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
The following commit(s) were added to refs/heads/main by this push:
new 08b301a PROTON-2370: [cpp] An accessor for the delivery tag (#309)
08b301a is described below
commit 08b301a97c834e002d41ee852bba1288fe83b936
Author: Rakhi Kumari <ra...@gmail.com>
AuthorDate: Thu Apr 22 15:03:07 2021 +0530
PROTON-2370: [cpp] An accessor for the delivery tag (#309)
---
cpp/include/proton/delivery.hpp | 4 ++
cpp/include/proton/sender.hpp | 1 +
cpp/src/delivery.cpp | 5 ++
cpp/src/delivery_test.cpp | 134 ++++++++++++++++++++++++++++++++++++++++
cpp/src/sender.cpp | 11 +++-
cpp/tests.cmake | 1 +
6 files changed, 154 insertions(+), 2 deletions(-)
diff --git a/cpp/include/proton/delivery.hpp b/cpp/include/proton/delivery.hpp
index c1823d9..4c4979b 100644
--- a/cpp/include/proton/delivery.hpp
+++ b/cpp/include/proton/delivery.hpp
@@ -22,6 +22,7 @@
*
*/
+#include "./binary.hpp"
#include "./fwd.hpp"
#include "./internal/export.hpp"
#include "./internal/object.hpp"
@@ -64,6 +65,9 @@ class delivery : public transfer {
/// Settle with MODIFIED state.
PN_CPP_EXTERN void modify();
+ /// Return the tag for this delivery.
+ PN_CPP_EXTERN binary tag() const;
+
/// @cond INTERNAL
friend class internal::factory<delivery>;
/// @endcond
diff --git a/cpp/include/proton/sender.hpp b/cpp/include/proton/sender.hpp
index caf2f51..0f2f29c 100644
--- a/cpp/include/proton/sender.hpp
+++ b/cpp/include/proton/sender.hpp
@@ -56,6 +56,7 @@ PN_CPP_CLASS_EXTERN sender : public link {
/// Send a message on the sender.
PN_CPP_EXTERN tracker send(const message &m);
+ PN_CPP_EXTERN tracker send(const message &m, const binary &tag);
/// Get the source node.
PN_CPP_EXTERN class source source() const;
diff --git a/cpp/src/delivery.cpp b/cpp/src/delivery.cpp
index 7baa475..a9d472a 100644
--- a/cpp/src/delivery.cpp
+++ b/cpp/src/delivery.cpp
@@ -27,6 +27,10 @@
#include <proton/delivery.h>
+#include "types_internal.hpp"
+
+#include "proton/binary.hpp"
+
namespace {
void settle_delivery(pn_delivery_t* o, uint64_t state) {
@@ -40,6 +44,7 @@ namespace proton {
delivery::delivery(pn_delivery_t* d): transfer(make_wrapper(d)) {}
receiver delivery::receiver() const { return make_wrapper<class receiver>(pn_delivery_link(pn_object())); }
+binary delivery::tag() const { return bin(pn_delivery_tag(pn_object())); }
delivery::~delivery() {}
void delivery::accept() { settle_delivery(pn_object(), ACCEPTED); }
void delivery::reject() { settle_delivery(pn_object(), REJECTED); }
diff --git a/cpp/src/delivery_test.cpp b/cpp/src/delivery_test.cpp
new file mode 100644
index 0000000..93dfa40
--- /dev/null
+++ b/cpp/src/delivery_test.cpp
@@ -0,0 +1,134 @@
+/*
+ *
+ * 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 <proton/connection.hpp>
+#include <proton/connection_options.hpp>
+#include <proton/container.hpp>
+#include <proton/delivery.h>
+#include <proton/delivery.hpp>
+#include <proton/link.hpp>
+#include <proton/listen_handler.hpp>
+#include <proton/listener.hpp>
+#include <proton/message.hpp>
+#include <proton/message_id.hpp>
+#include <proton/messaging_handler.hpp>
+#include <proton/tracker.hpp>
+#include <proton/types.h>
+#include <proton/types.hpp>
+#include <proton/value.hpp>
+
+#include "proton/error_condition.hpp"
+#include "proton/internal/pn_unique_ptr.hpp"
+#include "proton/receiver_options.hpp"
+#include "proton/transport.hpp"
+#include "proton/work_queue.hpp"
+#include "test_bits.hpp"
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+namespace {
+std::mutex m;
+std::condition_variable cv;
+bool listener_ready = false;
+int listener_port;
+} // namespace
+
+class test_recv : public proton::messaging_handler {
+ private:
+ class listener_ready_handler : public proton::listen_handler {
+ void on_open(proton::listener &l) PN_CPP_OVERRIDE {
+ {
+ std::lock_guard<std::mutex> lk(m);
+ listener_port = l.port();
+ listener_ready = true;
+ }
+ cv.notify_one();
+ }
+ };
+
+ std::string url;
+ proton::listener listener;
+ listener_ready_handler listen_handler;
+
+ public:
+ test_recv(const std::string &s) : url(s) {}
+
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ listener = c.listen(url, listen_handler);
+ }
+
+ void on_message(proton::delivery &d, proton::message &msg) PN_CPP_OVERRIDE {
+ proton::binary test_tag_recv("TESTTAG");
+ ASSERT_EQUAL(test_tag_recv, d.tag());
+ d.receiver().close();
+ d.connection().close();
+ listener.stop();
+ }
+};
+
+class test_send : public proton::messaging_handler {
+ private:
+ std::string url;
+ proton::sender sender;
+
+ public:
+ test_send(const std::string &s) : url(s) {}
+
+ void on_container_start(proton::container &c) PN_CPP_OVERRIDE {
+ proton::connection_options co;
+ sender = c.open_sender(url, co);
+ }
+
+ void on_sendable(proton::sender &s) PN_CPP_OVERRIDE {
+ proton::message msg;
+ msg.body("message");
+ proton::binary test_tag_send("TESTTAG");
+ s.send(msg, test_tag_send);
+ }
+};
+
+int test_delivery_tag() {
+ std::string recv_address("127.0.0.1:0/test");
+
+ test_recv recv(recv_address);
+ proton::container c(recv);
+ std::thread thread_recv([&c]() -> void { c.run(); });
+
+ // wait until listener is ready
+ std::unique_lock<std::mutex> lk(m);
+ cv.wait(lk, [] { return listener_ready; });
+
+ std::string send_address =
+ "127.0.0.1:" + std::to_string(listener_port) + "/test";
+ test_send send(send_address);
+ proton::container(send).run();
+ thread_recv.join();
+
+ return 0;
+}
+
+int main(int argc, char **argv) {
+ int failed = 0;
+ RUN_ARGV_TEST(failed, test_delivery_tag());
+ return failed;
+}
diff --git a/cpp/src/sender.cpp b/cpp/src/sender.cpp
index 003adf8..a956257 100644
--- a/cpp/src/sender.cpp
+++ b/cpp/src/sender.cpp
@@ -66,8 +66,15 @@ uint64_t tag_counter = 0;
tracker sender::send(const message &message) {
uint64_t id = ++tag_counter;
- pn_delivery_t *dlv =
- pn_delivery(pn_object(), pn_dtag(reinterpret_cast<const char*>(&id), sizeof(id)));
+ const uint8_t *begin = reinterpret_cast<const uint8_t *>(&id);
+ const binary tag(begin, begin + sizeof(id));
+ return send(message, tag);
+}
+
+tracker sender::send(const message &message, const binary &tag) {
+ pn_delivery_t *dlv = pn_delivery(
+ pn_object(),
+ pn_dtag((reinterpret_cast<const char *>(&tag[0])), tag.size()));
std::vector<char> buf;
message.encode(buf);
assert(!buf.empty());
diff --git a/cpp/tests.cmake b/cpp/tests.cmake
index d30fce9..2063ba2 100644
--- a/cpp/tests.cmake
+++ b/cpp/tests.cmake
@@ -61,6 +61,7 @@ add_cpp_test(container_test)
add_cpp_test(reconnect_test)
add_cpp_test(link_test)
add_cpp_test(credit_test)
+add_cpp_test(delivery_test)
if (ENABLE_JSONCPP)
add_cpp_test(connect_config_test)
target_link_libraries(connect_config_test qpid-proton-core) # For pn_sasl_enabled
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org