Merge pull request #45158 from dmarteau/fix-server-filter-streaming

[server] Allow better control of the response flow chain from server filters.

python/server/auto_generated/qgsserverfilter.sip.in

@@ -44,20 +44,32 @@ and must be passed to QgsServerFilter instances.
 Returns the :py:class:`QgsServerInterface` instance
 %End
 
-    virtual void requestReady();
+ virtual void requestReady() /Deprecated/;
 %Docstring
 Method called when the :py:class:`QgsRequestHandler` is ready and populated with
 parameters, just before entering the main switch for core services.
+
+This method is considered as deprecated and :py:func:`onRequestReady` should
+be used instead.
+
+.. deprecated::
+   Will be removed in QGIS 4.0
 %End
 
-    virtual void responseComplete();
+ virtual void responseComplete() /Deprecated/;
 %Docstring
 Method called when the :py:class:`QgsRequestHandler` processing has done and
 the response is ready, just after the main switch for core services
-and before final sending response to FCGI stdout.
+and before final sending response.
+
+This method is considered as deprecated and :py:func:`onResponseComplete` should
+be used instead.
+
+.. deprecated::
+   Will be removed in QGIS 4.0
 %End
 
-    virtual void sendResponse();
+ virtual void sendResponse() /Deprecated/;
 %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
@@ -65,8 +77,51 @@ 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`
+
+This method is considered as deprecated and :py:func:`onSendResponse` should
+be used instead.
+
+.. deprecated::
+   Will be removed in QGIS 4.0
 %End
 
+    virtual bool onRequestReady();
+%Docstring
+Method called when the :py:class:`QgsRequestHandler` is ready and populated with
+parameters, just before entering the main switch for core services.
+
+:return: true if the call must propagate to the subsequent filters, false otherwise
+
+.. versionadded:: 3.24
+%End
+
+    virtual bool onResponseComplete();
+%Docstring
+Method called when the :py:class:`QgsRequestHandler` processing has done and
+the response is ready, just after the main switch for core services
+and before final sending response to FCGI stdout.
+
+:return: true if the call must propagate to the subsequent filters, false otherwise
+
+.. versionadded:: 3.24
+%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.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`
+
+:return: true if the call must propagate to the subsequent filters, false otherwise
+
+.. versionadded:: 3.22
+%End
+
+
+
 };
 
 typedef QMultiMap<int, QgsServerFilter *> QgsServerFiltersMap;

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 )
@@ -33,32 +32,46 @@ void QgsFilterResponseDecorator::start()
   QgsServerFiltersMap::const_iterator filtersIterator;
   for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
   {
-    filtersIterator.value()->requestReady();
+    if ( ! filtersIterator.value()->onRequestReady() )
+    {
+      // stop propagation
+      return;
+    }
   }
 #endif
 }
 
 void QgsFilterResponseDecorator::finish()
 {
-
 #ifdef HAVE_SERVER_PYTHON_PLUGINS
   QgsServerFiltersMap::const_iterator filtersIterator;
   for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
   {
-    filtersIterator.value()->responseComplete();
+    if ( ! filtersIterator.value()->onResponseComplete() )
+    {
+      // stop propagation, 'finish' must be called on the wrapped
+      // response
+      break;
+    }
   }
 #endif
-  // Will call 'flush'
+  // Will call internal 'flush'
   mResponse.finish();
 }
 
