[evolution-ews/wip/mcrha/soup3] Make sure GInputStream received from libsoup3 is consumed in the same thread

[Milan Crha <mcrha src gnome org>]

commit e0559785f8d6903991d68ee3525e436b93482c12
Author: Milan Crha <mcrha redhat com>
Date:   Thu May 5 18:32:00 2022 +0200

    Make sure GInputStream received from libsoup3 is consumed in the same thread

 src/EWS/common/e-ews-connection.c | 154 +++++++++++++++++++++-----------------
 1 file changed, 84 insertions(+), 70 deletions(-)
---
diff --git a/src/EWS/common/e-ews-connection.c b/src/EWS/common/e-ews-connection.c
index 690831b2..e96954f7 100644
--- a/src/EWS/common/e-ews-connection.c
+++ b/src/EWS/common/e-ews-connection.c
@@ -246,15 +246,48 @@ e_ews_soup_worker_thread (gpointer user_data)
        return NULL;
 }
 
+static gboolean
+ews_connection_credentials_failed (EEwsConnection *connection,
+                                  SoupMessage *message,
+                                  gboolean can_emit,
+                                  GError **error)
+{
+       gint expire_in_days = 0;
+       gboolean expired = FALSE;
+       gchar *service_url = NULL;
+
+       g_return_val_if_fail (E_IS_EWS_CONNECTION (connection), FALSE);
+
+       if (!message)
+               return FALSE;
+
+       if (!e_ews_connection_utils_check_x_ms_credential_headers (message, &expire_in_days, &expired, 
&service_url))
+               return FALSE;
+
+       if (can_emit) {
+               if (expired) {
+                       e_ews_connection_utils_expired_password_to_error (service_url, error);
+               } else if (expire_in_days > 0) {
+                       g_signal_emit (connection, signals[PASSWORD_WILL_EXPIRE], 0, expire_in_days, 
service_url);
+               }
+       }
+
+       g_free (service_url);
+
+       return expired;
+}
+
 typedef struct _ProcessData {
        GMutex mutex;
        GCond cond;
        gboolean done;
+       gboolean repeat;
        EEwsConnection *cnc;
+       ESoapRequest *request;
+       ESoapResponse *response;
        SoupMessage *message;
        gpointer prepare_data;
        GCancellable *cancellable;
-       GInputStream *input_stream;
        GError *error;
        gchar **out_certificate_pem;
        GTlsCertificateFlags *out_certificate_errors;
@@ -266,12 +299,43 @@ e_ews_connection_process_request_ready_cb (GObject *source_object,
                                           gpointer user_data)
 {
        ProcessData *pd = user_data;
+       GInputStream *input_stream;
 
        g_mutex_lock (&pd->mutex);
-       pd->input_stream = e_soup_session_send_message_finish (E_SOUP_SESSION (source_object), result, 
pd->out_certificate_pem, pd->out_certificate_errors, &pd->error);
+
+       input_stream = e_soup_session_send_message_finish (E_SOUP_SESSION (source_object), result, 
pd->out_certificate_pem, pd->out_certificate_errors, &pd->error);
+
+       /* Need to process the 'input_stream' in this thread */
+       if (!ews_connection_credentials_failed (pd->cnc, pd->message, FALSE, NULL) &&
+           soup_message_get_status (pd->message) != SOUP_STATUS_UNAUTHORIZED) {
+               ESoapRequestCustomProcessFn custom_process_fn = NULL;
+               gpointer custom_process_data = NULL;
+
+               e_soap_request_get_custom_process_fn (pd->request, &custom_process_fn, &custom_process_data);
+
+               if (custom_process_fn) {
+                       custom_process_fn (pd->request, pd->message, input_stream, custom_process_data, 
&pd->repeat, pd->cancellable, &pd->error);
+               } else {
+                       pd->response = e_soap_response_new ();
+
+                       e_soap_request_setup_response (pd->request, pd->response);
+
+                       if (!e_soap_response_from_message_sync (pd->response, pd->message, input_stream, 
pd->cancellable, &pd->error)) {
+                               g_clear_object (&pd->response);
+
+                               if (!pd->error) {
+                                       g_set_error (&pd->error, EWS_CONNECTION_ERROR, 
EWS_CONNECTION_ERROR_NORESPONSE,
+                                               _("No response: %s"), soup_message_get_reason_phrase 
(pd->message));
+                               }
+                       }
+               }
+       }
+
        pd->done = TRUE;
        g_cond_signal (&pd->cond);
        g_mutex_unlock (&pd->mutex);
+
+       g_clear_object (&input_stream);
 }
 
 static gboolean
@@ -293,12 +357,13 @@ e_ews_connection_process_request_run_cb (gpointer user_data)
        return FALSE;
 }
 
