feat(rpc): stream ActiveSync Sync response bodies

lib/Horde/Rpc/ActiveSync.php

@@ -7,6 +7,7 @@
  * did not receive this file, see http://www.horde.org/licenses/lgpl21.
  *
  * @author   Michael J Rubinsky <mrubinsk@horde.org>
+ * @author   Torben Dannhauer <torben@dannhauer.de>
  *
  * @category Horde
  * @package  Rpc
@@ -34,6 +35,15 @@ class Horde_Rpc_ActiveSync extends Horde_Rpc
      */
     protected $_contentType = 'application/vnd.ms-sync.wbxml';
 
+    /**
+     * Whether the current request streams its response body to the client
+     * while the handler is still running (Sync command only, opt-in via the
+     * 'streaming' parameter).
+     *
+     * @var boolean
+     */
+    protected $_streaming = false;
+
     /**
      * Constructor.
      *
@@ -42,6 +52,10 @@ class Horde_Rpc_ActiveSync extends Horde_Rpc
      * @param array $params  A hash containing configuration parameters:
      *   - server: (Horde_ActiveSync) The ActiveSync server object.
      *             DEFAULT: none, REQUIRED
+     *   - streaming: (boolean) Stream Sync response bodies to the client
+     *                (chunked transfer-encoding) instead of buffering the
+     *                full response and sending it with Content-Length.
+     *                DEFAULT: false
      */
     public function __construct(Horde_Controller_Request_Http $request, array $params = [])
     {
@@ -89,8 +103,31 @@ public function getInput()
      */
     public function getResponse($request)
     {
-        ob_start(null, 1048576);
         $serverVars = $this->_request->getServerVars();
+
+        /* Stream Sync responses so WBXML bytes reach the client while the
+         * handler is still assembling messages. Some clients (Gmail Android)
+         * abort after ~30s without response body bytes and then never
+         * recover. Without a Content-Length header the webserver applies
+         * chunked transfer-encoding. All other commands keep the buffered
+         * Content-Length path (see Bug #12486 for GetAttachment). */
+        $this->_streaming = $this->_shouldStreamResponse($serverVars);
+
+        if ($this->_streaming) {
+            // Compression would buffer the body and defeat streaming.
+            @ini_set('zlib.output_compression', 0);
+            // Discard pre-existing output buffers: flushing them could leak
+            // stray bytes (whitespace, notices) ahead of the WBXML stream.
+            // Non-removable buffers return false; stop instead of looping.
+            while (ob_get_level()) {
+                if (!@ob_end_clean()) {
+                    break;
+                }
+            }
+            $this->_logger->debug('Horde_Rpc_ActiveSync: streaming response body for Sync.');
+        } else {
+            ob_start(null, 1048576);
+        }
         switch ($serverVars['REQUEST_METHOD']) {
             case 'OPTIONS':
             case 'GET':
@@ -159,7 +196,11 @@ public function getResponse($request)
                         $e->getMessage()
                     ));
                     $this->_handleError($e);
-                    header('HTTP/1.1 400 Invalid Request');
+                    // When streaming, body bytes may already be on the wire;
+                    // the status line can no longer be changed.
+                    if (!headers_sent()) {
+                        header('HTTP/1.1 400 Invalid Request');
+                    }
                     exit;
                 } catch (Horde_Exception_AuthenticationFailure $e) {
                     $this->_sendAuthenticationFailedHeaders($e);
@@ -171,7 +212,11 @@ public function getResponse($request)
                         $e->getMessage()
                     ));
                     $this->_handleError($e);
-                    header('HTTP/1.1 500');
+                    // When streaming, body bytes may already be on the wire;
+                    // the status line can no longer be changed.
+                    if (!headers_sent()) {
+                        header('HTTP/1.1 500');
+                    }
                     exit;
                 }
                 break;
@@ -196,6 +241,14 @@ public function authorize()
      */
     public function sendOutput($output)
     {
+        if ($this->_streaming) {
+            // Streaming Sync: the handler already flushed the body to the
+            // client; nothing is buffered here. Push any remaining SAPI
+            // buffer and finish the (chunked) response.
+            flush();
+            return;
+        }
+
         // Unfortunately, even though we can stream the data to the client
         // with a chunked encoding, using chunked encoding also breaks the
         // progress bar on the PDA. So we de-chunk here and just output a
@@ -219,6 +272,25 @@ public function sendOutput($output)
         }
     }
 