+
 void QgsFilterResponseDecorator::flush()
 {
 #ifdef HAVE_SERVER_PYTHON_PLUGINS
   QgsServerFiltersMap::const_iterator filtersIterator;
+
   for ( filtersIterator = mFilters.constBegin(); filtersIterator != mFilters.constEnd(); ++filtersIterator )
   {
-    filtersIterator.value()->sendResponse();
+    if ( ! filtersIterator.value()->onSendResponse() )
+    {
+      // Stop propagation
+      return;
+    }
   }
 #endif
   mResponse.flush();

src/server/qgsserverfilter.cpp

@@ -19,6 +19,7 @@
 
 #include "qgsserverfilter.h"
 #include "qgslogger.h"
+#include "qgis.h"
 
 /**
  * QgsServerFilter
@@ -42,8 +43,33 @@ void QgsServerFilter::responseComplete()
   QgsDebugMsg( QStringLiteral( "QgsServerFilter plugin default responseComplete called" ) );
 }
 
-
 void QgsServerFilter::sendResponse()
 {
   QgsDebugMsg( QStringLiteral( "QgsServerFilter plugin default sendResponse called" ) );
 }
+
+bool QgsServerFilter::onRequestReady()
+{
+  Q_NOWARN_DEPRECATED_PUSH
+  requestReady();
+  Q_NOWARN_DEPRECATED_POP
+  return true;
+}
+
+bool QgsServerFilter::onResponseComplete()
+{
+  Q_NOWARN_DEPRECATED_PUSH
+  responseComplete();
+  Q_NOWARN_DEPRECATED_POP
+  return true;
+}
+
+bool QgsServerFilter::onSendResponse()
+{
+  Q_NOWARN_DEPRECATED_PUSH
+  sendResponse();
+  Q_NOWARN_DEPRECATED_POP
+  return true;
+}
+
+

src/server/qgsserverfilter.h

@@ -60,15 +60,25 @@ class SERVER_EXPORT QgsServerFilter
     /**
      * Method called when the QgsRequestHandler is ready and populated with
      * parameters, just before entering the main switch for core services.
+     *
+     * This method is considered as deprecated and \see onRequestReady should
+     * be used instead.
+     *
+     * \deprecated Will be removed in QGIS 4.0
      */
-    virtual void requestReady();
+    Q_DECL_DEPRECATED virtual void requestReady() SIP_DEPRECATED;
 
     /**
      * Method called when the QgsRequestHandler processing has done and
      * the response is ready, just after the main switch for core services
-     * and before final sending response to FCGI stdout.
+     * and before final sending response.
+     *
+     * This method is considered as deprecated and \see onResponseComplete should
+     * be used instead.
+     *
+     * \deprecated Will be removed in QGIS 4.0
      */
-    virtual void responseComplete();
+    Q_DECL_DEPRECATED virtual void responseComplete() SIP_DEPRECATED;
 
     /**
      * Method called when the QgsRequestHandler sends its data to FCGI stdout.
@@ -77,11 +87,52 @@ 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()
+     *
+     * This method is considered as deprecated and \see onSendResponse should
+     * be used instead.
+     *
+     * \deprecated Will be removed in QGIS 4.0
      */
-    virtual void sendResponse();
+    Q_DECL_DEPRECATED virtual void sendResponse() SIP_DEPRECATED;
+
+    /**
+     * Method called when the QgsRequestHandler is ready and populated with
+     * parameters, just before entering the main switch for core services.
+     *
+     * \return true if the call must propagate to the subsequent filters, false otherwise
+     *
+     * \since QGIS 3.24
+     */
+    virtual bool onRequestReady();
+
+    /**
+     * Method called when the QgsRequestHandler processing has done and
+     * the response is ready, just after the main switch for core services
+     * and before final sending response to FCGI stdout.
+     *
+     * \return true if the call must propagate to the subsequent filters, false otherwise
+     *
+     * \since QGIS 3.24
+     */
+    virtual bool onResponseComplete();
+
+    /**
+     * 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, 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()
+     *
+     * \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 @@ __copyright__ = 'Copyright 2017, The QGIS Project'
 
 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 @@ class TestQgsServerPlugins(QgsServerTestBase):
             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,108 @@ class TestQgsServerPlugins(QgsServerTestBase):
 
         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 streaming 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 onRequestReady(self):
+                request = self.serverInterface().requestHandler()
+                return self.propagate
+
+            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 onResponseComplete(self):
+                request = self.serverInterface().requestHandler()
+                request.appendBody(b'C')
+                return self.propagate
+
+        # Methods should be called only if filter1 propagate
+        class Filter2(QgsServerFilter):
+            def __init__(self, iface):
+                super().__init__(iface)
+                self.request_ready = False
+
+            def onRequestReady(self):
+                request = self.serverInterface().requestHandler()
+                self.request_ready = True
+                return True
+
+            def onSendResponse(self):
+                request = self.serverInterface().requestHandler()
+                request.appendBody(b'D')
+                return True
+
+            def onResponseComplete(self):
+                request = self.serverInterface().requestHandler()
+                request.appendBody(b'E')
+                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.assertFalse(filter2.request_ready)
+        self.assertEqual(body, b'ABC')
+
+        # Test with propagation
+        filter1.propagate = True
+        _, body = self._execute_request_project('?service=%s' % service0.name(), project=project)
+        self.assertTrue(filter2.request_ready)
+        self.assertEqual(body, b'ABDCE')
+
+        serverIface.setFilters({})
+        reg.unregisterService(service0.name(), service0.version())
+
 
 if __name__ == '__main__':
     unittest.main()