Merge branch 'wip/carlosg/bus-cancellation' into 'master'

Fixes to bus cursor cancellation

Closes #383

See merge request https://gitlab.gnome.org/GNOME/tracker/-/merge_requests/547

2 files changed, 25 insertions, 17 deletions

diff --git a/src/libtracker-sparql/bus/tracker-bus-cursor.c b/src/libtracker-sparql/bus/tracker-bus-cursor.c
index ee0a08547..2b463299c 100644
--- a/src/libtracker-sparql/bus/tracker-bus-cursor.c
+++ b/src/libtracker-sparql/bus/tracker-bus-cursor.c
@@ -225,7 +225,7 @@ tracker_bus_cursor_next (TrackerSparqlCursor  *cursor,
 	 *              columns x 4 bytes for offsets]
 	 */
 	n_columns = g_data_input_stream_read_int32 (bus_cursor->data_stream,
-						    NULL, NULL);
+	                                            cancellable, NULL);
 
 	if (n_columns == 0) {
 		bus_cursor->finished = TRUE;
@@ -236,16 +236,16 @@ tracker_bus_cursor_next (TrackerSparqlCursor  *cursor,
 	bus_cursor->types = g_new0 (TrackerSparqlValueType, n_columns);
 
 	if (!g_input_stream_read_all (G_INPUT_STREAM (bus_cursor->data_stream),
-				      bus_cursor->types,
-				      n_columns * sizeof (gint32),
-				      NULL, NULL, error))
+	                              bus_cursor->types,
+	                              n_columns * sizeof (gint32),
+	                              NULL, cancellable, error))
 		return FALSE;
 
 	offsets = g_new0 (gint32, n_columns);
 	if (!g_input_stream_read_all (G_INPUT_STREAM (bus_cursor->data_stream),
 	                              offsets,
 	                              n_columns * sizeof (gint32),
-	                              NULL, NULL, error)) {
+	                              NULL, cancellable, error)) {
 		g_free (offsets);
 		return FALSE;
 	}
@@ -278,7 +278,7 @@ tracker_bus_cursor_next (TrackerSparqlCursor  *cursor,
 	if (!g_input_stream_read_all (G_INPUT_STREAM (bus_cursor->data_stream),
 	                              bus_cursor->row_data,
 	                              offsets[n_columns - 1] + 1,
-	                              NULL, NULL, error)) {
+	                              NULL, cancellable, error)) {
 		g_free (offsets);
 		return FALSE;
 	}
@@ -325,6 +325,7 @@ tracker_bus_cursor_next_async (TrackerSparqlCursor  *cursor,
 
 	task = g_task_new (cursor, cancellable, cb, user_data);
 	g_task_run_in_thread (task, next_in_thread);
+	g_object_unref (task);
 }
 
 static gboolean
diff --git a/src/libtracker-sparql/tracker-endpoint-dbus.c b/src/libtracker-sparql/tracker-endpoint-dbus.c
index 8ee65a3f4..4bc486c76 100644
--- a/src/libtracker-sparql/tracker-endpoint-dbus.c
+++ b/src/libtracker-sparql/tracker-endpoint-dbus.c
@@ -374,7 +374,8 @@ write_cursor (QueryRequest          *request,
 		glong cur_offset = -1;
 
 		if (!g_data_output_stream_put_int32 (request->data_stream, n_columns,
-		                                     NULL, &inner_error))
+		                                     request->cancellable,
+		                                     &inner_error))
 			break;
 
 		for (i = 0; i < n_columns; i++) {
@@ -382,7 +383,8 @@ write_cursor (QueryRequest          *request,
 
 			if (!g_data_output_stream_put_int32 (request->data_stream,
 			                                     tracker_sparql_cursor_get_value_type (cursor, i),
-			                                     NULL, &inner_error))
+			                                     request->cancellable,
+			                                     &inner_error))
 				goto out;
 
 			values[i] = tracker_sparql_cursor_get_string (cursor, i, &len);
@@ -393,17 +395,21 @@ write_cursor (QueryRequest          *request,
 
 		for (i = 0; i < n_columns; i++) {
 			if (!g_data_output_stream_put_int32 (request->data_stream,
-			                                     offsets[i], NULL, &inner_error))
+			                                     offsets[i],
+			                                     request->cancellable,
+			                                     &inner_error))
 				goto out;
 		}
 
 		for (i = 0; i < n_columns; i++) {
 			if (!g_data_output_stream_put_string (request->data_stream,
 			                                      values[i] ? values[i] : "",
-			                                      NULL, &inner_error))
+			                                      request->cancellable,
+			                                      &inner_error))
 				goto out;
 
-			if (!g_data_output_stream_put_byte (request->data_stream, 0, NULL,
+			if (!g_data_output_stream_put_byte (request->data_stream, 0,
+			                                    request->cancellable,
 			                                    &inner_error))
 				goto out;
 		}
@@ -431,7 +437,8 @@ handle_cursor_reply (GTask        *task,
 	TrackerSparqlCursor *cursor = source_object;
 	QueryRequest *request = task_data;
 	const gchar **variable_names = NULL;
-	GError *write_error = NULL;
+	GError *error = NULL;
+	gboolean retval;
 	gint i, n_columns;
 
 	n_columns = tracker_sparql_cursor_get_n_columns (cursor);
@@ -441,15 +448,15 @@ handle_cursor_reply (GTask        *task,
 
 	g_dbus_method_invocation_return_value (request->invocation, g_variant_new ("(^as)", variable_names));
 
-	if (!write_cursor (request, cursor, &write_error) &&
-	    !g_error_matches (write_error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
-		g_warning ("Endpoint failed to fully write cursor: %s\n", write_error->message);
+	retval = write_cursor (request, cursor, &error);
 	g_free (variable_names);
-	g_clear_error (&write_error);
 
 	tracker_sparql_cursor_close (cursor);
 
-	g_task_return_boolean (task, TRUE);
+	if (error)
+		g_task_return_error (task, error);
+	else
+		g_task_return_boolean (task, retval);
 }
 
 static void