bind and setup manual puback

Author: sbSteveK

crt/aws-c-mqtt

@@ -250,13 +250,50 @@ namespace Aws
                 OnStoppedEventData() {}
             };
 
+            /**
+             * An opaque handle representing manual control over a QoS 1 PUBACK for a received PUBLISH packet.
+             *
+             * Obtained by calling acquirePubackControl() within the OnPublishReceivedHandler callback.
+             * Pass this handle to Mqtt5Client::InvokePuback() at any later time to send the PUBACK.
+             *
+             * @note acquirePubackControl() must be called within the OnPublishReceivedHandler callback.
+             *       Calling it after the callback returns will return nullptr.
+             */
+            class AWS_CRT_CPP_API PubackControlHandle
+            {
+                friend class Mqtt5Client;
+                friend class Mqtt5ClientCore;
+
+              public:
+                PubackControlHandle() noexcept : m_controlId(0) {}
+
+              private:
+                explicit PubackControlHandle(uint64_t controlId) noexcept : m_controlId(controlId) {}
+
+                uint64_t m_controlId;
+            };
+
             /**
              * The data returned when a publish is made to a topic the MQTT5 client is subscribed to.
              */
             struct AWS_CRT_CPP_API PublishReceivedEventData
             {
                 PublishReceivedEventData() : publishPacket(nullptr) {}
+
                 std::shared_ptr<PublishPacket> publishPacket;
+
+                /**
+                 * Call this function within the OnPublishReceivedHandler callback to take manual control of the
+                 * PUBACK for this QoS 1 message, preventing the client from automatically sending a PUBACK.
+                 *
+                 * Returns a shared_ptr to a PubackControlHandle that can be passed to Mqtt5Client::InvokePuback()
+                 * to send the PUBACK to the broker.
+                 *
+                 * @note This function must be called within the OnPublishReceivedHandler callback.
+                 *       Calling it after the callback returns will return nullptr.
+                 * @note Only relevant for QoS 1 messages. Returns nullptr for QoS 0 messages.
+                 */
+                std::function<std::shared_ptr<PubackControlHandle>()> acquirePubackControl;
             };
 
             /**
@@ -433,6 +470,19 @@ namespace Aws
                  */
                 const Mqtt5ClientOperationStatistics &GetOperationStatistics() noexcept;
 
+                /**
+                 * Sends a PUBACK packet for a QoS 1 PUBLISH that was previously acquired for manual control.
+                 *
+                 * To use manual PUBACK control, call eventData.acquirePubackControl() within the
+                 * OnPublishReceivedHandler callback to obtain a PubackControlHandle. Then call this method
+                 * to send the PUBACK.
+                 *
+                 * @param pubackControlHandle handle obtained from acquirePubackControl()
+                 *
+                 * @return true if the operation succeeded, otherwise false
+                 */
+                bool InvokePuback(const PubackControlHandle &pubackControlHandle) noexcept;
+
                 ~Mqtt5Client();
 
                 struct aws_mqtt5_client *GetUnderlyingHandle() const noexcept;

include/aws/crt/mqtt/Mqtt5Client.h

@@ -96,6 +96,15 @@ namespace Aws
                     std::shared_ptr<UnsubscribePacket> unsubscribeOptions,
                     OnUnsubscribeCompletionHandler onUnsubscribeCompletionCallback = NULL) noexcept;
 
+                /**
+                 * Sends a PUBACK packet for a QoS 1 PUBLISH that was previously acquired for manual control.
+                 *
+                 * @param pubackControlHandle handle obtained from acquirePubackControl()
+                 *
+                 * @return true if the operation succeeded, otherwise false
+                 */
+                bool InvokePuback(const PubackControlHandle &pubackControlHandle) noexcept;
+
                 /**
                  * Tells the Mqtt5ClientCore to release the native client and clean up unhandled the resources
                  * and operations before destroying it. You MUST only call this function when you

include/aws/crt/mqtt/private/Mqtt5ClientCore.h

@@ -158,6 +158,16 @@ namespace Aws
                 return m_client_core->Unsubscribe(unsubscribeOptions, onUnsubscribeCompletionCallback);
             }
 
+            bool Mqtt5Client::InvokePuback(const PubackControlHandle &pubackControlHandle) noexcept
+            {
+                if (m_client_core == nullptr)
+                {
+                    AWS_LOGF_DEBUG(AWS_LS_MQTT5_CLIENT, "Failed to invoke puback: Mqtt5 Client is invalid.");
+                    return false;
+                }
+                return m_client_core->InvokePuback(pubackControlHandle);
+            }
+
             const Mqtt5ClientOperationStatistics &Mqtt5Client::GetOperationStatistics() noexcept
             {
                 aws_mqtt5_client_operation_statistics m_operationStatisticsNative = {0, 0, 0, 0};

source/mqtt/Mqtt5Client.cpp

@@ -189,6 +189,29 @@ namespace Aws
                             client_core->m_allocator, *publish, client_core->m_allocator);
                         PublishReceivedEventData eventData;
                         eventData.publishPacket = packet;
+
+                        /*
+                         * Set up the acquirePubackControl function for QoS 1 messages.
+                         * This lambda captures the raw client pointer and the publish view pointer,
+                         * both of which are valid only during this callback invocation.
+                         * Calling acquirePubackControl after the callback returns will return nullptr
+                         * because the publish pointer will no longer be valid. This is handled.
+                         */
+                        if (publish->qos == AWS_MQTT5_QOS_AT_LEAST_ONCE)
+                        {
+                            aws_mqtt5_client *rawClient = client_core->m_client;
+                            eventData.acquirePubackControl = [rawClient,
+                                                              publish]() -> std::shared_ptr<PubackControlHandle>
+                            {
+                                if (rawClient == nullptr)
+                                {
+                                    return nullptr;
+                                }
+                                uint64_t controlId = aws_mqtt5_client_acquire_puback(rawClient, publish);
+                                return std::make_shared<PubackControlHandle>(PubackControlHandle(controlId));
+                            };
+                        }
+
                         client_core->onPublishReceived(eventData);
                     }
                     else
@@ -612,6 +635,17 @@ namespace Aws
                 return result == AWS_OP_SUCCESS;
             }
 
+            bool Mqtt5ClientCore::InvokePuback(const PubackControlHandle &pubackControlHandle) noexcept
+            {
+                if (m_client == nullptr)
+                {
+                    AWS_LOGF_DEBUG(AWS_LS_MQTT5_CLIENT, "Failed to invoke puback: Mqtt5ClientCore is invalid.");
+                    return false;
+                }
+                return aws_mqtt5_client_invoke_puback(m_client, pubackControlHandle.m_controlId, nullptr) ==
+                       AWS_OP_SUCCESS;
+            }
+
             void Mqtt5ClientCore::Close() noexcept
             {
                 std::lock_guard<std::recursive_mutex> lock(m_callback_lock);