rtmp2/connection: pass the parent cancellable down to the connection

Otherwise, when rtpm2src cancels an inflight operation that has a queued
message stored, then the rtmp connection operation is not stopped.

If the cancellation occurs during rtmp connection start up, then
rtpm2src does not have any way of accessing the connection object as it
has not been returned yet.  As a result, rtpm2src will cancel, the
connection will still be processing things and the
GMainContext/GMainLoop associated with the outstanding operation will be
destroyed.  All outstanding operations and the rtmpconnection object will
therefore be leaked in this case.

Fixes: https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1425
Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/1865>

3 files changed, 12 insertions, 6 deletions

diff --git a/gst/rtmp2/rtmp/rtmpclient.c b/gst/rtmp2/rtmp/rtmpclient.c
index dae50c698..d2746ef74 100644
--- a/gst/rtmp2/rtmp/rtmpclient.c
+++ b/gst/rtmp2/rtmp/rtmpclient.c
@@ -499,7 +499,8 @@ handshake_done (GObject * source, GAsyncResult * result, gpointer user_data)
     return;
   }
 
-  data->connection = gst_rtmp_connection_new (socket_connection);
+  data->connection = gst_rtmp_connection_new (socket_connection,
+      g_task_get_cancellable (task));
   data->error_handler_id = g_signal_connect (data->connection,
       "error", G_CALLBACK (connection_error), task);
 
@@ -510,8 +511,9 @@ static void
 connection_error (GstRtmpConnection * connection, gpointer user_data)
 {
   GTask *task = user_data;
-  g_task_return_new_error (task, G_IO_ERROR, G_IO_ERROR_FAILED,
-      "error during connection attempt");
+  if (!g_task_had_error (task))
+    g_task_return_new_error (task, G_IO_ERROR, G_IO_ERROR_FAILED,
+        "error during connection attempt");
 }
 
 static gchar *
diff --git a/gst/rtmp2/rtmp/rtmpconnection.c b/gst/rtmp2/rtmp/rtmpconnection.c
index 22d1e69eb..a6eb5ad58 100644
--- a/gst/rtmp2/rtmp/rtmpconnection.c
+++ b/gst/rtmp2/rtmp/rtmpconnection.c
@@ -251,7 +251,6 @@ gst_rtmp_connection_class_init (GstRtmpConnectionClass * klass)
 static void
 gst_rtmp_connection_init (GstRtmpConnection * rtmpconnection)
 {
-  rtmpconnection->cancellable = g_cancellable_new ();
   rtmpconnection->output_queue =
       g_async_queue_new_full ((GDestroyNotify) gst_buffer_unref);
   rtmpconnection->input_streams = gst_rtmp_chunk_streams_new ();
@@ -334,11 +333,16 @@ gst_rtmp_connection_set_socket_connection (GstRtmpConnection * sc,
 }
 
 GstRtmpConnection *
-gst_rtmp_connection_new (GSocketConnection * connection)
+gst_rtmp_connection_new (GSocketConnection * connection,
+    GCancellable * cancellable)
 {
   GstRtmpConnection *sc;
 
   sc = g_object_new (GST_TYPE_RTMP_CONNECTION, NULL);
+  if (cancellable)
+    sc->cancellable = g_object_ref (cancellable);
+  else
+    sc->cancellable = g_cancellable_new ();
 
   gst_rtmp_connection_set_socket_connection (sc, connection);
 
diff --git a/gst/rtmp2/rtmp/rtmpconnection.h b/gst/rtmp2/rtmp/rtmpconnection.h
index b20bc19ab..43e443062 100644
--- a/gst/rtmp2/rtmp/rtmpconnection.h
+++ b/gst/rtmp2/rtmp/rtmpconnection.h
@@ -51,7 +51,7 @@ typedef void (*GstRtmpCommandCallback) (const gchar * command_name,
 
 GType gst_rtmp_connection_get_type (void);
 
-GstRtmpConnection *gst_rtmp_connection_new (GSocketConnection * connection);
+GstRtmpConnection *gst_rtmp_connection_new (GSocketConnection * connection, GCancellable * cancellable);
 
 GSocket *gst_rtmp_connection_get_socket (GstRtmpConnection * connection);