Adds rewind function to allow curl seek and retry sending data to server

ACL/include/ACL/Transport/DownchannelHandler.h

@@ -67,6 +67,7 @@ class DownchannelHandler
     /// @{
     std::vector<std::string> getRequestHeaderLines() override;
     avsCommon::utils::http2::HTTP2SendDataResult onSendData(char* bytes, size_t size) override;
+    bool rewindData() override;
     /// @}
 
     /// @name MimeResponseStatusHandlerInterface

ACL/include/ACL/Transport/MessageRequestHandler.h

@@ -88,6 +88,7 @@ class MessageRequestHandler
     std::vector<std::string> getRequestHeaderLines() override;
     avsCommon::utils::http2::HTTP2GetMimeHeadersResult getMimePartHeaderLines() override;
     avsCommon::utils::http2::HTTP2SendDataResult onSendMimePartData(char* bytes, size_t size) override;
+    bool rewindData() override;
     /// @}
 
     /// @name MimeResponseStatusHandlerInterface
@@ -124,6 +125,9 @@ class MessageRequestHandler
 
     /// Response code received through @c onReciveResponseCode (or zero).
     long m_responseCode;
+
+    /// Number of bytes read from the attachment
+    int64_t m_bytesReadFromAttachmentReader;
 };
 
 }  // namespace acl

ACL/include/ACL/Transport/PingHandler.h

@@ -64,6 +64,7 @@ class PingHandler
     /// @{
     std::vector<std::string> getRequestHeaderLines() override;
     avsCommon::utils::http2::HTTP2SendDataResult onSendData(char* bytes, size_t size) override;
+    bool rewindData() override;
     /// @}
 
     /// @name HTTP2ResponseSinkInterface methods

ACL/src/Transport/DownchannelHandler.cpp

@@ -93,6 +93,14 @@ HTTP2SendDataResult DownchannelHandler::onSendData(char* bytes, size_t size) {
     return HTTP2SendDataResult::COMPLETE;
 }
 
+bool DownchannelHandler::rewindData() {
+    ACSDK_DEBUG9(LX(__func__));
+    // no-op: this function is only called
+    // during data upload and this class
+    // does not upload data
+    return true;
+}
+
 DownchannelHandler::DownchannelHandler(
     std::shared_ptr<ExchangeHandlerContextInterface> context,
     const std::string& authToken) :

ACL/src/Transport/MessageRequestHandler.cpp

