refactor completion/buffering of chunks (#707)

* work on #689

* further improvements of chunk completion

src/ua_connection.c

@@ -32,103 +32,102 @@ void UA_Connection_deleteMembers(UA_Connection *connection) {
 UA_StatusCode
 UA_Connection_completeMessages(UA_Connection *connection, UA_ByteString * UA_RESTRICT message,
                               UA_Boolean * UA_RESTRICT realloced) {
-    UA_ByteString *current = message;
-    *realloced = false;
+    UA_StatusCode retval = UA_STATUSCODE_GOOD;
+
+    /* We have a stored an incomplete chunk. Concat the received message to the end.
+     * After this block, connection->incompleteMessage is always empty. */
     if(connection->incompleteMessage.length > 0) {
-        /* concat the existing incomplete message with the new message */
-        UA_Byte *data = UA_realloc(connection->incompleteMessage.data,
-                                   connection->incompleteMessage.length + message->length);
+        size_t length = connection->incompleteMessage.length + message->length;
+        UA_Byte *data = UA_realloc(connection->incompleteMessage.data, length);
         if(!data) {
-            /* not enough memory */
-            UA_ByteString_deleteMembers(&connection->incompleteMessage);
-            connection->releaseRecvBuffer(connection, message);
-            return UA_STATUSCODE_BADOUTOFMEMORY;
+            retval = UA_STATUSCODE_BADOUTOFMEMORY;
+            goto cleanup;
         }
-        memcpy(&data[connection->incompleteMessage.length], message->data, message->length);
-        connection->incompleteMessage.data = data;
-        connection->incompleteMessage.length += message->length;
+        memcpy(&data[connection->incompleteMessage.length], message->data, length);
         connection->releaseRecvBuffer(connection, message);
-        current = &connection->incompleteMessage;
+        message->data = data;
+        message->length += length;
         *realloced = true;
+        connection->incompleteMessage = UA_BYTESTRING_NULL;
     }
 
-    /* the while loop sets offset to the first element after the last complete message. if a message
-       contains garbage, the buffer length is set to contain only the "good" messages before. */
-    size_t offset = 0;
-    size_t delete_at = current->length-1; // garbled message after this point
-    while(current->length - offset >= 16) {
-        UA_UInt32 msgtype = (UA_UInt32)current->data[offset] +
-            ((UA_UInt32)current->data[offset+1] << 8) +
-            ((UA_UInt32)current->data[offset+2] << 16);
+    /* Loop over the chunks in the received buffer */
+    size_t complete_until = 0; /* the received complete chunks end at this point */
+    UA_Boolean garbage_end = false; /* garbage after the last complete message */
+    while(message->length - complete_until >= 8) {
+        /* Check the message type */
+        UA_UInt32 msgtype = (UA_UInt32)message->data[complete_until] +
+            ((UA_UInt32)message->data[complete_until+1] << 8) +
+            ((UA_UInt32)message->data[complete_until+2] << 16);
         if(msgtype != ('M' + ('S' << 8) + ('G' << 16)) &&
            msgtype != ('O' + ('P' << 8) + ('N' << 16)) &&
            msgtype != ('H' + ('E' << 8) + ('L' << 16)) &&
            msgtype != ('A' + ('C' << 8) + ('K' << 16)) &&
            msgtype != ('C' + ('L' << 8) + ('O' << 16))) {
-            /* the message type is not recognized */
-            delete_at = offset; // throw the remaining message away
+            garbage_end = true; /* the message type is not recognized */
             break;
         }
-        UA_UInt32 length = 0;
-        size_t length_pos = offset + 4;
-        UA_StatusCode retval = UA_UInt32_decodeBinary(current, &length_pos, &length);
-        if(retval != UA_STATUSCODE_GOOD || length < 16 || length > connection->localConf.recvBufferSize) {
-            /* the message size is not allowed. throw the remaining bytestring away */
-            delete_at = offset;
+
+        /* Decode the length of the chunk */
+        UA_UInt32 chunk_length = 0;
+        size_t length_pos = complete_until + 4;
+        UA_StatusCode decode_retval = UA_UInt32_decodeBinary(message, &length_pos, &chunk_length);
+
+        /* The message size is not allowed. Throw the remaining bytestring away */
+        if(decode_retval != UA_STATUSCODE_GOOD || chunk_length < 16 || chunk_length > connection->localConf.recvBufferSize) {
+            garbage_end = true;
             break;
         }
-        if(length + offset > current->length)
-            break; /* the message is incomplete. keep the beginning */
-        offset += length;
-    }
 
-    /* throw the message away */
-    if(delete_at == 0) {
-        if(!*realloced) {
-            connection->releaseRecvBuffer(connection, message);
-            *realloced = true;
-        } else
-            UA_ByteString_deleteMembers(current);
-        return UA_STATUSCODE_GOOD;
+        /* The chunk is okay but incomplete. Store the end. */
+        if(chunk_length + complete_until > message->length)
+            break;
+
+        complete_until += chunk_length; /* Go to the next chunk */
     }
 
-    /* no complete message at all */
-    if(offset == 0) {
-        if(!*realloced) {
-            /* store the buffer in the connection */
-            UA_ByteString_copy(current, &connection->incompleteMessage);
-            connection->releaseRecvBuffer(connection, message);
-            *realloced = true;
+    /* Separate incomplete chunks */
+    if(complete_until != message->length) {
+        /* Garbage after the last good chunk. No need to keep a buffer */
+        if(garbage_end) {
+            if(complete_until == 0)
+                goto cleanup; /* All garbage, only happens on messages from the network layer */
+            message->length = complete_until;
+            return UA_STATUSCODE_GOOD;
         }
-        return UA_STATUSCODE_GOOD;
-    }
 
-    /* there remains an incomplete message at the end */
-    if(current->length != offset) {
-        UA_Byte *data = UA_malloc(current->length - offset);
-        if(!data) {
-            UA_ByteString_deleteMembers(&connection->incompleteMessage);
+        /* No good chunk, only an incomplete one */
+        if(complete_until == 0) {
             if(!*realloced) {
+                retval = UA_ByteString_allocBuffer(&connection->incompleteMessage, message->length);
+                if(retval != UA_STATUSCODE_GOOD)
+                    goto cleanup;
+                memcpy(connection->incompleteMessage.data, message->data, message->length);
                 connection->releaseRecvBuffer(connection, message);
                 *realloced = true;
+            } else {
+                connection->incompleteMessage = *message;
+                *message = UA_BYTESTRING_NULL;
             }
-            return UA_STATUSCODE_BADOUTOFMEMORY;
+            return UA_STATUSCODE_GOOD;
         }
-        size_t newlength = current->length - offset;
-        memcpy(data, &current->data[offset], newlength);
-        current->length = offset;
-        if(*realloced)
-            *message = *current;
-        connection->incompleteMessage.data = data;
-        connection->incompleteMessage.length = newlength;
-        return UA_STATUSCODE_GOOD;
-    }
 
-    if(current == &connection->incompleteMessage) {
-        *message = *current;
-        connection->incompleteMessage = UA_BYTESTRING_NULL;
+        /* At least one good chunk and an incomplete one */
+        size_t incomplete_length = message->length - complete_until;
+        retval = UA_ByteString_allocBuffer(&connection->incompleteMessage, incomplete_length);
+        if(retval != UA_STATUSCODE_GOOD)
+            goto cleanup;
+        memcpy(&connection->incompleteMessage.data, &message->data[complete_until], incomplete_length);
+        message->length = complete_until;
     }
+
     return UA_STATUSCODE_GOOD;
+
+ cleanup:
+    if(!*realloced)
+        connection->releaseRecvBuffer(connection, message);
+    UA_ByteString_deleteMembers(&connection->incompleteMessage);
+    return retval;
 }
 
 #if (__GNUC__ >= 4 && __GNUC_MINOR__ >= 6)