fix: httpclient - header return 100-continue and write queue with empty value

src/Stream/CurlFactory.php

@@ -34,9 +34,11 @@ public static function create(
         $contentType = null;
         $options[CURLOPT_HEADERFUNCTION] = function (?CurlHandle $curlHandle, string $header) use (&$queueHeader, &$contentType): int {
             $queueHeader->write($header);
+
             if (preg_match('/^Content-Type:\s*(.+)$/i', $header, $matches)) {
                 $contentType = strtolower(trim($matches[1]));
             }
+
             return strlen($header);
         };
 
@@ -66,17 +68,31 @@ public static function create(
             $options[CURLOPT_TIMEOUT] = $timeout;
         }
 
-        if ($request->getBody() !== null) {
+        if (is_string($request->getBody())) {
             $options[CURLOPT_POSTFIELDS] = $request->getBody();
         }
 
+        if ($request->getBody() instanceof FileUpload) {
+            $fileUpload = $request->getBody();
+
+            $options[CURLOPT_UPLOAD] = true;
+            $options[CURLOPT_RETURNTRANSFER] = true;
+            $options[CURLOPT_INFILESIZE] = $fileUpload->getSize();
+            $options[CURLOPT_READFUNCTION] = function () use ($fileUpload): string {
+                return $fileUpload->read();
+            };
+        }
+
         if ($request->hasHeader('Accept-Encoding')) {
             $options[CURLOPT_ENCODING] = implode(',', $request->getHeader('Accept-Encoding'));
         }
 
         if ($request->getMethod() === 'HEAD') {
             $options[CURLOPT_NOBODY] = true;
             unset(
+                $options[CURLOPT_UPLOAD],
+                $options[CURLOPT_INFILESIZE],
+                $options[CURLOPT_READFUNCTION],
                 $options[CURLOPT_WRITEFUNCTION],
             );
         }

src/Stream/CurlMultiHandler.php

@@ -76,6 +76,8 @@ public function execute(): void
             throw new RuntimeException('Internal HttpClient: Failed to add cURL handle to multi handle: ' . curl_multi_strerror(curl_multi_errno($multiHandle)));
         }
 
+        $infinityLoop = 0;
+
         do {
             $status = curl_multi_exec($multiHandle, $isRunning);
             if ($isRunning) {
@@ -84,6 +86,10 @@ public function execute(): void
 
             $this->verifyCurlHandle($multiHandle);
 
+            if (++$infinityLoop === 5) {
+                throw new RuntimeException('Internal HttpClient: cURL multi exec loop exceeded maximum iterations.');
+            }
+
         } while ($this->header->isEmpty() && $isRunning && $status === CURLM_OK);
 
         $this->multiHandle = $multiHandle;

src/Stream/FileUpload.php

@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thenativeweb\Eventsourcingdb\Stream;
+
+use InvalidArgumentException;
+use SplFileObject;
+
+class FileUpload
+{
+    private const SUPPORTED_CONTENT_TYPES = [
+        'application/x-ndjson',
+    ];
+
+    public function __construct(
+        private readonly SplFileObject $splFileObject,
+        private readonly string $contentType = 'application/x-ndjson',
+    ) {
+        if (!$splFileObject->isReadable()) {
+            throw new InvalidArgumentException("The file {$this->splFileObject->getRealPath()} must be readable.");
+        }
+    }
+
+    public function getContentType(): string
+    {
+        if (!in_array($this->contentType, self::SUPPORTED_CONTENT_TYPES, true)) {
+            $supportedContentTypes = implode("', '", self::SUPPORTED_CONTENT_TYPES);
+            throw new InvalidArgumentException(
+                "Unsupported content type: '{$this->contentType}', expected '{$supportedContentTypes}'."
+            );
+        }
+
+        return $this->contentType;
+    }
+
+    public function isReadable(): bool
+    {
+        return $this->splFileObject->isReadable();
+    }
+
+    public function getRealPath(): string
+    {
+        return $this->splFileObject->getRealPath();
+    }
+
+    public function getSize(): int
+    {
+        return $this->splFileObject->getSize();
+    }
+
+    public function read(): string
+    {
+        return $this->splFileObject->fgets();
+    }
+}

src/Stream/HttpClient.php

@@ -31,34 +31,53 @@ public function buildUri(string $uri): string
         return $buildUri;
     }
 
-    public function get(string $uri, ?string $apiToken = null): Response
+    public function buildHeaders(?string $apiToken, null|array|object $body = null): array
+    {
+        $headers = ['Expect:'];
+        if ($apiToken !== null) {
+            $headers[] = 'Authorization: Bearer ' . $apiToken;
+        }
+        if ($body !== null && !$body instanceof FileUpload) {
+            $headers[] = 'Content-Type: application/json';
+        }
+        if ($body instanceof FileUpload) {
+            $headers[] = 'Content-Type: ' . $body->getContentType();
+        }
+
+        return $headers;
+    }
+
+    public function buildBody(null|array|object $body): string|FileUpload
     {
-        $header = $apiToken !== null ? ['Authorization: Bearer ' . $apiToken] : [];
+        if ($body === null) {
+            return '';
+        }
 
+        if ($body instanceof FileUpload) {
+            return $body;
+        }
+
+        return json_encode($body);
+    }
+
+    public function get(string $uri, ?string $apiToken = null): Response
+    {
         $request = new Request(
             'GET',
             $this->buildUri($uri),
-            $header,
+            $this->buildHeaders($apiToken),
         );
 
         return $this->sendRequest($request);
     }
 
-    public function post(string $uri, ?string $apiToken = null, null|array|object $jsonValue = null): Response
+    public function post(string $uri, ?string $apiToken = null, null|array|object $body = null): Response
     {
-        $header = [];
-        if ($apiToken !== null) {
-            $header[] = 'Authorization: Bearer ' . $apiToken;
-        }
-        if ($jsonValue !== null) {
-            $header[] = 'Content-Type: application/json';
-        }
-
         $request = new Request(
             'POST',
             $this->buildUri($uri),
-            $header,
-            json_encode($jsonValue),
+            $this->buildHeaders($apiToken, $body),
+            $this->buildBody($body),
         );
 
         return $this->sendRequest($request);

src/Stream/Queue.php

@@ -32,7 +32,7 @@ public function read(): string
 
     public function write(string $data): void
     {
-        if ($data === '') {
+        if (trim($data) === '') {
             return;
         }
 

src/Stream/Request.php

@@ -14,7 +14,7 @@ public function __construct(
         private readonly string $method,
         string $uri,
         array $headers = [],
-        private readonly ?string $body = null,
+        private readonly null|string|FileUpload $body = null,
         private readonly string $protocolVersion = '1.1'
     ) {
         $this->uri = new Uri($uri);
@@ -36,7 +36,7 @@ public function getProtocolVersion(): string
         return $this->protocolVersion;
     }
 
-    public function getBody(): ?string
+    public function getBody(): null|string|FileUpload
     {
         return $this->body;
     }

tests/Stream/CurlFactoryTest.php

@@ -7,6 +7,7 @@
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
 use Thenativeweb\Eventsourcingdb\Stream\CurlFactory;
+use Thenativeweb\Eventsourcingdb\Stream\FileUpload;
 use Thenativeweb\Eventsourcingdb\Stream\Queue;
 use Thenativeweb\Eventsourcingdb\Stream\Request;
 use Thenativeweb\Eventsourcingdb\Stream\Uri;
@@ -121,6 +122,38 @@ public function testCreateSetsPostFieldsIfBodyExists(): void
         $this->assertSame($body, $options[CURLOPT_POSTFIELDS]);
     }
 
+    public function testCreateSetsReadFunctionIfBodyFileUpload(): void
+    {
+        $fileUpload = $this->createMock(FileUpload::class);
+        $fileUpload->method('getSize')->willReturn(123);
+        $fileUpload->method('read')->willReturn('chunk');
+
+        $this->uriMock->method('__toString')->willReturn('https://example.com/upload');
+        $this->uriMock->method('getScheme')->willReturn('https');
+
+        $this->requestMock->method('getMethod')->willReturn('POST');
+        $this->requestMock->method('getProtocolVersion')->willReturn('1.1');
+        $this->requestMock->method('getHeaders')->willReturn([]);
+        $this->requestMock->method('getUri')->willReturn($this->uriMock);
+        $this->requestMock->method('getBody')->willReturn($fileUpload);
+        $this->requestMock->method('hasHeader')->willReturn(false);
+
+        $options = CurlFactory::create(
+            $this->requestMock,
+            $this->headerQueueMock,
+            $this->writeQueueMock,
+        );
+
+        $this->assertArrayHasKey(CURLOPT_INFILESIZE, $options);
+        $this->assertArrayHasKey(CURLOPT_READFUNCTION, $options);
+        $this->assertArrayHasKey(CURLOPT_RETURNTRANSFER, $options);
+        $this->assertArrayHasKey(CURLOPT_UPLOAD, $options);
+        $this->assertSame(123, $options[CURLOPT_INFILESIZE]);
+        $this->assertIsCallable($options[CURLOPT_READFUNCTION]);
+        $this->assertTrue($options[CURLOPT_RETURNTRANSFER]);
+        $this->assertTrue($options[CURLOPT_UPLOAD]);
+    }
+
     public function testCreateSetsNoBodyForHeadMethod(): void
     {
         $this->requestMock->method('getMethod')->willReturnCallback(static fn (): string => 'HEAD');

tests/Stream/FileUploadTest.php

@@ -0,0 +1,91 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Stream;
+
+use InvalidArgumentException;
+use PHPUnit\Framework\TestCase;
+use SplFileObject;
+use Thenativeweb\Eventsourcingdb\Stream\FileUpload;
+
+final class FileUploadTest extends TestCase
+{
+    private string $filePath;
+
+    protected function setUp(): void
+    {
+        $this->filePath = tempnam(sys_get_temp_dir(), 'upload_');
+        file_put_contents($this->filePath, "line1\nline2\n");
+    }
+
+    protected function tearDown(): void
+    {
+        if (file_exists($this->filePath)) {
+            unlink($this->filePath);
+        }
+    }
+
+    public function testItConstructsSuccessfully(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+        $fileUpload = new FileUpload($file);
+
+        $this->assertInstanceOf(FileUpload::class, $fileUpload);
+        $this->assertTrue($fileUpload->isReadable());
+    }
+
+    public function testItThrowsWhenFileIsNotReadable(): void
+    {
+        $file = new SplFileObject('php://memory', 'r');
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('must be readable.');
+
+        new FileUpload($file);
+    }
+
+    public function testItReturnsValidContentType(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+        $fileUpload = new FileUpload($file, 'application/x-ndjson');
+
+        $this->assertSame('application/x-ndjson', $fileUpload->getContentType());
+    }
+
+    public function testItThrowsOnUnsupportedContentType(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Unsupported content type');
+
+        $fileUpload = new FileUpload($file, 'text/plain');
+        $fileUpload->getContentType();
+    }
+
+    public function testItReadsFirstLine(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+        $fileUpload = new FileUpload($file);
+
+        $line = $fileUpload->read();
+        $this->assertSame("line1\n", $line);
+    }
+
+    public function testItReturnsFileSize(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+        $fileUpload = new FileUpload($file);
+
+        $this->assertSame(filesize($this->filePath), $fileUpload->getSize());
+    }
+
+    public function testItReturnsRealPath(): void
+    {
+        $file = new SplFileObject($this->filePath, 'r');
+        $fileUpload = new FileUpload($file);
+
+        $this->assertSame(realpath($this->filePath), $fileUpload->getRealPath());
+    }
+}