@@ -135,7 +135,8 @@ MessageRequestHandler::MessageRequestHandler(
         m_countOfPartsSent{0},
         m_wasMessageRequestAcknowledgeReported{false},
         m_wasMessageRequestFinishedReported{false},
-        m_responseCode{0} {
+        m_responseCode{0},
+        m_bytesReadFromAttachmentReader{0} {
     ACSDK_DEBUG7(LX(__func__).d("context", context.get()).d("messageRequest", messageRequest.get()));
 }
 
@@ -204,6 +205,7 @@ HTTP2SendDataResult MessageRequestHandler::onSendMimePartData(char* bytes, size_
     } else if (m_namedReader) {
         auto readStatus = AttachmentReader::ReadStatus::OK;
         auto bytesRead = m_namedReader->reader->read(bytes, size, &readStatus);
+        m_bytesReadFromAttachmentReader += bytesRead;
         ACSDK_DEBUG9(LX("attachmentRead").d("readStatus", (int)readStatus).d("bytesRead", bytesRead));
         switch (readStatus) {
             // The good cases.
@@ -236,6 +238,25 @@ HTTP2SendDataResult MessageRequestHandler::onSendMimePartData(char* bytes, size_
     return HTTP2SendDataResult::ABORT;
 }
 
+bool MessageRequestHandler::rewindData() {
+    ACSDK_DEBUG9(LX(__func__).d("m_bytesReadFromAttachmentReader", m_bytesReadFromAttachmentReader));
+    if (!m_namedReader->reader->seekRelativeBytes(-m_bytesReadFromAttachmentReader)) {
+        ACSDK_ERROR(LX(__func__).m("Could not seek!"));
+        return false;
+    }
+
+    // Reset mime parts
+    m_jsonNext = m_json.c_str();
+    m_countOfJsonBytesLeft = m_json.size();
+    m_countOfPartsSent = 0;
+    m_wasMessageRequestAcknowledgeReported = false;
+    m_wasMessageRequestFinishedReported = false;
+    m_responseCode = 0;
+    m_bytesReadFromAttachmentReader = 0;
+
+    return true;
+}
+
 void MessageRequestHandler::onActivity() {
     m_context->onActivity();
 }

ACL/src/Transport/PingHandler.cpp

@@ -106,6 +106,14 @@ HTTP2SendDataResult PingHandler::onSendData(char* bytes, size_t size) {
     return HTTP2SendDataResult::COMPLETE;
 }
 
+bool PingHandler::rewindData() {
+    ACSDK_DEBUG9(LX(__func__));
+    // no-op: this function is only called
+    // during data upload and this class
+    // does not upload data
+    return true;
+}
+
 bool PingHandler::onReceiveResponseCode(long responseCode) {
     ACSDK_DEBUG5(LX(__func__).d("responseCode", responseCode));
 

AVSCommon/AVS/include/AVSCommon/AVS/Attachment/AttachmentReader.h

@@ -93,6 +93,16 @@ class AttachmentReader {
      */
     virtual bool seek(uint64_t offset) = 0;
 
+    /**
+     * Seeks a relative number of bytes from the
+     * current position in the stream
+     *
+     * @param offset The offset (in bytes! not words!) to seek to within the @c Attachment.
+     * @return @c true if the specified position points at unexpired data, or @c false otherwise. Note that it is valid
+     * to seek into a future index that has not been written to yet.
+     */
+    virtual bool seekRelativeBytes(int64_t offsetInBytes) = 0;
+
     /**
      * Utility function to return the number of bytes in an attachment.
      *

AVSCommon/AVS/include/AVSCommon/AVS/Attachment/InProcessAttachmentReader.h

@@ -74,6 +74,8 @@ class InProcessAttachmentReader : public AttachmentReader {
 
     bool seek(uint64_t offset) override;
 
+    bool seekRelativeBytes(int64_t offsetInBytes) override;
+
     uint64_t getNumUnreadBytes() override;
 
 private:

AVSCommon/AVS/src/Attachment/InProcessAttachmentReader.cpp

@@ -190,6 +190,21 @@ bool InProcessAttachmentReader::seek(uint64_t offset) {
     return false;
 }
 
+bool InProcessAttachmentReader::seekRelativeBytes(int64_t offsetInBytes) {
+    if (m_reader) {
+        int64_t wordOffset = offsetInBytes / static_cast<int64_t>(m_reader->getWordSize());
+        ACSDK_DEBUG9(LX("seekRelativeBytes").d("wordOffset", wordOffset));
+        if (wordOffset < 0) {
+            // Seek back
+            return m_reader->seek(-wordOffset, utils::sds::InProcessSDS::Reader::Reference::BEFORE_READER);
+        } else {
+            // Seek forward
+            return m_reader->seek(wordOffset, utils::sds::InProcessSDS::Reader::Reference::AFTER_READER);
+        }
+    }
+    return false;
+}
+
 uint64_t InProcessAttachmentReader::getNumUnreadBytes() {
     if (m_reader) {
         return m_reader->tell(utils::sds::InProcessSDS::Reader::Reference::BEFORE_WRITER);

AVSCommon/Utils/include/AVSCommon/Utils/HTTP2/HTTP2MimeRequestEncoder.h

@@ -52,6 +52,7 @@ class HTTP2MimeRequestEncoder : public HTTP2RequestSourceInterface {
     /// @name HTTP2RequestSourceInterface methods.
     /// @{
     HTTP2SendDataResult onSendData(char* bytes, size_t size) override;
+    bool rewindData() override;
     std::vector<std::string> getRequestHeaderLines() override;
     /// @}
 

AVSCommon/Utils/include/AVSCommon/Utils/HTTP2/HTTP2MimeRequestSourceInterface.h

@@ -74,6 +74,14 @@ class HTTP2MimeRequestSourceInterface {
      * @see HTTPSendMimePartDataResult.
      */
     virtual HTTP2SendDataResult onSendMimePartData(char* bytes, size_t size) = 0;
+
+    /**
+     * Rewinds the data to the beginning. Used for uploading
+     * data to the server again if there is a connection issue.
+     *
+     * @return true, if the rewind was successful. False otherwise
+     */
+    virtual bool rewindData() = 0;
 };
 
 }  // namespace http2

AVSCommon/Utils/include/AVSCommon/Utils/HTTP2/HTTP2RequestSourceInterface.h

@@ -59,6 +59,14 @@ class HTTP2RequestSourceInterface {
      * @return Result indicating the disposition of the operation and number of bytes copied.  @see HTTPSendDataResult.
      */
     virtual HTTP2SendDataResult onSendData(char* bytes, size_t size) = 0;
+
+    /**
+     * Rewinds the data to the beginning. Used for uploading
+     * data to the server again if there is a connection issue.
+     *
+     * @return true, if the rewind was successful. False otherwise
+     */
+    virtual bool rewindData() = 0;
 };
 
 }  // namespace http2

AVSCommon/Utils/include/AVSCommon/Utils/LibcurlUtils/CurlEasyHandleWrapper.h

@@ -64,6 +64,18 @@ class CurlEasyHandleWrapper {
     using CurlDebugCallback =
         int (*)(CURL* handle, curl_infotype infoType, char* buffer, size_t blockSize, void* userData);
 
+
+    /**
+     * Callback signature of the libcurl SEEK function
+     *
+     * https://curl.haxx.se/libcurl/c/CURLOPT_SEEKFUNCTION.html
+     *
+     * @param userp User data as set with CURLOPT_SEEKDATA
+     * @param offset Absolute index if `SEEK_SET`. Relative delta if `SEEK_CUR` or `SEEK_END`
+     * @param origin Always SEEK_SET from <stdio.h>
+     */
+    using CurlSeekCallback = int (*)(void *userp, curl_off_t offset, int origin);
+
     /**
      * Definitions for HTTP action types
      */
@@ -213,6 +225,16 @@ class CurlEasyHandleWrapper {
      */
     bool setReadCallback(CurlCallback callback, void* userData);
 
+    /**
+     * Sets the callback to call when libcurl needs to retry
+     * sending post data
+     *
+     * @param callback A function pointer to the seek callback
+     * @param userData Any data to be passed to the callback
+     * @return Whether the addition was successful
+     */
+    bool setSeekCallback(CurlSeekCallback callback, void* userData);
+
     /**
      * Helper function for calling curl_easy_setopt and checking the result.
      *

AVSCommon/Utils/include/AVSCommon/Utils/LibcurlUtils/LibcurlHTTP2Request.h

@@ -148,6 +148,20 @@ class LibcurlHTTP2Request : public alexaClientSDK::avsCommon::utils::http2::HTTP
      */
     static size_t readCallback(char* data, size_t size, size_t nmemb, void* userData);
 
+    /**
+     * Callback that gets executed when curl needs to retry sending data
+     *
+     * @see CurlEasyHandleWrapper::CurlSeekCallback for details
+     * The function shall work like fseek(3) or lseek(3) and it gets SEEK_SET, SEEK_CUR or SEEK_END as argument for
+     * origin, although libcurl currently only passes SEEK_SET.
+     *
+     * @param userData  Context passed back from @c libcurl.  Should always be a pointer to @c LibcurlHTTP2Request.
+     * @param offset Absolute index if `SEEK_SET`. Relative delta if `SEEK_CUR` or `SEEK_END`
+     * @param origin always SEEK_SET from <stdio.h>
+     * @return CURL_SEEKFUNC_OK(0) for success, CURL_SEEKFUNC_FAIL(1) for fail, CURL_SEEKFUNC_CANTSEEK(2) for can't seek
+     */
+    static int seekCallback(void *userData, curl_off_t offset, int origin);
+
     /**
      * Returns the HTTP response code to this request.
      *

AVSCommon/Utils/src/HTTP2/HTTP2MimeRequestEncoder.cpp

@@ -239,6 +239,18 @@ HTTP2SendDataResult HTTP2MimeRequestEncoder::onSendData(char* bytes, size_t size
     }
 }
 
+bool HTTP2MimeRequestEncoder::rewindData() {
+    ACSDK_DEBUG9(LX(__func__));
+    if (m_source->rewindData()) {
+        m_state = State::NEW;
+        m_getMimeHeaderLinesResult = HTTP2GetMimeHeadersResult::ABORT;
+        m_headerLine = m_getMimeHeaderLinesResult.headers.begin();
+        m_stringIndex = 0;
+        return true;
+    }
+    return false;
+}
+
 std::vector<std::string> HTTP2MimeRequestEncoder::getRequestHeaderLines() {
     ACSDK_DEBUG9(LX(__func__));
     if (m_source) {

AVSCommon/Utils/src/LibcurlUtils/CurlEasyHandleWrapper.cpp

@@ -270,6 +270,10 @@ bool CurlEasyHandleWrapper::setReadCallback(CurlCallback callback, void* userDat
     return setopt(CURLOPT_READFUNCTION, callback) && (!userData || setopt(CURLOPT_READDATA, userData));
 }
 
+bool CurlEasyHandleWrapper::setSeekCallback(CurlSeekCallback callback, void* userData) {
+    return setopt(CURLOPT_SEEKFUNCTION, callback) && (!userData || setopt(CURLOPT_SEEKDATA, userData));
+}
+
 std::string CurlEasyHandleWrapper::urlEncode(const std::string& in) const {
     std::string result;
     auto temp = curl_easy_escape(m_handle, in.c_str(), 0);

AVSCommon/Utils/src/LibcurlUtils/LibcurlHTTP2Request.cpp

@@ -117,6 +117,35 @@ size_t LibcurlHTTP2Request::readCallback(char* data, size_t size, size_t nmemb,
     return CURL_READFUNC_ABORT;
 }
 
+int LibcurlHTTP2Request::seekCallback(void *userData, curl_off_t offset, int origin) {
+    if (!userData) {
+        ACSDK_ERROR(LX(__func__).d("reason", "nullUserData"));
+        return CURL_SEEKFUNC_CANTSEEK;
+    }
+
+    LibcurlHTTP2Request* stream = static_cast<LibcurlHTTP2Request*>(userData);
+    ACSDK_DEBUG9(LX(__func__).d("id", stream->getId()).d("offset", offset).d("origin", origin));
+
+    if (offset != 0 || origin != SEEK_SET) {
+        // According the CURL documentation,
+        // they will only ever send `SEEK_SET`
+        // If they send something else, bail
+        //
+        // Our rewind function only supports rewinding
+        // to a specific index. In practice
+        // curl only sends 0. If they don't,
+        // that's unexpected and let's bail early
+        ACSDK_INFO(LX(__func__).m("seekFailed. Invalid offset/origin."));
+        return CURL_SEEKFUNC_CANTSEEK;
+    }
+
+    if (stream->m_source->rewindData()) {
+        return CURL_SEEKFUNC_OK;
+    } else {
+        return CURL_SEEKFUNC_FAIL;
+    }
+}
+
 long LibcurlHTTP2Request::getResponseCode() {
     long responseCode = 0;
     CURLcode ret = curl_easy_getinfo(m_stream.getCurlHandle(), CURLINFO_RESPONSE_CODE, &responseCode);
@@ -152,6 +181,7 @@ LibcurlHTTP2Request::LibcurlHTTP2Request(
         case HTTP2RequestType::POST:
             curl_easy_setopt(m_stream.getCurlHandle(), CURLOPT_POST, 1L);
             m_stream.setReadCallback(LibcurlHTTP2Request::readCallback, this);
+            m_stream.setSeekCallback(LibcurlHTTP2Request::seekCallback, this);
             break;
     }
     m_stream.setURL(config.getUrl());

AVSCommon/Utils/test/AVSCommon/Utils/HTTP2/MockHTTP2MimeRequestEncodeSource.h

@@ -44,6 +44,7 @@ class MockHTTP2MimeRequestEncodeSource : public HTTP2MimeRequestSourceInterface
     /// @{
     HTTP2GetMimeHeadersResult getMimePartHeaderLines() override;
     HTTP2SendDataResult onSendMimePartData(char* bytes, size_t size) override;
+    bool rewindData() override;
     std::vector<std::string> getRequestHeaderLines() override;
     /// @}
 

AVSCommon/Utils/test/Common/MockHTTP2MimeRequestEncodeSource.cpp

@@ -64,6 +64,10 @@ HTTP2SendDataResult MockHTTP2MimeRequestEncodeSource::onSendMimePartData(char* b
     return HTTP2SendDataResult(bytesToWrite);
 }
 
+bool MockHTTP2MimeRequestEncodeSource::rewindData() {
+    return true; // NOT IMPLEMENTED IN TEST
+}
+
 std::vector<std::string> MockHTTP2MimeRequestEncodeSource::getRequestHeaderLines() {
     return std::vector<std::string>();
 }