Support streams

Author: Qard

Cargo.toml

@@ -22,17 +22,18 @@ path = "src/main.rs"
 [dependencies]
 bytes = "1.10.1"
 hostname = "0.4.1"
-ext-php-rs = { version = "0.14.0", features = ["embed"] }
-http-handler = { git = "https://github.com/platformatic/http-handler.git", branch = "core-async-trait" }
+ext-php-rs = { path = "../ext-php-rs", features = ["embed"] }
+http-body-util = "0.1"
+http-handler = { git = "https://github.com/platformatic/http-handler.git", branch = "streams-and-ws" }
 # http-handler = { path = "../http-handler" }
-http-rewriter = { git = "https://github.com/platformatic/http-rewriter.git" }
+http-rewriter = { git = "https://github.com/platformatic/http-rewriter.git", branch = "stream-and-ws" }
 # 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, '')
+})