# HG changeset patch
# Node ID 4c240082bd93
# Parent  841b70e8dbf5
diff --git a/xeme/tests/testinputstream.c b/xeme/tests/testinputstream.c
--- a/xeme/tests/testinputstream.c
+++ b/xeme/tests/testinputstream.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 Xeme Developers
+ * Copyright (C) 2023-2024 Xeme Developers
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -69,7 +69,8 @@
  *****************************************************************************/
 static void
 test_xeme_input_stream_start(void) {
-	XemeInputStream *stream = NULL;
+	XemeInputStream *input_stream = NULL;
+	XemeOutputStream *output_stream = NULL;
 	GError *error = NULL;
 	GMainLoop *loop = NULL;
 	GInputStream *input = NULL;
@@ -84,13 +85,17 @@
 	const char *value = NULL;
 	guint counter = 0;
 
-	stream = xeme_input_stream_new();
-	g_signal_connect(stream, "closed",
+	output_stream = xeme_output_stream_new("julient@im.example.com",
+	                                       "im.example.com", NULL);
+
+	input_stream = xeme_input_stream_new();
+	g_signal_connect(input_stream, "closed",
 	                 G_CALLBACK(xeme_input_stream_stream_closed_cb), &counter);
 
 	input = g_memory_input_stream_new_from_data(input_data, -1, NULL);
 
-	ret = xeme_input_stream_start(stream, input, &error);
+	ret = xeme_input_stream_start(input_stream, input, output_stream, &error);
+	g_clear_object(&output_stream);
 
 	g_assert_no_error(error);
 	g_assert_true(ret);
@@ -103,26 +108,26 @@
 	g_main_loop_run(loop);
 
 	/* If we made it here, the quit cb got called, so verify everything. */
-	value = xeme_stream_get_from(XEME_STREAM(stream));
+	value = xeme_stream_get_from(XEME_STREAM(input_stream));
 	g_assert_cmpstr(value, ==, "juliet@im.example.com");
 
-	value = xeme_stream_get_to(XEME_STREAM(stream));
+	value = xeme_stream_get_to(XEME_STREAM(input_stream));
 	g_assert_cmpstr(value, ==, "im.example.com");
 
-	value = xeme_stream_get_id(XEME_STREAM(stream));
+	value = xeme_stream_get_id(XEME_STREAM(input_stream));
 	g_assert_cmpstr(value, ==, "t7AMCin9zjMNwQKDnplntZPIDEI=");
 
-	value = xeme_stream_get_version(XEME_STREAM(stream));
+	value = xeme_stream_get_version(XEME_STREAM(input_stream));
 	g_assert_cmpstr(value, ==, "1.0");
 
-	value = xeme_stream_get_language(XEME_STREAM(stream));
+	value = xeme_stream_get_language(XEME_STREAM(input_stream));
 	g_assert_cmpstr(value, ==, "en");
 
 	/* Make sure the close signal was emitted. */
 	g_assert_cmpuint(counter, ==, 1);
 
 	g_clear_object(&input);
-	g_clear_object(&stream);
+	g_clear_object(&input_stream);
 }
 
 /******************************************************************************
diff --git a/xeme/xemeconnection.c b/xeme/xemeconnection.c
--- a/xeme/xemeconnection.c
+++ b/xeme/xemeconnection.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 Xeme Developers
+ * Copyright (C) 2023-2024 Xeme Developers
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -90,27 +90,34 @@
 
 	iostream = G_IO_STREAM(connection->socket_connection);
 
-	istream = g_io_stream_get_input_stream(iostream);
-	if(!xeme_input_stream_start(connection->input, istream, &error)) {
+	ostream = g_io_stream_get_output_stream(iostream);
+	output = xeme_output_stream_new(connection->domain, connection->jid,
+	                                connection->cancellable);
+	if(!xeme_output_stream_start(output, ostream, &error)) {
 		g_task_return_error(task, error);
-		g_clear_object(&task);
+
+		g_clear_object(&output);
 
 		xeme_connection_close(connection, NULL);
 
 		return;
 	}
 
-	ostream = g_io_stream_get_output_stream(iostream);
-	output = xeme_output_stream_new(connection->domain, connection->jid,
-	                                connection->cancellable);
-	if(!xeme_output_stream_start(output, ostream, &error)) {
+	istream = g_io_stream_get_input_stream(iostream);
+	if(!xeme_input_stream_start(connection->input, istream, output, &error)) {
 		g_task_return_error(task, error);
 
+		g_clear_object(&task);
+		g_clear_object(&output);
+
 		xeme_connection_close(connection, NULL);
-	} else {
-		g_task_return_pointer(task, output, g_object_unref);
+
+		return;
 	}
 
+	/* Finally return the output stream. */
+	g_task_return_pointer(task, output, g_object_unref);
+
 	g_clear_object(&task);
 }
 
