[evolution-data-server/wip/mcrha/soup3] e-soup-session: Create new request_body input stream on message restart

[Milan Crha <mcrha src gnome org>]

commit 690ede5ac50ae58c32c643d1e47b779ea849c00c
Author: Milan Crha <mcrha redhat com>
Date:   Thu Apr 21 16:50:29 2022 +0200

    e-soup-session: Create new request_body input stream on message restart
    
    to avoid problem with stream closing on the libsoup side.

 src/libedataserver/e-soup-session.c | 142 ++++++++++++++++++++++++++++++++----
 src/libedataserver/e-soup-session.h |   2 +-
 2 files changed, 127 insertions(+), 17 deletions(-)
---
diff --git a/src/libedataserver/e-soup-session.c b/src/libedataserver/e-soup-session.c
index 286140eb4..cb774970c 100644
--- a/src/libedataserver/e-soup-session.c
+++ b/src/libedataserver/e-soup-session.c
@@ -305,7 +305,7 @@ e_soup_session_maybe_prepare_auth (ESoupSession *session,
        g_mutex_unlock (&session->priv->property_lock);
 
        /* Provide credentials beforehand only on secure connections */
-       if (!strcmp(g_uri_get_scheme (g_uri), "https")) {
+       if (g_strcmp0 (g_uri_get_scheme (g_uri), "https") == 0) {
                if (g_strcmp0 (auth_method, "OAuth2") == 0 ||
                    e_oauth2_services_is_oauth2_alias_static (auth_method)) {
                        success = e_soup_session_maybe_prepare_bearer_auth (session, g_uri, message, 
cancellable, error);
@@ -1604,10 +1604,120 @@ e_soup_session_util_normalize_uri_path (GUri *uri)
        return nuri;
 }
 
+typedef struct _EInputStreamWrapper {
+       GInputStream parent;
+
+       GInputStream *input_stream;
+       goffset read_from;
+} EInputStreamWrapper;
+
+typedef struct _EInputStreamWrapperClass {
+       GInputStreamClass parent_class;
+} EInputStreamWrapperClass;
+
+GType e_input_stream_wrapper_get_type (void);
+
+G_DEFINE_TYPE (EInputStreamWrapper, e_input_stream_wrapper, G_TYPE_INPUT_STREAM)
+
+#define E_INPUT_STREAM_WRAPPER(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), e_input_stream_wrapper_get_type (), 
EInputStreamWrapper))
+
+static gssize
+e_input_stream_wrapper_read_fn (GInputStream *stream,
+                                void *buffer,
+                               gsize count,
+                               GCancellable *cancellable,
+                               GError **error)
+{
+       EInputStreamWrapper *wrapper = E_INPUT_STREAM_WRAPPER (stream);
+
+       return g_input_stream_read (wrapper->input_stream, buffer, count, cancellable, error);
+}
+
+static gssize
+e_input_stream_wrapper_skip (GInputStream *stream,
+                            gsize count,
+                            GCancellable *cancellable,
+                            GError **error)
+{
+       EInputStreamWrapper *wrapper = E_INPUT_STREAM_WRAPPER (stream);
+
+       return g_input_stream_skip (wrapper->input_stream, count, cancellable, error);
+}
+
+static gboolean
+e_input_stream_wrapper_close_fn (GInputStream *stream,
+                                GCancellable *cancellable,
+                                GError **error)
+{
+       /* Always success, but without closing the self::input_stream */
+       return TRUE;
+}
+
+static void
+e_input_stream_wrapper_dispose (GObject *object)
+{
+       EInputStreamWrapper *wrapper = E_INPUT_STREAM_WRAPPER (object);
+
+       g_clear_object (&wrapper->input_stream);
+
+       /* Chain up to parent's method. */
+       G_OBJECT_CLASS (e_input_stream_wrapper_parent_class)->dispose (object);
+}
+
+static void
+e_input_stream_wrapper_class_init (EInputStreamWrapperClass *klass)
+{
+       GInputStreamClass *input_stream_class;
+       GObjectClass *object_class;
+
+       input_stream_class = G_INPUT_STREAM_CLASS (klass);
+       input_stream_class->read_fn = e_input_stream_wrapper_read_fn;
+       input_stream_class->skip = e_input_stream_wrapper_skip;
+       input_stream_class->close_fn = e_input_stream_wrapper_close_fn;
+
+       object_class = G_OBJECT_CLASS (klass);
+       object_class->dispose = e_input_stream_wrapper_dispose;
+}
+
+static void
+e_input_stream_wrapper_init (EInputStreamWrapper *self)
+{
+}
+
+static void
+e_input_stream_wrapper_assign (EInputStreamWrapper *self,
+                              GInputStream *input_stream)
+{
+       self->input_stream = g_object_ref (input_stream);
+
+       if (G_IS_SEEKABLE (input_stream))
+               self->read_from = g_seekable_tell (G_SEEKABLE (input_stream));
+}
+
+static void
+e_input_stream_wrapper_rewind (EInputStreamWrapper *self)
+{
+       if (G_IS_SEEKABLE (self->input_stream) && self->read_from != g_seekable_tell (G_SEEKABLE 
(self->input_stream)))
+               g_seekable_seek (G_SEEKABLE (self->input_stream), self->read_from, G_SEEK_SET, NULL, NULL);
+}
+
+static GInputStream *
+e_input_stream_wrapper_dup (EInputStreamWrapper *self)
+{
+       EInputStreamWrapper *dup;
+
+       e_input_stream_wrapper_rewind (self);
+
+       dup = g_object_new (e_input_stream_wrapper_get_type (), NULL);
+       dup->input_stream = g_object_ref (self->input_stream);
+       dup->read_from = self->read_from;
+
+       return G_INPUT_STREAM (dup);
+}
+
 typedef struct _MessageData {
        GInputStream *input_stream;
        gssize length;
-       goffset read_from;
 } MessageData;
 
 static MessageData *
@@ -1615,14 +1725,15 @@ message_data_new (GInputStream *input_stream,
                  gssize length)
 {
        MessageData *md;
+       EInputStreamWrapper *wrapper;
+
+       wrapper = g_object_new (e_input_stream_wrapper_get_type (), NULL);
+       e_input_stream_wrapper_assign (wrapper, input_stream);
 
        md = g_slice_new0 (MessageData);
-       md->input_stream = g_object_ref (input_stream);
+       md->input_stream = G_INPUT_STREAM (wrapper);
        md->length = length;
 
-       if (G_IS_SEEKABLE (input_stream))
-               md->read_from = g_seekable_tell (G_SEEKABLE (input_stream));
-
        return md;
 }
 
@@ -1644,11 +1755,13 @@ e_soup_session_message_restarted_cb (SoupMessage *message,
        GInputStream *input_stream;
        gssize length = 0;
 
-       input_stream = e_soup_session_util_get_message_request_body (message, &length);
+       input_stream = e_soup_session_util_ref_message_request_body (message, &length);
 
        g_return_if_fail (input_stream != NULL);
 
        soup_message_set_request_body (message, NULL, input_stream, length);
+
+       g_clear_object (&input_stream);
 }
 
 #define MESSAGE_DATA_KEY "ESoupSession::message-data"
@@ -1686,7 +1799,7 @@ e_soup_session_util_set_message_request_body (SoupMessage *message,
        g_signal_connect (message, "restarted",
                G_CALLBACK (e_soup_session_message_restarted_cb), NULL);
 
-       soup_message_set_request_body (message, content_type, input_stream, length);
+       soup_message_set_request_body (message, content_type, md->input_stream, length);
 }
 
 /**
@@ -1732,7 +1845,7 @@ e_soup_session_util_set_message_request_body_from_data (SoupMessage *message,
 }
 
 /**
- * e_soup_session_util_get_message_request_body:
+ * e_soup_session_util_ref_message_request_body:
  * @message: a #SoupMessage
  * @out_length: (out) (optional): length of the input stream
  *
@@ -1744,14 +1857,14 @@ e_soup_session_util_set_message_request_body_from_data (SoupMessage *message,
  * a #SoupSession, nor modify the input stream position until
  * the @message lefts the #SoupSession.
  *
- * Returns: (nullable) (transfer none): a #GInputStream with the request body
- *    being previously set, on %NULL. The @out_length is set to the length
+ * Returns: (nullable) (transfer full): a new #GInputStream with the request
+ *    body being previously set, or %NULL. The @out_length is set to the length
  *    of the returned input stream.
  *
  * Since: 3.48
  **/
 GInputStream *
-e_soup_session_util_get_message_request_body (SoupMessage *message,
+e_soup_session_util_ref_message_request_body (SoupMessage *message,
                                              gssize *out_length)
 {
        MessageData *md;
@@ -1763,11 +1876,8 @@ e_soup_session_util_get_message_request_body (SoupMessage *message,
        if (!md || !md->input_stream)
                return NULL;
 
-       if (G_IS_SEEKABLE (md->input_stream) && md->read_from != g_seekable_tell (G_SEEKABLE 
(md->input_stream)))
-               g_seekable_seek (G_SEEKABLE (md->input_stream), md->read_from, G_SEEK_SET, NULL, NULL);
-
        if (out_length)
                *out_length = md->length;
 
-       return md->input_stream;
+       return e_input_stream_wrapper_dup (E_INPUT_STREAM_WRAPPER (md->input_stream));
 }
diff --git a/src/libedataserver/e-soup-session.h b/src/libedataserver/e-soup-session.h
index a7be9e725..d5112397d 100644
--- a/src/libedataserver/e-soup-session.h
+++ b/src/libedataserver/e-soup-session.h
@@ -147,7 +147,7 @@ void                e_soup_session_util_set_message_request_body_from_data
                                                         gconstpointer data,
                                                         gssize length,
                                                         GDestroyNotify free_func);
-GInputStream * e_soup_session_util_get_message_request_body
+GInputStream * e_soup_session_util_ref_message_request_body
                                                        (SoupMessage *message,
                                                         gssize *out_length);
 G_END_DECLS