Streams and ws

.github/workflows/CI.yml

@@ -45,16 +45,6 @@ jobs:
               brew install autoconf automake libtool re2c bison libiconv \
               argon2 libzip postgresql@16
 
-          # TODO: Do we need to care about x86_64 macOS?
-          # NOTE: Unable to force link bison on macOS 13, which php-src requires.
-          - host: macos-13
-            target: x86_64-apple-darwin
-            # build: pnpm build --target x86_64-apple-darwin
-            build: pnpm build
-            setup: |
-              brew install autoconf automake libtool re2c bison libiconv \
-              argon2 libzip postgresql@16
-
           #
           # Linux
           #
@@ -302,11 +292,6 @@ jobs:
       fail-fast: false
       matrix:
         settings:
-          - host: macos-13
-            target: x86_64-apple-darwin
-            architecture: x64
-            setup: |
-              brew install openssl@3 argon2 postgresql@16
           - host: macos-15
             target: aarch64-apple-darwin
             architecture: arm64
@@ -353,7 +338,20 @@ jobs:
         run: ls -R ./npm
         shell: bash
       - name: Test bindings
-        run: pnpm test
+        run: |
+          # Run tests with LLDB to catch segfaults on macOS
+          # Use the ava entrypoint directly instead of the bin script
+          lldb -b \
+            -o "settings set auto-confirm true" \
+            -o "process launch" \
+            -o "bt all" \
+            -- node node_modules/ava/entrypoints/cli.mjs __test__/handler.spec.mjs __test__/headers.spec.mjs __test__/request.spec.mjs __test__/response.spec.mjs __test__/rewriter.spec.mjs __test__/streaming.spec.mjs
+
+          EXIT_CODE=$?
+          if [ $EXIT_CODE -ne 0 ]; then
+            echo "=== Test failed with exit code $EXIT_CODE ==="
+            exit $EXIT_CODE
+          fi
 
   test-linux-binding:
     name: Test bindings on ${{ matrix.target }} - node@${{ matrix.node }}
@@ -431,9 +429,34 @@ jobs:
               libcurl4-openssl-dev autoconf libxml2-dev libsqlite3-dev \
               bison re2c libonig-dev libargon2-dev libzip-dev zlib1g-dev \
               openssh-client libclang-dev libreadline-dev libpng-dev \
-              libjpeg-dev libsodium-dev libpq5
-
-            npm run test
+              libjpeg-dev libsodium-dev libpq5 gdb
+
+            # Enable core dumps
+            ulimit -c unlimited
+            echo '/tmp/core.%e.%p' > /proc/sys/kernel/core_pattern || true
+
+            # Run tests with GDB to catch segfaults
+            # Use the ava entrypoint directly instead of the bin script
+            gdb -batch \
+              -ex "set pagination off" \
+              -ex "set confirm off" \
+              -ex "handle SIGSEGV stop print" \
+              -ex "run" \
+              -ex "thread apply all bt full" \
+              --args node node_modules/ava/entrypoints/cli.mjs __test__/handler.spec.mjs __test__/headers.spec.mjs __test__/request.spec.mjs __test__/response.spec.mjs __test__/rewriter.spec.mjs __test__/streaming.spec.mjs
+
+            EXIT_CODE=$?
+            if [ $EXIT_CODE -ne 0 ]; then
+              echo "=== Test failed with exit code $EXIT_CODE ==="
+              # Try to find and analyze core dumps
+              if ls /tmp/core.* 2>/dev/null; then
+                for core in /tmp/core.*; do
+                  echo "=== Analyzing core dump: $core ==="
+                  gdb -batch -ex "bt full" -ex "thread apply all bt" node "$core" || true
+                done
+              fi
+              exit $EXIT_CODE
+            fi
 
   publish:
     name: Publish a release

Cargo.toml

@@ -20,20 +20,20 @@ name = "php-main"
 path = "src/main.rs"
 
 [dependencies]
-async-trait = "0.1.88"
 bytes = "1.10.1"
 hostname = "0.4.1"
