# HG changeset patch
# Node ID 2aa5d09210a98df0546601af08f4277df67861a5
# Parent  9e89897451b4cb7fa8f1d01bb83d2f9dc6f7e93d
diff --git a/xeme/tests/testinputstream.c b/xeme/tests/testinputstream.c
--- a/xeme/tests/testinputstream.c
+++ b/xeme/tests/testinputstream.c
@@ -44,41 +44,15 @@
 	*counter = *counter + 1;
 }
 
-/**
- * test_xeme_connection_timeout_cb: (skip)
- * @data: The main loop to quit.
- *
- * Used for callbacks for timing out the main loops when running tests. If this
- * is called, it will assert that it was reached and cause the test to fail.
- *
- * Returns: G_SOURCE_REMOVE to not be called again.
- */
-static gboolean
-test_xeme_input_stream_timeout_cb(gpointer data) {
-	g_main_loop_quit(data);
-
-	g_assert_not_reached();
-
-	return G_SOURCE_REMOVE;
-}
-
-static gboolean
-test_xeme_input_stream_quit_cb(gpointer data) {
-	g_main_loop_quit(data);
-
-	return G_SOURCE_REMOVE;
-}
-
 /******************************************************************************
  * Tests
  *****************************************************************************/
 static void
 test_xeme_input_stream_start(void) {
-	XemeInputStream *input_stream = NULL;
-	XemeOutputStream *output_stream = NULL;
+	XemeInputStream *input = NULL;
+	XemeOutputStream *output = NULL;
 	GError *error = NULL;
-	GMainLoop *loop = NULL;
-	GInputStream *input = NULL;
+	GInputStream *input_stream = NULL;
 	gboolean ret = FALSE;
 	const char *input_data = "" \
 		"<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
@@ -90,50 +64,49 @@
 	const char *value = NULL;
 	guint counter = 0;
 
-	output_stream = xeme_output_stream_new(NULL);
-	xeme_stream_set_to(XEME_STREAM(output_stream), "juliet@im.example.com");
-	xeme_stream_set_from(XEME_STREAM(output_stream), "im.example.com");
+	output = xeme_output_stream_new(NULL);
+	xeme_stream_set_to(XEME_STREAM(output), "juliet@im.example.com");
+	xeme_stream_set_from(XEME_STREAM(output), "im.example.com");
 
-	input_stream = xeme_input_stream_new();
-	g_signal_connect(input_stream, "closed",
+	input = xeme_input_stream_new();
+	g_signal_connect(input, "closed",
 	                 G_CALLBACK(xeme_input_stream_stream_closed_cb), &counter);
 
-	input = g_memory_input_stream_new_from_data(input_data, -1, NULL);
+	input_stream = g_memory_input_stream_new_from_data(input_data, -1, NULL);
 
-	ret = xeme_input_stream_start(input_stream, input, output_stream, &error);
-	g_clear_object(&output_stream);
+	ret = xeme_input_stream_start(input, input_stream, output, &error);
+	g_clear_object(&output);
 
 	g_assert_no_error(error);
 	g_assert_true(ret);
 
-	/* Create our main loop and add some timeouts and stuff. */
-	loop = g_main_loop_new(NULL, FALSE);
-	g_timeout_add(2, test_xeme_input_stream_quit_cb, loop);
-	g_timeout_add(100, test_xeme_input_stream_timeout_cb, loop);
-
-	g_main_loop_run(loop);
+	g_main_context_iteration(NULL, FALSE);
 
 	/* If we made it here, the quit cb got called, so verify everything. */
-	value = xeme_stream_get_from(XEME_STREAM(input_stream));
+	value = xeme_stream_get_from(XEME_STREAM(input));
 	g_assert_cmpstr(value, ==, "juliet@im.example.com");
 
-	value = xeme_stream_get_to(XEME_STREAM(input_stream));
+	value = xeme_stream_get_to(XEME_STREAM(input));
 	g_assert_cmpstr(value, ==, "im.example.com");
 
-	value = xeme_stream_get_id(XEME_STREAM(input_stream));
+	value = xeme_stream_get_id(XEME_STREAM(input));
 	g_assert_cmpstr(value, ==, "t7AMCin9zjMNwQKDnplntZPIDEI=");
 
-	value = xeme_stream_get_version(XEME_STREAM(input_stream));
+	value = xeme_stream_get_version(XEME_STREAM(input));
 	g_assert_cmpstr(value, ==, "1.0");
 
-	value = xeme_stream_get_language(XEME_STREAM(input_stream));
+	value = xeme_stream_get_language(XEME_STREAM(input));
 	g_assert_cmpstr(value, ==, "en");
 
 	/* Make sure the close signal was emitted. */
 	g_assert_cmpuint(counter, ==, 1);
 
-	g_clear_object(&input);
-	g_clear_object(&input_stream);
+	ret = xeme_input_stream_stop(input, &error);
+	g_assert_no_error(error);
+	g_assert_true(ret);
+
+	g_assert_finalize_object(input_stream);
+	g_assert_finalize_object(input);
 }
 
 /******************************************************************************
diff --git a/xeme/xemeinputstream.c b/xeme/xemeinputstream.c
--- a/xeme/xemeinputstream.c
+++ b/xeme/xemeinputstream.c
@@ -59,13 +59,6 @@
  * Helpers
  *****************************************************************************/
 static void
-xeme_input_stream_close(XemeInputStream *stream,
-                        G_GNUC_UNUSED GError **error)
-{
-	g_signal_emit(stream, signals[SIG_CLOSED], 0);
-}
-
-static void
 xeme_input_stream_read_bytes_async_cb(G_GNUC_UNUSED GObject *source,
                                       GAsyncResult *result, gpointer data)
 {
@@ -84,7 +77,7 @@
 		          error != NULL ? error->message : "unknown error");
 		g_clear_error(&error);
 
-		xeme_input_stream_close(stream, NULL);
+		xeme_input_stream_stop(stream, NULL);
 
 		return;
 	}
