Make it possible to pass functions to change output streaming

src/main/php/web/Response.class.php

@@ -12,6 +12,7 @@
  */
 class Response {
   private $output;
+  private $streaming= null;
   private $flushed= false;
   private $status= 200;
   private $message= 'OK';
@@ -109,13 +110,30 @@ public function headers() {
     return $r;
   }
 
-  /** @param web.io.Output $output */
+  /**
+   * Begins output by sending the status line and headers
+   *
+   * @param  web.io.Output $output
+   * @return web.io.Output
+   */
   private function begin($output) {
     $output->begin($this->status, $this->message, $this->cookies
-      ? array_merge($this->headers, ['Set-Cookie' => array_map(function($c) { return $c->header(); }, $this->cookies)])
+      ? $this->headers + ['Set-Cookie' => array_map(function($c) { return $c->header(); }, $this->cookies)]
       : $this->headers
     );
     $this->flushed= true;
+    return $output;
+  }
+
+  /**
+   * Changes the implementation used inside `stream()` to determine the output.
+   *
+   * @param  function(self, ?int): web.io.Output $implementation
+   * @return self
+   */
+  public function streaming(callable $implementation) {
+    $this->streaming= $implementation;
+    return $this;
   }
 
   /**
@@ -129,7 +147,7 @@ public function flush() {
       throw new IllegalStateException('Response already flushed');
     }
 
-    $this->begin($this->output);
+    $this->begin($this->streaming ? ($this->streaming)($this, 0) : $this->output);
   }
 
   /**
@@ -152,26 +170,20 @@ public function end() {
   }
 
   /**
-   * Returns a stream to write on
+   * Returns a stream to write on: When given a length, sets the `Content-Length`
+   * header to this value and writes the subsequently given bytes unmodified to
+   * the output. Otheriwse, chunked transfer encoding is used.
    *
-   * @param  int $size If omitted, uses chunked transfer encoding
+   * @param  ?int $length
    * @return io.streams.OutputStream
    * @throws lang.IllegalStateException
    */
-  public function stream($size= null) {
+  public function stream($length= null) {
     if ($this->flushed) {
       throw new IllegalStateException('Response already flushed');
     }
 
-    if (null === $size) {
-      $output= $this->output->stream();
-    } else {
-      $this->headers['Content-Length']= [$size];
-      $output= $this->output;
-    }
-
-    $this->begin($output);
-    return $output;
+    return $this->begin($this->streaming ? ($this->streaming)($this, $length) : $this->output->stream($length));
   }
 
   /**

src/main/php/web/io/Buffered.class.php

@@ -1,14 +1,12 @@
 <?php namespace web\io;
 
-/**
- * Buffers data
- */
+/** Buffers data */
 class Buffered extends Output {
   private $target;
   private $status, $message, $headers;
   private $buffer= '';
 
-  /** @param web.io.Output $target */
+  /** @param parent $target */
   public function __construct($target) {
     $this->target= $target;
   }

src/main/php/web/io/Output.class.php

@@ -36,9 +36,10 @@ public abstract function write($bytes);
    * Returns an output used when the content-length is not known at the
    * time of starting the output.
    *
+   * @param  ?int $length
    * @return self
    */
-  public function stream() { return $this; }
+  public function stream($length= null) { return $this; }
 
   /** @return void */
   public function finish() { }
@@ -47,7 +48,7 @@ public function finish() { }
   public function flush() { }
 
   /** @return void */
-  public function close() {
+  public final function close() {
     if ($this->closed) return;
     $this->finish();
     $this->closed= true;

src/main/php/web/io/TestOutput.class.php

@@ -5,7 +5,7 @@
 /**
  * Input for testing purposes
  *
- * @test  xp://web.unittest.io.TestOutputTest
+ * @test  web.unittest.io.TestOutputTest
  */
 class TestOutput extends Output {
   private $stream;
@@ -53,8 +53,15 @@ public function begin($status, $message, $headers) {
     $this->bytes.= "\r\n";
   }
 
-  /** @return web.io.Output */
-  public function stream() { return $this->stream->newInstance($this); }
+  /**
+   * Returns writer with length if known, using the configured stream otherwise
+   *
+   * @param  ?int $length
+   * @return web.io.Output
+   */
+  public function stream($length= null) {
+    return null === $length ? $this->stream->newInstance($this) : new WriteLength($this, $length);
+  }
 
   /**
    * Writes the bytes (in this case, to the internal buffer which can be

src/main/php/web/io/WriteChunks.class.php

@@ -4,15 +4,15 @@
  * Writes Chunked transfer encoding
  *
  * @see   https://tools.ietf.org/html/rfc7230#section-4.1
- * @test  xp://web.unittest.io.WriteChunksTest
+ * @test  web.unittest.io.WriteChunksTest
  */
 class WriteChunks extends Output {
-  const BUFFER_SIZE = 4096;
+  const BUFFER_SIZE= 4096;
 
   private $target;
   private $buffer= '';
 
-  /** @param web.io.Output $target */
+  /** @param parent $target */
   public function __construct($target) {
     $this->target= $target;
   }

src/main/php/web/io/WriteLength.class.php

@@ -0,0 +1,54 @@
+<?php namespace web\io;
+
+/**
+ * Writes content with length by adding `Content-Length` header
+ *
+ * @test  web.unittest.io.WriteLengthTest
+ */
+class WriteLength extends Output {
+  private $target, $length;
+
+  /**
+   * Creates a new instance
+   *
+   * @param  parent $target
+   * @param  int $length
+   */
+  public function __construct($target, $length) {
+    $this->target= $target;
+    $this->length= $length;
+  }
+
+  /**
+   * Begins output
+   *
+   * @param  int $status
+   * @param  string $message
+   * @param  [:string] $headers
+   * @return void
+   */
+  public function begin($status, $message, $headers) {
+    $headers['Content-Length']= [$this->length];
+    $this->target->begin($status, $message, $headers);
+  }
+
+  /**
+   * Writes a chunk of data
+   *
+   * @param  string $chunk
+   * @return void
+   */
+  public function write($chunk) {
+    $this->target->write($chunk);
+  }
+
+  /** @return void */
+  public function flush() {
+    $this->target->flush();
+  }
+
+  /** @return void */
+  public function finish() {
+    $this->target->finish();
+  }
+}

src/main/php/xp/web/SAPI.class.php

@@ -1,6 +1,6 @@
 <?php namespace xp\web;
 
-use web\io\{Buffered, Input, Output, Param, ReadStream, ReadLength, WriteChunks, Incomplete};
+use web\io\{Buffered, Input, Output, Param, ReadStream, ReadLength, WriteChunks, WriteLength, Incomplete};
 
 /**
  * Wrapper for PHP's Server API ("SAPI").
@@ -144,17 +144,19 @@ public function read($length= -1) {
    * Uses chunked TE for HTTP/1.1, buffering for HTTP/1.0 or when using
    * Apache and FastCGI, which is broken.
    *
+   * @param  ?int $length
    * @return web.io.Output
    * @see    https://tools.ietf.org/html/rfc2068#section-19.7.1
    * @see    https://bz.apache.org/bugzilla/show_bug.cgi?id=53332
    */
-  public function stream() {
-    $buffered= (
+  public function stream($length= null) {
+    if (null !== $length) return new WriteLength($this, $length);
+
+    $buffered= $_SERVER['SERVER_PROTOCOL'] < 'HTTP/1.1' || (
       isset($_SERVER['GATEWAY_INTERFACE']) &&
       stristr($_SERVER['GATEWAY_INTERFACE'], 'CGI') &&
       stristr($_SERVER['SERVER_SOFTWARE'], 'Apache')
-    ) || $_SERVER['SERVER_PROTOCOL'] < 'HTTP/1.1';
-
+    );
     return $buffered ? new Buffered($this) : new WriteChunks($this);
   }
 

src/main/php/xp/web/dev/CaptureOutput.class.php

@@ -0,0 +1,71 @@
+<?php namespace xp\web\dev;
+
+use web\io\Output;
+
+class CaptureOutput extends Output {
+  public $status, $message, $headers;
+  public $bytes= '';
+  private $length= -1;
+
+  /**
+   * Sets length
+   *
+   * @param  ?int $length
+   * @return self
+   */
+  public function length($length) {
+    $this->length= $length;
+    return $this;
+  }
+
+  /**
+   * Begins output
+   *
+   * @param  int $status
+   * @param  string $message
+   * @param  [:string] $headers
+   * @return void
+   */
+  public function begin($status, $message, $headers) {
+    $this->status= $status;
+    $this->message= $message;
+    $this->headers= $headers;
+  }
+
+  /**
+   * Writes a chunk of data
+   *
+   * @param  string $chunk
+   * @return void
+   */
+  public function write($bytes) {
+    $this->bytes.= $bytes;
+  }
+
+  /**
+   * Ensure response is flushed
+   *
+   * @param  web.Response $response
+   * @return void
+   */
+  public function end($response) {
+    if (-1 === $this->length) $response->flush();
+  }
+
+  /**
+   * Drain this buffered output to a given output instance, closing it
+   * once finished.
+   *
+   * @param  web.Response $response
+   * @return void
+   */
+  public function drain($response) {
+    $out= $response->output()->stream($this->length);
+    try {
+      $out->begin($this->status, $this->message, $this->headers);
+      $out->write($this->bytes);
+    } finally {
+      $out->close();
+    }
+  }
+}