diff --git a/xeme/xemeinputstream.h b/xeme/xemeinputstream.h
--- a/xeme/xemeinputstream.h
+++ b/xeme/xemeinputstream.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 Xeme Developers
+ * Copyright (C) 2023-2024 Xeme Developers
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -24,6 +24,7 @@
 #include <gio/gio.h>
 
 #include <xeme/xememessage.h>
+#include <xeme/xemeoutputstream.h>
 #include <xeme/xemestream.h>
 #include <xeme/xemeversion.h>
 
@@ -67,6 +68,7 @@
  * @stream: The instance.
  * @input: (transfer none): A [class@Gio.InputStream] for reading data from the
  *         server.
+ * @output: (transfer none): The output stream for sending data.
  * @error: (nullable): A return address for a #GError.
  *
  * Starts processing @input as an incoming XMPP stream. It is the
@@ -82,7 +84,7 @@
  * Since: 0.1
  */
 XEME_AVAILABLE_IN_0_1
-gboolean xeme_input_stream_start(XemeInputStream *stream, GInputStream *input, GError **error);
+gboolean xeme_input_stream_start(XemeInputStream *stream, GInputStream *input, XemeOutputStream *output, GError **error);
 
 /**
  * xeme_input_stream_restart_requested:
@@ -102,6 +104,24 @@
 XEME_AVAILABLE_IN_0_1
 void xeme_input_stream_restart_requested(XemeInputStream *stream);
 
+/**
+ * xeme_input_stream_get_output_stream:
+ * @stream: The instance.
+ *
+ * Gets the [class@OutputStream] that should be used for requests and responses
+ * with the remote end of the connection.
+ *
+ * Note: this will only be set after [method@InputStream.start] has been
+ * called.
+ *
+ * Returns: (transfer none) (nullable): The output stream if set, otherwise
+ *          %NULL.
+ *
+ * Since: 0.1
+ */
+XEME_AVAILABLE_IN_0_1
+XemeOutputStream *xeme_input_stream_get_output_stream(XemeInputStream *stream);
+
 G_END_DECLS
 
 #endif /* XEME_INPUT_STREAM_H */
diff --git a/xeme/xemeinputstream.c b/xeme/xemeinputstream.c
--- a/xeme/xemeinputstream.c
+++ b/xeme/xemeinputstream.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 Xeme Developers
+ * Copyright (C) 2023-2024 Xeme Developers
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -24,6 +24,13 @@
 const gsize XEME_INPUT_STREAM_BUFFER_LEN = 4096;
 
 enum {
+	PROP_0,
+	PROP_OUTPUT_STREAM,
+	N_PROPERTIES,
+};
+static GParamSpec *properties[N_PROPERTIES] = {NULL, };
+
+enum {
 	SIG_CLOSED,
 	SIG_RESTART_REQUESTED,
 	N_SIGNALS,
@@ -34,6 +41,7 @@
 	gboolean running;
 
 	GInputStream *input;
+	XemeOutputStream *output;
 	GMarkupParseContext *context;
 
 	XemeExtensionMap *features;
@@ -224,12 +232,29 @@
 	priv = xeme_input_stream_get_instance_private(stream);
 
 	g_clear_pointer(&priv->context, g_markup_parse_context_free);
+	g_clear_object(&priv->output);
 	g_clear_object(&priv->features);
 
 	G_OBJECT_CLASS(xeme_input_stream_parent_class)->finalize(obj);
 }
 
 static void