-static GInputStream *
+static ESoapResponse *
 e_ews_connection_process_request_sync (EEwsConnection *cnc,
                                       ESoapRequest *request,
                                       SoupMessage **out_message,
                                       gchar **out_certificate_pem,
                                       GTlsCertificateFlags *out_certificate_errors,
+                                      gboolean *out_repeat,
                                       GCancellable *cancellable,
                                       GError **error)
 {
@@ -309,8 +374,10 @@ e_ews_connection_process_request_sync (EEwsConnection *cnc,
 
        g_return_val_if_fail (E_IS_SOAP_REQUEST (request), NULL);
        g_return_val_if_fail (out_message != NULL, NULL);
+       g_return_val_if_fail (out_repeat != NULL, NULL);
 
        *out_message = NULL;
+       *out_repeat = FALSE;
 
        settings = e_ews_connection_ref_settings (cnc);
 
@@ -326,6 +393,8 @@ e_ews_connection_process_request_sync (EEwsConnection *cnc,
        }
        g_mutex_unlock (&cnc->priv->soup.mutex);
 
+       pd.request = request;
+       pd.response = NULL;
        pd.message = e_soap_request_persist (request, E_SOUP_SESSION (cnc->priv->soup.session), settings, 
error);
 
        g_clear_object (&settings);
@@ -356,9 +425,9 @@ e_ews_connection_process_request_sync (EEwsConnection *cnc,
        g_mutex_init (&pd.mutex);
        g_cond_init (&pd.cond);
        pd.done = FALSE;
+       pd.repeat = FALSE;
        pd.cnc = cnc;
        pd.cancellable = cancellable;
-       pd.input_stream = NULL;
        pd.error = NULL;
        pd.out_certificate_pem = out_certificate_pem;
        pd.out_certificate_errors = out_certificate_errors;
@@ -387,48 +456,17 @@ e_ews_connection_process_request_sync (EEwsConnection *cnc,
                }
        }
 
-       if (pd.input_stream)
-               *out_message = g_steal_pointer (&pd.message);
+       *out_message = g_steal_pointer (&pd.message);
+       *out_repeat = pd.repeat;
 
        g_clear_object (&pd.message);
        g_mutex_clear (&pd.mutex);
        g_cond_clear (&pd.cond);
 
-       g_warn_if_fail ((pd.error == NULL && pd.input_stream != NULL) ||
-                       (pd.error != NULL && pd.input_stream == NULL));
-
        if (pd.error)
                g_propagate_error (error, pd.error);
 
-       return pd.input_stream;
-}
-
-static gboolean
-ews_connection_credentials_failed (EEwsConnection *connection,
-                                  SoupMessage *message,
-                                  GError **error)
-{
-       gint expire_in_days = 0;
-       gboolean expired = FALSE;
-       gchar *service_url = NULL;
-
-       g_return_val_if_fail (E_IS_EWS_CONNECTION (connection), FALSE);
-
-       if (!message)
-               return FALSE;
-
-       if (!e_ews_connection_utils_check_x_ms_credential_headers (message, &expire_in_days, &expired, 
&service_url))
-               return FALSE;
-
-       if (expired) {
-               e_ews_connection_utils_expired_password_to_error (service_url, error);
-       } else if (expire_in_days > 0) {
-               g_signal_emit (connection, signals[PASSWORD_WILL_EXPIRE], 0, expire_in_days, service_url);
-       }
-
-       g_free (service_url);
-
-       return expired;
+       return pd.response;
 }
 
 static void
@@ -534,7 +572,6 @@ e_ews_connection_send_request_sync (EEwsConnection *cnc,
 {
        ESoapResponse *response = NULL;
        SoupMessage *message = NULL;
-       GInputStream *input_stream;
        gchar *certificate_pem = NULL;
        GTlsCertificateFlags certificate_errors = 0;
        gboolean retrying_after_io_error = FALSE;
@@ -548,7 +585,7 @@ e_ews_connection_send_request_sync (EEwsConnection *cnc,
 
                g_clear_error (&local_error);
 
-               input_stream = e_ews_connection_process_request_sync (cnc, request, &message, 
&certificate_pem, &certificate_errors, cancellable, &local_error);
+               response = e_ews_connection_process_request_sync (cnc, request, &message, &certificate_pem, 
&certificate_errors, &repeat, cancellable, &local_error);
 
                g_mutex_lock (&cnc->priv->property_lock);
                g_clear_pointer (&cnc->priv->ssl_certificate_pem, g_free);
@@ -559,7 +596,7 @@ e_ews_connection_send_request_sync (EEwsConnection *cnc,
 
                e_soap_request_take_tls_error_details (request, certificate_pem, certificate_errors);
 
-               if (message && !ews_connection_credentials_failed (cnc, message, &local_error2)) {
+               if (message && !ews_connection_credentials_failed (cnc, message, TRUE, &local_error2)) {
                        if (soup_message_get_status (message) == SOUP_STATUS_UNAUTHORIZED) {
                                /* This can happen when the server terminated the connection before the 
request started */
                                if (!retrying_after_io_error &&
@@ -594,35 +631,11 @@ e_ews_connection_send_request_sync (EEwsConnection *cnc,
                        local_error = local_error2;
                }
 
-               if (!local_error) {
-                       ESoapRequestCustomProcessFn custom_process_fn = NULL;
-                       gpointer custom_process_data = NULL;
-
-                       e_soap_request_get_custom_process_fn (request, &custom_process_fn, 
&custom_process_data);
-
-                       if (custom_process_fn) {
-                               custom_process_fn (request, message, input_stream, custom_process_data, 
&repeat, cancellable, &local_error);
-                       } else {
-                               response = e_soap_response_new ();
-
-                               e_soap_request_setup_response (request, response);
-
-                               if (!e_soap_response_from_message_sync (response, message, input_stream, 
cancellable, &local_error)) {
-                                       g_clear_object (&response);
-
-                                       if (!local_error) {
-                                               g_set_error (&local_error, EWS_CONNECTION_ERROR, 
EWS_CONNECTION_ERROR_NORESPONSE,
-                                                       _("No response: %s"), soup_message_get_reason_phrase 
(message));
-                                       }
-                               }
-
-                               if (response) {
-                                       repeat = e_ews_connection_handle_backoff_policy (cnc, response, 
cancellable, &local_error);
+               if (!local_error && response && !repeat) {
+                       repeat = e_ews_connection_handle_backoff_policy (cnc, response, cancellable, 
&local_error);
 
-                                       if (repeat || local_error)
-                                               g_clear_object (&response);
-                               }
-                       }
+                       if (repeat || local_error)
+                               g_clear_object (&response);
                }
 
                if (!repeat && !retrying_after_io_error && local_error &&
@@ -636,7 +649,8 @@ e_ews_connection_send_request_sync (EEwsConnection *cnc,
                        repeat = !g_cancellable_set_error_if_cancelled (cancellable, &local_error);
                }
 
-               g_clear_object (&input_stream);
+               if (repeat)
+                       g_clear_object (&response);
                g_clear_object (&message);
        }