refactor: using length-prefixed chunking for Backwards invocations (#342)

- Adjusted the header byte manipulation in chunking functions to correctly use the first four bytes for data length.
- Modified the HTTP request streaming function to include the length-prefixed option for improved data handling.

Author: Yeuoly

internal/core/dify_invocation/real/http_request.go

@@ -57,9 +57,15 @@ func StreamResponse[T any](i *RealBackwardsInvocation, method string, path strin
 		}),
 		http_requests.HttpWriteTimeout(i.writeTimeout),
 		http_requests.HttpReadTimeout(i.readTimeout),
+		http_requests.HttpUsingLengthPrefixed(true),
 	)
 
-	response, err := http_requests.RequestAndParseStream[BaseBackwardsInvocationResponse[T]](i.client, i.difyPath(path), method, options...)
+	response, err := http_requests.RequestAndParseStream[BaseBackwardsInvocationResponse[T]](
+		i.client,
+		i.difyPath(path),
+		method,
+		options...,
+	)
 	if err != nil {
 		return nil, err
 	}

internal/utils/http_requests/http_warpper.go

@@ -95,7 +95,6 @@ func RequestAndParseStream[T any](client *http.Client, url string, method string
 	for _, option := range options {
 		if option.Type == HttpOptionTypeReadTimeout {
 			readTimeout = option.Value.(int64)
-			break
 		} else if option.Type == HttpOptionTypeRaiseErrorWhenStreamDataNotMatch {
 			raiseErrorWhenStreamDataNotMatch = option.Value.(bool)
 		} else if option.Type == HttpOptionTypeUsingLengthPrefixed {

internal/utils/parser/chunking.go

@@ -83,7 +83,7 @@ func LengthPrefixedChunking(
 		}
 
 		// decoding data length
-		dataLength := binary.LittleEndian.Uint32(header[4:8])
+		dataLength := binary.LittleEndian.Uint32(header[:4])
 		if dataLength > maxChunkSize {
 			return fmt.Errorf("data length is too long: %d", dataLength)
 		}

internal/utils/parser/chunking_test.go

@@ -137,7 +137,7 @@ func TestLengthPrefixedChunking(t *testing.T) {
 				// First 4 bytes are already used for data length placeholder
 				// Write data length in bytes 4-7 (little endian)
 				dataLen := uint32(len(chunk))
-				binary.LittleEndian.PutUint32(header[4:8], dataLen)
+				binary.LittleEndian.PutUint32(header[:4], dataLen)
 				// Remaining 6 bytes are reserved (already zero)
 
 				buf.Write(header)
@@ -247,7 +247,7 @@ func TestLengthPrefixedChunking_DataTooLarge(t *testing.T) {
 
 	// Create header with large data size
 	header := make([]byte, 10)
-	binary.LittleEndian.PutUint32(header[4:8], largeDataSize)
+	binary.LittleEndian.PutUint32(header[:4], largeDataSize)
 
 	buf.Write(header)