Fix for curl GH workflow failures

Author: embhorn

examples/mqttnet.c

@@ -952,6 +952,7 @@ static int NetConnect(void *context, const char* host, word16 port,
         return rc;
     }
 
+    sock->bytes = 0;
     sock->stat = SOCK_CONN;
     return MQTT_CODE_SUCCESS;
 }
@@ -995,59 +996,44 @@ static int NetWrite(void *context, const byte* buf, int buf_len,
     PRINTF("sock->curl = %p, sockfd = %d", (void *)sock->curl, sockfd);
 #endif
 
-    /* A very simple retry with timeout example. This assumes the entire
-     * payload will be transferred in a single shot without buffering.
-     * todo: add buffering? */
-    for (size_t i = 0; i < MQTT_CURL_NUM_RETRY; ++i) {
-    #ifdef WOLFMQTT_MULTITHREAD
+#ifdef WOLFMQTT_MULTITHREAD
+    {
         int rc = wm_SemLock(&sock->mqttCtx->client.lockCURL);
         if (rc != 0) {
             return rc;
         }
-    #endif
-
-        res = curl_easy_send(sock->curl, buf, buf_len, &sent);
-
-    #ifdef WOLFMQTT_MULTITHREAD
-        wm_SemUnlock(&sock->mqttCtx->client.lockCURL);
-    #endif
-
-        if (res == CURLE_OK) {
-            #if defined(WOLFMQTT_DEBUG_SOCKET)
-            PRINTF("info: curl_easy_send(%d) returned: %d, %s", buf_len, res,
-                   curl_easy_strerror(res));
-            #endif
-            break;
-        }
+    }
+#endif
 
-        if (res == CURLE_AGAIN) {
-            #if defined(WOLFMQTT_DEBUG_SOCKET)
-            PRINTF("info: curl_easy_send(%d) returned: %d, %s", buf_len, res,
-                   curl_easy_strerror(res));
-            #endif
+    res = curl_easy_send(sock->curl, &buf[sock->bytes],
+                         buf_len - sock->bytes, &sent);
 
-            wait_rc = mqttcurl_wait(sockfd, 0, timeout_ms,
-                                    sock->mqttCtx->test_mode);
+#ifdef WOLFMQTT_MULTITHREAD
+    wm_SemUnlock(&sock->mqttCtx->client.lockCURL);
+#endif
 
-            if (wait_rc == MQTT_CODE_CONTINUE) {
-                continue;
-            }
-            else {
-                return wait_rc;
-            }
+    if (res == CURLE_OK) {
+        sock->bytes += (int)sent;
+        if (sock->bytes < buf_len) {
+            /* Partial write, return continue to retry */
+            return MQTT_CODE_CONTINUE;
         }
-
-        PRINTF("error: curl_easy_send(%d) returned: %d, %s", buf_len, res,
-               curl_easy_strerror(res));
-        return MQTT_CODE_ERROR_CURL;
+        /* Complete, reset for next operation */
+        sent = sock->bytes;
+        sock->bytes = 0;
+        return (int)sent;
     }
 
-    if ((int) sent != buf_len) {
-        PRINTF("error: sent %d bytes, expected %d", (int)sent, buf_len);
-        return MQTT_CODE_ERROR_CURL;
+    if (res == CURLE_AGAIN) {
+        wait_rc = mqttcurl_wait(sockfd, 0, timeout_ms,
+                                sock->mqttCtx->test_mode);
+        return wait_rc;
     }
 
-    return buf_len;
+    PRINTF("error: curl_easy_send(%d) returned: %d, %s", buf_len, res,
+           curl_easy_strerror(res));
+    sock->bytes = 0;
+    return MQTT_CODE_ERROR_CURL;
 }
 
 static int NetRead(void *context, byte* buf, int buf_len,
@@ -1089,59 +1075,50 @@ static int NetRead(void *context, byte* buf, int buf_len,
     PRINTF("sock->curl = %p, sockfd = %d", (void *)sock->curl, sockfd);
 #endif
 
-    /* A very simple retry with timeout example. This assumes the entire
-     * payload will be transferred in a single shot without buffering.
-     * todo: add buffering? */
-    for (size_t i = 0; i < MQTT_CURL_NUM_RETRY; ++i) {
-    #ifdef WOLFMQTT_MULTITHREAD
+#ifdef WOLFMQTT_MULTITHREAD
+    {
         int rc = wm_SemLock(&sock->mqttCtx->client.lockCURL);
         if (rc != 0) {
             return rc;
         }
-    #endif
+    }
+#endif
 
-        res = curl_easy_recv(sock->curl, buf, buf_len, &recvd);
+    res = curl_easy_recv(sock->curl, &buf[sock->bytes],
+                         buf_len - sock->bytes, &recvd);
 
-    #ifdef WOLFMQTT_MULTITHREAD
-        wm_SemUnlock(&sock->mqttCtx->client.lockCURL);
-    #endif
+#ifdef WOLFMQTT_MULTITHREAD
+    wm_SemUnlock(&sock->mqttCtx->client.lockCURL);
+#endif
 
-        if (res == CURLE_OK) {
-            #if defined(WOLFMQTT_DEBUG_SOCKET)
-            PRINTF("info: curl_easy_recv(%d) returned: %d, %s", buf_len, res,
-                   curl_easy_strerror(res));
-            #endif
-            break;
+    if (res == CURLE_OK) {
+        if (recvd == 0) {
+            /* Connection closed */
+            PRINTF("error: connection closed by peer");
+            sock->bytes = 0;
+            return MQTT_CODE_ERROR_NETWORK;
         }
-
-        if (res == CURLE_AGAIN) {
-            #if defined(WOLFMQTT_DEBUG_SOCKET)
-            PRINTF("info: curl_easy_recv(%d) returned: %d, %s", buf_len, res,
-                   curl_easy_strerror(res));
-            #endif
-
-            wait_rc = mqttcurl_wait(sockfd, 1, timeout_ms,
-                                    sock->mqttCtx->test_mode);
-
-            if (wait_rc == MQTT_CODE_CONTINUE) {
-                continue;
-            }
-            else {
-                return wait_rc;
-            }
+        sock->bytes += (int)recvd;
+        if (sock->bytes < buf_len) {
+            /* Partial read, return continue to retry */
+            return MQTT_CODE_CONTINUE;
         }
-
-        PRINTF("error: curl_easy_recv(%d) returned: %d, %s", buf_len, res,
-               curl_easy_strerror(res));
-        return MQTT_CODE_ERROR_CURL;
+        /* Complete, reset for next operation */
+        recvd = sock->bytes;
+        sock->bytes = 0;
+        return (int)recvd;
     }
 
-    if ((int) recvd != buf_len) {
-        PRINTF("error: recvd %d bytes, expected %d", (int)recvd, buf_len);
-        return MQTT_CODE_ERROR_CURL;
+    if (res == CURLE_AGAIN) {
+        wait_rc = mqttcurl_wait(sockfd, 1, timeout_ms,
+                                sock->mqttCtx->test_mode);
+        return wait_rc;
     }
 
-    return buf_len;
+    PRINTF("error: curl_easy_recv(%d) returned: %d, %s", buf_len, res,
+           curl_easy_strerror(res));
+    sock->bytes = 0;
+    return MQTT_CODE_ERROR_CURL;
 }
 
 static int NetDisconnect(void *context)
@@ -1159,6 +1136,7 @@ static int NetDisconnect(void *context)
         curl_easy_cleanup(sock->curl);
         sock->curl = NULL;
     }
+    sock->bytes = 0;
 
     return 0;
 }

examples/mqttnet.h

@@ -58,6 +58,7 @@ typedef struct _SocketContext {
 #endif
 #ifdef ENABLE_MQTT_CURL
     CURL  *  curl;
+    int      bytes; /* track partial read/write */
 #endif
 #ifdef ENABLE_MQTT_WEBSOCKET
     void* websocket_ctx;