-ext-php-rs = { version = "0.14.0", features = ["embed"] }
+ext-php-rs = { version = "0.15.2", features = ["embed"] }
+http-body-util = "0.1"
 http-handler = { git = "https://github.com/platformatic/http-handler.git" }
 # http-handler = { path = "../http-handler" }
 http-rewriter = { git = "https://github.com/platformatic/http-rewriter.git" }
 # http-rewriter = { path = "../http-rewriter" }
 libc = "0.2.171"
 # Default enable napi4 feature, see https://nodejs.org/api/n-api.html#node-api-version-matrix
-napi = { version = "3", default-features = false, features = ["napi4"], optional = true }
+napi = { version = "3", default-features = false, features = ["napi4", "tokio_rt", "async"], optional = true }
 napi-derive = { version = "3", optional = true }
 once_cell = "1.21.0"
-tokio = { version = "1.45", features = ["rt", "macros", "rt-multi-thread"] }
+tokio = { version = "1.45", features = ["rt", "macros", "rt-multi-thread", "sync"] }
 regex = "1.0"
 
 [dev-dependencies]

__test__/request.spec.mjs

@@ -10,8 +10,6 @@ test('minimum construction requirements', (t) => {
 
   t.is(req.method, 'GET')
   t.is(req.url, 'http://example.com/test.php')
-  t.assert(req.body instanceof Buffer)
-  t.is(req.body.length, 0)
   t.assert(req.headers instanceof Headers)
   t.is(req.headers.size, 0)
 })

__test__/response.spec.mjs

