[server] Allow filters to control streamed response

python/server/auto_generated/qgsserverfilter.sip.in

@@ -65,6 +65,30 @@ the :py:func:`~QgsServerFilter.responseComplete` plugin hook. For streaming serv
 getFeature requests, :py:func:`~QgsServerFilter.sendResponse` might have been called several times
 before the response is complete: in this particular case, :py:func:`~QgsServerFilter.sendResponse`
 is called once for each feature before hitting :py:func:`~QgsServerFilter.responseComplete`
+
+Note that this function is called recursively call if response is flushed
+during the course of its execution.
+
+.. deprecated:: QGIS 3.22
+
+%End
+
+    virtual bool onSendResponse();
+%Docstring
+Method called when the :py:class:`QgsRequestHandler` sends its data to FCGI stdout.
+This normally occurs at the end of core services processing just after
+the :py:func:`~QgsServerFilter.responseComplete` plugin hook. For streaming services (like WFS on
+getFeature requests, :py:func:`~QgsServerFilter.onSendResponse` might have been called several times
+before the response is complete: in this particular case, :py:func:`~QgsServerFilter.onSendResponse`
+is called once for each feature before hitting :py:func:`~QgsServerFilter.responseComplete`
+
+Propagation to subsequent filters and flushing of the response is controlled
+by the return value. Note that, the response is always flushed
+after a call to :py:func:`responseComplete`.
+
+:return: true if the call must propagate to the subsequent filters, false otherwise
+
+.. versionadded:: 3.22
 %End
 
 };

src/server/qgsfilterresponsedecorator.cpp

@@ -19,7 +19,6 @@
 
 #include "qgsconfig.h"
 #include "qgsfilterresponsedecorator.h"
-#include "qgsserverexception.h"
 
 QgsFilterResponseDecorator::QgsFilterResponseDecorator( QgsServerFiltersMap filters, QgsServerResponse &response )
   : mFilters( filters )
@@ -40,26 +39,37 @@ void QgsFilterResponseDecorator::start()
 
 void QgsFilterResponseDecorator::finish()
 {
-
 #ifdef HAVE_SERVER_PYTHON_PLUGINS
   QgsServerFiltersMap::const_iterator filtersIterator;
   for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
   {
     filtersIterator.value()->responseComplete();
   }
 #endif
-  // Will call 'flush'
+  // Will call internal 'flush'
   mResponse.finish();
 }
 
+
 void QgsFilterResponseDecorator::flush()
 {
 #ifdef HAVE_SERVER_PYTHON_PLUGINS
   QgsServerFiltersMap::const_iterator filtersIterator;
+
+  // Legacy method (deprecated)
   for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
   {
     filtersIterator.value()->sendResponse();
   }
+
+  for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
+  {
+    if ( ! filtersIterator.value()->onSendResponse() )
+    {
+      // Stop propagation
+      return;
+    }
+  }
 #endif
   mResponse.flush();
 }

src/server/qgsserverfilter.cpp

@@ -42,8 +42,14 @@ void QgsServerFilter::responseComplete()
   QgsDebugMsg( QStringLiteral( "QgsServerFilter plugin default responseComplete called" ) );
 }
 
-
 void QgsServerFilter::sendResponse()
 {
   QgsDebugMsg( QStringLiteral( "QgsServerFilter plugin default sendResponse called" ) );
 }
+
+bool QgsServerFilter::onSendResponse()
+{
+  QgsDebugMsg( QStringLiteral( "QgsServerFilter plugin default onSendResponse called" ) );
+  return true;
+}
+

src/server/qgsserverfilter.h

@@ -77,9 +77,32 @@ class SERVER_EXPORT QgsServerFilter
      * getFeature requests, sendResponse() might have been called several times
      * before the response is complete: in this particular case, sendResponse()
      * is called once for each feature before hitting responseComplete()
+     *
+     * Note that this function is called recursively call if response is flushed
+     * during the course of its execution.
+     *
+     * \deprecated since QGIS 3.22
      */
     virtual void sendResponse();
 