+xeme_input_stream_get_property(GObject *obj, guint param_id, GValue *value,
+                               GParamSpec *pspec)
+{
+	XemeInputStream *stream = XEME_INPUT_STREAM(obj);
+
+	switch(param_id) {
+	case PROP_OUTPUT_STREAM:
+		g_value_set_object(value, xeme_input_stream_get_output_stream(stream));
+		break;
+	default:
+		G_OBJECT_WARN_INVALID_PROPERTY_ID(obj, param_id, pspec);
+		break;
+	}
+}
+
+static void
 xeme_input_stream_init(XemeInputStream *stream) {
 	XemeInputStreamPrivate *priv = NULL;
 
@@ -246,6 +271,24 @@
 	GObjectClass *obj_class = G_OBJECT_CLASS(klass);
 
 	obj_class->finalize = xeme_input_stream_finalize;
+	obj_class->get_property = xeme_input_stream_get_property;
+
+	/**
+	 * XemeInputStream:output-stream:
+	 *
+	 * The output stream that should be used for requests and responses.
+	 *
+	 * This is only set after [method@InputStream.start] is called.
+	 *
+	 * Since: 0.1
+	 */
+	properties[PROP_OUTPUT_STREAM] = g_param_spec_object(
+		"output-stream", "output-stream",
+		"The output stream to use for requests/responses.",
+		XEME_TYPE_OUTPUT_STREAM,
+		G_PARAM_READABLE | G_PARAM_STATIC_STRINGS);
+
+	g_object_class_install_properties(obj_class, N_PROPERTIES, properties);
 
 	/**
 	 * XemeInputStream::closed:
@@ -298,13 +341,14 @@
 
 gboolean
 xeme_input_stream_start(XemeInputStream *stream, GInputStream *input,
-                        GError **error)
+                        XemeOutputStream *output, GError **error)
 {
 	XemeInputStreamPrivate *priv = NULL;
 	GCancellable *cancellable = NULL;
 
 	g_return_val_if_fail(XEME_IS_INPUT_STREAM(stream), FALSE);
 	g_return_val_if_fail(G_IS_INPUT_STREAM(input), FALSE);
+	g_return_val_if_fail(XEME_IS_OUTPUT_STREAM(output), FALSE);
 	g_return_val_if_fail(error == NULL || *error == NULL, FALSE);
 
 	priv = xeme_input_stream_get_instance_private(stream);
@@ -318,6 +362,12 @@
 
 	cancellable = xeme_stream_get_cancellable(XEME_STREAM(stream));
 
+	/* Store the output stream. */
+	if(g_set_object(&priv->output, output)) {
+		g_object_notify_by_pspec(G_OBJECT(stream),
+		                         properties[PROP_OUTPUT_STREAM]);
+	}
+
 	/* Setup the read handler. */
 	priv->input = g_object_ref(input);
 	g_input_stream_read_bytes_async(priv->input,
@@ -337,3 +387,14 @@
 
 	g_signal_emit(stream, signals[SIG_RESTART_REQUESTED], 0);
 }
+
+XemeOutputStream *
+xeme_input_stream_get_output_stream(XemeInputStream *stream) {
+	XemeInputStreamPrivate *priv = NULL;
+
+	g_return_val_if_fail(XEME_IS_INPUT_STREAM(stream), NULL);
+
+	priv = xeme_input_stream_get_instance_private(stream);
+
+	return priv->output;
+}
