proposal for unique_ptr and const ref

Signed-off-by: aw <[email protected]>

include/utils/message_queue.hpp

@@ -23,44 +23,44 @@ using json = nlohmann::json;
 struct Message {
     std::string topic;   ///< The MQTT topic where this message originated from
     std::string payload; ///< The message payload
+};
 
-    Message(const std::string& topic, const std::string& payload);
+struct ParsedMessage {
+    std::string topic;
+    json data;
 };
 
+using MessageCallback = std::function<void(const Message&)>;
+
 /// \brief Simple message queue that takes std::string messages, parsed them and dispatches them to handlers
 class MessageQueue {
+
 private:
     std::thread worker_thread;
-    std::queue<std::shared_ptr<Message>> message_queue;
+    std::queue<std::unique_ptr<Message>> message_queue;
     std::mutex queue_ctrl_mutex;
-    std::function<void(std::shared_ptr<Message> message)> message_callback;
+    MessageCallback message_callback;
     std::condition_variable cv;
     bool running;
 
 public:
     /// \brief Creates a message queue with the provided \p message_callback
-    explicit MessageQueue(const std::function<void(std::shared_ptr<Message> message)>& message_callback);
+    explicit MessageQueue(MessageCallback);
     ~MessageQueue();
 
     /// \brief Adds a \p message to the message queue which will then be delivered to the message callback
-    void add(std::shared_ptr<Message> message);
+    void add(std::unique_ptr<Message>);
 
     /// \brief Stops the message queue
     void stop();
 };
 
 /// \brief Contains a message queue driven list of handler callbacks
 class MessageHandler {
-public:
-    struct MessageDetails {
-        std::string topic;
-        std::shared_ptr<json> data;
-    };
-
 private:
     std::unordered_set<std::shared_ptr<TypedHandler>> handlers;
     std::thread handler_thread;
-    std::queue<MessageDetails> message_queue;
+    std::queue<std::unique_ptr<ParsedMessage>> message_queue;
     std::mutex handler_ctrl_mutex;
     std::mutex handler_list_mutex;
     std::condition_variable cv;
@@ -74,7 +74,7 @@ class MessageHandler {
     ~MessageHandler();
 
     /// \brief Adds a \p message to the message queue which will be delivered to the registered handlers
-    void add(MessageDetails message);
+    void add(std::unique_ptr<ParsedMessage>);
 
     /// \brief Stops the message handler
     void stop();

include/utils/mqtt_abstraction_impl.hpp

@@ -129,7 +129,7 @@ class MQTTAbstractionImpl {
     static int open_nb_socket(const char* addr, const char* port);
     bool connectBroker(std::string& socket_path);
     bool connectBroker(const char* host, const char* port);
-    void on_mqtt_message(std::shared_ptr<Message> message);
+    void on_mqtt_message(const Message& message);
     void on_mqtt_connect();
     static void on_mqtt_disconnect();
 

lib/message_queue.cpp

@@ -10,34 +10,30 @@
 
 namespace Everest {
 
-Message::Message(const std::string& topic, const std::string& payload) : topic(topic), payload(payload) {
-}
-
-MessageQueue::MessageQueue(const std::function<void(std::shared_ptr<Message> message)>& message_callback) :
-    message_callback(message_callback), running(true) {
+MessageQueue::MessageQueue(MessageCallback message_callback_) :
+    message_callback(std::move(message_callback_)), running(true) {
     this->worker_thread = std::thread([this]() {
         while (true) {
-            std::shared_ptr<Message> message;
             std::unique_lock<std::mutex> lock(this->queue_ctrl_mutex);
             this->cv.wait(lock, [this]() { return !this->message_queue.empty() || this->running == false; });
             if (!this->running) {
                 return;
             }
 
-            message = this->message_queue.front();
+            const auto message = std::move(this->message_queue.front());
             this->message_queue.pop();
             lock.unlock();
 
             // pass the message to the message callback
-            this->message_callback(message);
+            this->message_callback(*message);
         }
     });
 }
 
-void MessageQueue::add(std::shared_ptr<Message> message) {
+void MessageQueue::add(std::unique_ptr<Message> message) {
     {
         std::lock_guard<std::mutex> lock(this->queue_ctrl_mutex);
-        this->message_queue.push(message);
+        this->message_queue.push(std::move(message));
     }
     this->cv.notify_all();
 }
@@ -64,11 +60,11 @@ MessageHandler::MessageHandler() : running(true) {
                 return;
             }
 
-            auto message = std::move(this->message_queue.front());
+            const auto message = std::move(this->message_queue.front());
             this->message_queue.pop();
             lock.unlock();
 
-            auto& data = *message.data;
+            const auto& data = message->data;
 
             // get the registered handlers
             std::vector<std::shared_ptr<TypedHandler>> local_handlers;
@@ -89,7 +85,7 @@ MessageHandler::MessageHandler() : running(true) {
                         continue;
                     }
                     if (data.at("type") == "call") {
-                        handler_fn(message.topic, data.at("data"));
+                        handler_fn(message->topic, data.at("data"));
                     }
                 } else if (handler->type == HandlerType::Result) {
                     // unpack result
@@ -99,25 +95,25 @@ MessageHandler::MessageHandler() : running(true) {
                     if (data.at("type") == "result") {
                         // only deliver result to handler with matching id
                         if (handler->id == data.at("data").at("id")) {
-                            handler_fn(message.topic, data.at("data"));
+                            handler_fn(message->topic, data.at("data"));
                         }
                     }
                 } else if (handler->type == HandlerType::SubscribeVar) {
                     // unpack var
                     if (handler->name != data.at("name")) {
                         continue;
                     }
-                    handler_fn(message.topic, data.at("data"));
+                    handler_fn(message->topic, data.at("data"));
                 } else {
                     // external or unknown, no preprocessing
-                    handler_fn(message.topic, data);
+                    handler_fn(message->topic, data);
                 }
             }
         }
     });
 }
 
-void MessageHandler::add(MessageDetails message) {
+void MessageHandler::add(std::unique_ptr<ParsedMessage> message) {
     {
         std::lock_guard<std::mutex> lock(this->handler_ctrl_mutex);
         this->message_queue.push(std::move(message));

lib/mqtt_abstraction_impl.cpp

@@ -26,13 +26,13 @@ namespace Everest {
 const auto mqtt_keep_alive = 600;
 
 MessageWithQOS::MessageWithQOS(const std::string& topic, const std::string& payload, QOS qos) :
-    Message(topic, payload), qos(qos) {
+    Message{topic, payload}, qos(qos) {
 }
 
 MQTTAbstractionImpl::MQTTAbstractionImpl(const std::string& mqtt_server_address, const std::string& mqtt_server_port,
                                          const std::string& mqtt_everest_prefix,
                                          const std::string& mqtt_external_prefix) :
-    message_queue(([this](std::shared_ptr<Message> message) { this->on_mqtt_message(message); })),
+    message_queue(([this](const Message& message) { this->on_mqtt_message(message); })),
     mqtt_server_address(mqtt_server_address),
     mqtt_server_port(mqtt_server_port),
     mqtt_everest_prefix(mqtt_everest_prefix),
@@ -57,7 +57,7 @@ MQTTAbstractionImpl::MQTTAbstractionImpl(const std::string& mqtt_server_address,
 MQTTAbstractionImpl::MQTTAbstractionImpl(const std::string& mqtt_server_socket_path,
                                          const std::string& mqtt_everest_prefix,
                                          const std::string& mqtt_external_prefix) :
-    message_queue(([this](std::shared_ptr<Message> message) { this->on_mqtt_message(message); })),
+    message_queue(([this](const Message& message) { this->on_mqtt_message(message); })),
     mqtt_server_socket_path(mqtt_server_socket_path),
     mqtt_everest_prefix(mqtt_everest_prefix),
     mqtt_external_prefix(mqtt_external_prefix),
@@ -265,28 +265,28 @@ std::future<void> MQTTAbstractionImpl::spawn_main_loop_thread() {
     return future;
 }
 
-void MQTTAbstractionImpl::on_mqtt_message(std::shared_ptr<Message> message) {
+void MQTTAbstractionImpl::on_mqtt_message(const Message& message) {
     BOOST_LOG_FUNCTION();
 
-    const std::string& topic = message->topic;
-    const std::string& payload = message->payload;
+    const auto& topic = message.topic;
+    const auto& payload = message.payload;
 
     try {
-        std::shared_ptr<json> data;
+        json data;
         bool is_everest_topic = false;
         if (topic.find(mqtt_everest_prefix) == 0) {
             EVLOG_verbose << fmt::format("topic {} starts with {}", topic, mqtt_everest_prefix);
             is_everest_topic = true;
             try {
-                data = std::make_shared<json>(json::parse(payload));
+                data = json::parse(payload);
             } catch (nlohmann::detail::parse_error& e) {
                 EVLOG_warning << fmt::format("Could not decode json for incoming topic '{}': {}", topic, payload);
                 return;
             }
         } else {
             EVLOG_debug << fmt::format("Message parsing for topic '{}' not implemented. Wrapping in json object.",
                                        topic);
-            data = std::make_shared<json>(json(payload));
+            data = json(payload);
         }
 
         bool found = false;
@@ -306,7 +306,7 @@ void MQTTAbstractionImpl::on_mqtt_message(std::shared_ptr<Message> message) {
 
             if (topic_matches) {
                 found = true;
-                handler.add({topic, data});
+                handler.add(std::unique_ptr<ParsedMessage>(new ParsedMessage{topic, std::move(data)}));
             }
         }
         lock.unlock();
@@ -642,9 +642,9 @@ void MQTTAbstractionImpl::publish_callback(void** state, struct mqtt_response_pu
     auto message_queue = static_cast<MessageQueue*>(*state);
 
     // topic_name and application_message are NOT null-terminated, hence copy construct strings
-    message_queue->add(std::make_shared<Message>(
+    message_queue->add(std::unique_ptr<Message>(new Message{
         std::string(static_cast<const char*>(published->topic_name), published->topic_name_size),
-        std::string(static_cast<const char*>(published->application_message), published->application_message_size)));
+        std::string(static_cast<const char*>(published->application_message), published->application_message_size)}));
 }
 
 } // namespace Everest