@@ -9,8 +9,6 @@ test('Minimal response construction', (t) => {
 
   t.is(res.status, 200)
   t.assert(res.headers instanceof Headers)
-  t.assert(res.body instanceof Buffer)
-  t.deepEqual(res.body.toString(), '')
   t.assert(res.log instanceof Buffer)
   t.deepEqual(res.log.toString(), '')
   t.is(res.exception, null)
@@ -37,8 +35,6 @@ test('Full Response construction', (t) => {
   t.assert(res.headers instanceof Headers)
   t.deepEqual(res.headers.get('Content-Type'), 'application/json')
   t.deepEqual(res.headers.getAll('Accept'), ['application/json', 'text/plain'])
-  t.assert(res.body instanceof Buffer)
-  t.deepEqual(res.body.toString(), json)
   t.assert(res.log instanceof Buffer)
   t.deepEqual(res.log.toString(), 'Hello, from error_log!')
   t.deepEqual(res.exception, 'Hello, from PHP!')

__test__/streaming.spec.mjs

@@ -0,0 +1,251 @@
+import test from 'ava'
+
+import { Php, Request } from '../index.js'
+import { MockRoot } from './util.mjs'
+
+test('handleStream - basic response', async (t) => {
+  const mockroot = await MockRoot.from({
+    'index.php': `<?php
+      echo 'Hello, from PHP!';
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'GET',
+    url: 'http://example.com/index.php'
+  })
+
+  const [res] = await Promise.all([
+    php.handleStream(req),
+    req.end()
+  ])
+
+  t.is(res.status, 200)
+
+  // Collect streaming body
+  let body = ''
+  for await (const chunk of res) {
+    body += chunk.toString('utf8')
+  }
+  t.is(body, 'Hello, from PHP!')
+})
+
+test('handleStream - chunked output', async (t) => {
+  const mockroot = await MockRoot.from({
+    'stream.php': `<?php
+      echo 'Chunk 1';
+      flush();
+      echo 'Chunk 2';
+      flush();
+      echo 'Chunk 3';
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'GET',
+    url: 'http://example.com/stream.php'
+  })
+
+  const [res] = await Promise.all([
+    php.handleStream(req),
+    req.end()
+  ])
+
+  t.is(res.status, 200)
+
+  // Collect all chunks
+  const chunks = []
+  for await (const chunk of res) {
+    chunks.push(chunk.toString('utf8'))
+  }
+
+  // Should have received all chunks
+  const body = chunks.join('')
+  t.is(body, 'Chunk 1Chunk 2Chunk 3')
+})
+
+test('handleStream - headers available immediately', async (t) => {
+  const mockroot = await MockRoot.from({
+    'headers.php': `<?php
+      header('X-Custom-Header: test-value');
+      header('Content-Type: application/json');
+      echo '{"status": "ok"}';
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'GET',
+    url: 'http://example.com/headers.php'
+  })
+
+  const [res] = await Promise.all([
+    php.handleStream(req),
+    req.end()
+  ])
+
+  // Headers should be available immediately
+  t.is(res.status, 200)
+  t.is(res.headers.get('x-custom-header'), 'test-value')
+  t.is(res.headers.get('content-type'), 'application/json')
+
+  // Body can be consumed after
+  let body = ''
+  for await (const chunk of res) {
+    body += chunk.toString('utf8')
+  }
+  t.is(body, '{"status": "ok"}')
+})
+
+test('handleStream - POST with buffered body', async (t) => {
+  const mockroot = await MockRoot.from({
+    'echo.php': `<?php
+      $input = file_get_contents('php://input');
+      echo "Received: " . $input;
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'POST',
+    url: 'http://example.com/echo.php',
+    headers: {
+      'Content-Type': 'text/plain'
+    },
+    body: Buffer.from('Hello from client!')
+  })
+
+  const res = await php.handleStream(req)
+  t.is(res.status, 200)
+
+  let body = ''
+  for await (const chunk of res) {
+    body += chunk.toString('utf8')
+  }
+  t.is(body, 'Received: Hello from client!')
+})
+
+test('handleStream - POST with streamed body', async (t) => {
+  const mockroot = await MockRoot.from({
+    'echo.php': `<?php
+      $input = file_get_contents('php://input');
+      echo "Received: " . $input;
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'POST',
+    url: 'http://example.com/echo.php',
+    headers: {
+      'Content-Type': 'text/plain'
+    }
+  })
+
+  // Run handleStream and writes concurrently using Promise.all
+  const [res] = await Promise.all([
+    php.handleStream(req),
+    (async () => {
+      await req.write('Hello ')
+      await req.write('from ')
+      await req.write('streaming!')
+      await req.end()
+    })()
+  ])
+
+  t.is(res.status, 200)
+
+  let body = ''
+  for await (const chunk of res) {
+    body += chunk.toString('utf8')
+  }
+  t.is(body, 'Received: Hello from streaming!')
+})
+
+test.skip('handleStream - exception handling', async (t) => {
+  // TODO: Implement proper exception handling in streaming mode
+  // See EXCEPTIONS.md for implementation approaches
+  const mockroot = await MockRoot.from({
+    'error.php': `<?php
+      throw new Exception('Test exception');
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'GET',
+    url: 'http://example.com/error.php'
+  })
+
+  const res = await php.handleStream(req)
+
+  // Exception should be sent through the stream
+  let errorOccurred = false
+  try {
+    for await (const chunk of res) {
+      // Should not receive chunks, should throw
+    }
+  } catch (err) {
+    errorOccurred = true
+    t.true(err.message.includes('Exception'))
+  }
+
+  t.true(errorOccurred, 'Exception should be thrown during iteration')
+})
+
+test('handleStream - empty response', async (t) => {
+  const mockroot = await MockRoot.from({
+    'empty.php': `<?php
+      // No output
+    ?>`
+  })
+  t.teardown(() => mockroot.clean())
+
+  const php = new Php({
+    docroot: mockroot.path
+  })
+
+  const req = new Request({
+    method: 'GET',
+    url: 'http://example.com/empty.php'
+  })
+
+  const [res] = await Promise.all([
+    php.handleStream(req),
+    req.end()
+  ])
+
+  t.is(res.status, 200)
+
+  let body = ''
+  for await (const chunk of res) {
+    body += chunk.toString('utf8')
+  }
+  t.is(body, '')
+})