+    /**
+     * Method called when the QgsRequestHandler sends its data to FCGI stdout.
+     * This normally occurs at the end of core services processing just after
+     * the responseComplete() plugin hook. For streaming services (like WFS on
+     * getFeature requests, onSendResponse() might have been called several times
+     * before the response is complete: in this particular case, onSendResponse()
+     * is called once for each feature before hitting responseComplete()
+     *
+     * Propagation to subsequent filters and flushing of the response is controlled
+     * by the return value. Note that, the response is always flushed
+     * after a call to \see responseComplete().
+     *
+     * \return true if the call must propagate to the subsequent filters, false otherwise
+     *
+     * \since QGIS 3.22
+     */
+    virtual bool onSendResponse();
+
   private:
 
     QgsServerInterface *mServerInterface = nullptr;

tests/src/python/test_qgsserver_plugins.py

@@ -16,7 +16,7 @@
 
 import os
 
-from qgis.server import QgsServer
+from qgis.server import QgsServer, QgsServiceRegistry, QgsService
 from qgis.core import QgsMessageLog
 from qgis.testing import unittest
 from utilities import unitTestDataPath
@@ -65,6 +65,9 @@ def requestReady(self):
             def sendResponse(self):
                 QgsMessageLog.logMessage("SimpleHelloFilter.sendResponse")
 
+            def onSendResponse(self):
+                QgsMessageLog.logMessage("SimpleHelloFilter.onSendResponse")
+
             def responseComplete(self):
                 request = self.serverInterface().requestHandler()
                 params = request.parameterMap()
@@ -268,6 +271,87 @@ def responseComplete(self):
 
         serverIface.setFilters({})
 
+    def test_streaming_pipeline(self):
+        """ Test streaming pipeline propagation
+        """
+        try:
+            from qgis.server import QgsServerFilter
+            from qgis.core import QgsProject
+        except ImportError:
+            print("QGIS Server plugins are not compiled. Skipping test")
+            return
+
+        # create a service for streming data
+        class StreamedService(QgsService):
+
+            def __init__(self):
+                super().__init__()
+                self._response = b"Should never appear"
+                self._name = "TestStreamedService"
+                self._version = "1.0"
+
+            def name(self):
+                return self._name
+
+            def version(self):
+                return self._version
+
+            def executeRequest(self, request, response, project):
+                response.setStatusCode(206)
+                response.write(self._response)
+                response.flush()
+
+        class Filter1(QgsServerFilter):
+
+            def onSendResponse(self, ):
+                request = self.serverInterface().requestHandler()
+                request.clearBody()
+                request.appendBody(b'A')
+                request.sendResponse()
+                request.appendBody(b'B')
+                request.sendResponse()
+                # Stop propagating
+                return self.propagate
+
+            def responseComplete(self):
+                request = self.serverInterface().requestHandler()
+                request.appendBody(b'C')
+
+        class Filter2(QgsServerFilter):
+            # This should be called only if filter1 propagate
+            def onSendResponse(self):
+                request = self.serverInterface().requestHandler()
+                request.appendBody(b'D')
+                return True
+
+        serverIface = self.server.serverInterface()
+        serverIface.setFilters({})
+
+        service0 = StreamedService()
+
+        reg = serverIface.serviceRegistry()
+        reg.registerService(service0)
+
+        filter1 = Filter1(serverIface)
+        filter2 = Filter2(serverIface)
+        serverIface.registerFilter(filter1, 200)
+        serverIface.registerFilter(filter2, 300)
+
+        project = QgsProject()
+
+        # Test no propagation
+        filter1.propagate = False
+        _, body = self._execute_request_project('?service=%s' % service0.name(), project=project)
+        self.assertEqual(body, b'ABC')
+
+        # Test with propagation
+        filter1.propagate = True
+        _, body = self._execute_request_project('?service=%s' % service0.name(), project=project)
+        self.assertEqual(body, b'ABDC')
+
+        serverIface.setFilters({})
+        reg.unregisterService(service0.name(), service0.version())
+
 
 if __name__ == '__main__':
     unittest.main()