@@ -93,6 +86,10 @@
 	if(!g_markup_parse_context_parse(priv->context, raw, length, &error)) {
 		g_warning("we hit an error: %s", error->message);
 		length = 0;
+
+		xeme_input_stream_stop(stream, NULL);
+
+		return;
 	}
 
 	/* Unref the data we read. */
@@ -213,7 +210,7 @@
 
 	if(xeme_str_equal(name, "stream:stream")) {
 		/* The connection is being terminated. */
-		xeme_input_stream_close(stream, error);
+		xeme_input_stream_stop(stream, error);
 
 		return;
 	} else {
@@ -237,8 +234,9 @@
 	priv = xeme_input_stream_get_instance_private(stream);
 
 	g_clear_pointer(&priv->context, g_markup_parse_context_free);
+	g_clear_object(&priv->features);
+	g_clear_object(&priv->input);
 	g_clear_object(&priv->output);
-	g_clear_object(&priv->features);
 
 	G_OBJECT_CLASS(xeme_input_stream_parent_class)->finalize(obj);
 }
@@ -365,6 +363,9 @@
 	}
 
 	cancellable = xeme_stream_get_cancellable(XEME_STREAM(stream));
+	if(G_IS_CANCELLABLE(cancellable)) {
+		g_cancellable_reset(cancellable);
+	}
 
 	/* Store the output stream. */
 	if(g_set_object(&priv->output, output)) {
@@ -385,6 +386,35 @@
 	return TRUE;
 }
 
+gboolean
+xeme_input_stream_stop(XemeInputStream *stream, GError **error) {
+	XemeInputStreamPrivate *priv = NULL;
+	GCancellable *cancellable = NULL;
+	gboolean ret = FALSE;
+
+	g_return_val_if_fail(XEME_IS_INPUT_STREAM(stream), FALSE);
+
+	priv = xeme_input_stream_get_instance_private(stream);
+	if(priv->running == FALSE) {
+		return TRUE;
+	}
+
+	cancellable = xeme_stream_get_cancellable(XEME_STREAM(stream));
+	if(G_IS_CANCELLABLE(cancellable) && !g_cancellable_is_cancelled(cancellable)) {
+		g_cancellable_cancel(cancellable);
+	}
+
+	priv->running = FALSE;
+
+	ret = g_input_stream_close(priv->input, NULL, error);
+	if(ret) {
+		g_signal_emit(stream, signals[SIG_CLOSED], 0);
+		g_clear_object(&priv->input);
+	}
+
+	return ret;
+}
+
 void
 xeme_input_stream_restart_requested(XemeInputStream *stream) {
 	g_return_if_fail(XEME_IS_INPUT_STREAM(stream));
diff --git a/xeme/xemeinputstream.h b/xeme/xemeinputstream.h
--- a/xeme/xemeinputstream.h
+++ b/xeme/xemeinputstream.h
@@ -92,6 +92,19 @@
 gboolean xeme_input_stream_start(XemeInputStream *stream, GInputStream *input, XemeOutputStream *output, GError **error);
 
 /**
+ * xeme_input_stream_stop:
+ * @error: (out) (optional): a return address for a #GError
+ *
+ * Stops the stream and closes its backing stream.
+ *
+ * Returns: true if successful; otherwise false
+ *
+ * Since: 0.1
+ */
+XEME_AVAILABLE_IN_0_1
+gboolean xeme_input_stream_stop(XemeInputStream *stream, GError **error);
+
+/**
  * xeme_input_stream_restart_requested:
  * @stream: The instance.
  *