+    /**
+     * Should the response body be streamed to the client?
+     *
+     * Only Sync POST requests stream, and only when enabled via the
+     * 'streaming' parameter. All other commands (GetAttachment,
+     * ItemOperations, ...) keep the buffered Content-Length response.
+     *
+     * @param array $serverVars  The request's server variables.
+     *
+     * @return boolean
+     */
+    protected function _shouldStreamResponse($serverVars)
+    {
+        return !empty($this->_params['streaming'])
+            && $serverVars['REQUEST_METHOD'] == 'POST'
+            && !empty($this->_get['Cmd'])
+            && $this->_get['Cmd'] == 'Sync';
+    }
+
     /**
      * Output exception information to the logger.
      *
@@ -229,7 +301,8 @@ public function sendOutput($output)
     protected function _handleError($e)
     {
         $m = $e->getMessage();
-        $buffer = ob_get_clean();
+        // No output buffer is active when streaming.
+        $buffer = ob_get_level() ? ob_get_clean() : '';
 
         $this->_logger->err('Error in communicating with ActiveSync server: ' . $m);
         $b = new Horde_Support_Backtrace($e);

test/Unit/ActiveSyncStreamingTest.php

@@ -0,0 +1,103 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * Unit tests for the streaming Sync response path in Horde_Rpc_ActiveSync.
+ *
+ * Copyright 2026 The Horde Project (http://www.horde.org/)
+ *
+ * See the enclosed file LICENSE for license information (LGPL). If you
+ * did not receive this file, see http://www.horde.org/licenses/lgpl21.
+ *
+ * @author Torben Dannhauer <torben@dannhauer.de>
+ */
+
+namespace Horde\Rpc\Test\Unit;
+
+use Horde_ActiveSync;
+use Horde_Controller_Request_Http;
+use Horde_Rpc_ActiveSync;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use ReflectionClass;
+
+#[CoversClass(Horde_Rpc_ActiveSync::class)]
+class ActiveSyncStreamingTest extends TestCase
+{
+    public function testStreamsOnlySyncPostWithStreamingEnabled(): void
+    {
+        $rpc = $this->rpc(['streaming' => true], 'Sync');
+        $this->assertTrue($this->shouldStream($rpc, 'POST'));
+    }
+
+    public function testNoStreamingWhenDisabled(): void
+    {
+        $rpc = $this->rpc([], 'Sync');
+        $this->assertFalse($this->shouldStream($rpc, 'POST'));
+    }
+
+    public function testNoStreamingForOtherCommands(): void
+    {
+        $rpc = $this->rpc(['streaming' => true], 'GetAttachment');
+        $this->assertFalse($this->shouldStream($rpc, 'POST'));
+
+        $rpc = $this->rpc(['streaming' => true], 'ItemOperations');
+        $this->assertFalse($this->shouldStream($rpc, 'POST'));
+    }
+
+    public function testNoStreamingForNonPost(): void
+    {
+        $rpc = $this->rpc(['streaming' => true], 'Sync');
+        $this->assertFalse($this->shouldStream($rpc, 'GET'));
+    }
+
+    public function testSendOutputReturnsEarlyWhenStreaming(): void
+    {
+        $rpc = $this->rpc(['streaming' => true], 'Sync');
+
+        $ref = new ReflectionClass(Horde_Rpc_ActiveSync::class);
+        $prop = $ref->getProperty('_streaming');
+        $prop->setAccessible(true);
+        $prop->setValue($rpc, true);
+
+        // The buffered path would call ob_end_clean()/echo; streaming must
+        // touch neither an output buffer nor the passed data.
+        $this->expectOutputString('');
+        $level = ob_get_level();
+        $rpc->sendOutput('ignored');
+        $this->assertSame($level, ob_get_level());
+    }
+
+    protected function rpc(array $params, string $cmd): Horde_Rpc_ActiveSync
+    {
+        $activeSync = $this->createStub(Horde_ActiveSync::class);
+        $activeSync->method('getGetVars')->willReturn([
+            'Cmd' => $cmd,
+            'DeviceId' => 'test',
+            'DeviceType' => 'test',
+        ]);
+
+        $request = $this->createStub(Horde_Controller_Request_Http::class);
+        $request->method('getMethod')->willReturn('POST');
+        $request->method('getServerVars')->willReturn([
+            'QUERY_STRING' => 'Cmd=' . $cmd,
+            'REQUEST_METHOD' => 'POST',
+            'REQUEST_URI' => '/Microsoft-Server-ActiveSync',
+        ]);
+
+        return new Horde_Rpc_ActiveSync(
+            $request,
+            $params + ['server' => $activeSync]
+        );
+    }
+
+    protected function shouldStream(Horde_Rpc_ActiveSync $rpc, string $method): bool
+    {
+        $ref = new ReflectionClass(Horde_Rpc_ActiveSync::class);
+        $m = $ref->getMethod('_shouldStreamResponse');
+        $m->setAccessible(true);
+
+        return $m->invoke($rpc, ['REQUEST_METHOD' => $method]);
+    }
+}