diff --git a/src/modules/module-protocol-pulse/pulse-server.c b/src/modules/module-protocol-pulse/pulse-server.c
index 939add174..c98ae4f3b 100644
--- a/src/modules/module-protocol-pulse/pulse-server.c
+++ b/src/modules/module-protocol-pulse/pulse-server.c
@@ -470,8 +470,6 @@ static int reply_create_playback_stream(struct stream *stream, struct pw_manager
 	if (stream->buffer == NULL)
 		return -errno;
 
-	spa_ringbuffer_init(&stream->ring);
-
 	if (lat.num * defs->min_quantum.denom / lat.denom < defs->min_quantum.num)
 		lat.num = (defs->min_quantum.num * lat.denom +
 				(defs->min_quantum.denom -1)) / defs->min_quantum.denom;
@@ -611,8 +609,6 @@ static int reply_create_record_stream(struct stream *stream, struct pw_manager_o
 	if (stream->buffer == NULL)
 		return -errno;
 
-	spa_ringbuffer_init(&stream->ring);
-
 	if (lat.num * defs->min_quantum.denom / lat.denom < defs->min_quantum.num)
 		lat.num = (defs->min_quantum.num * lat.denom +
 				(defs->min_quantum.denom -1)) / defs->min_quantum.denom;
@@ -1566,30 +1562,17 @@ static int do_create_playback_stream(struct client *client, uint32_t command, ui
 	if (n_valid_formats == 0)
 		goto error_no_formats;
 
-	stream = calloc(1, sizeof(struct stream));
+	stream = stream_new(client, STREAM_TYPE_PLAYBACK, tag, &ss, &map, &attr);
 	if (stream == NULL)
 		goto error_errno;
 
-	spa_list_init(&stream->link);
-	stream->impl = impl;
-	stream->client = client;
 	stream->corked = corked;
 	stream->adjust_latency = adjust_latency;
 	stream->early_requests = early_requests;
-	stream->channel = pw_map_insert_new(&client->streams, stream);
-	if (stream->channel == SPA_ID_INVALID)
-		goto error_errno;
-
-	stream->type = STREAM_TYPE_PLAYBACK;
-	stream->direction = PW_DIRECTION_OUTPUT;
-	stream->create_tag = tag;
-	stream->ss = ss;
-	stream->map = map;
 	stream->volume = volume;
 	stream->volume_set = volume_set;
 	stream->muted = muted;
 	stream->muted_set = muted_set;
-	stream->attr = attr;
 	stream->is_underrun = true;
 	stream->underrun_for = -1;
 
@@ -1829,30 +1812,17 @@ static int do_create_record_stream(struct client *client, uint32_t command, uint
 	if (n_valid_formats == 0)
 		goto error_no_formats;
 
-	stream = calloc(1, sizeof(struct stream));
+	stream = stream_new(client, STREAM_TYPE_RECORD, tag, &ss, &map, &attr);
 	if (stream == NULL)
 		goto error_errno;
 
-	spa_list_init(&stream->link);
-	stream->type = STREAM_TYPE_RECORD;
-	stream->direction = PW_DIRECTION_INPUT;
-	stream->impl = impl;
-	stream->client = client;
 	stream->corked = corked;
 	stream->adjust_latency = adjust_latency;
 	stream->early_requests = early_requests;
-	stream->channel = pw_map_insert_new(&client->streams, stream);
-	if (stream->channel == SPA_ID_INVALID)
-		goto error_errno;
-
-	stream->create_tag = tag;
-	stream->ss = ss;
-	stream->map = map;
 	stream->volume = volume;
 	stream->volume_set = volume_set;
 	stream->muted = muted;
 	stream->muted_set = muted_set;
-	stream->attr = attr;
 
 	if (client->quirks & QUIRK_REMOVE_CAPTURE_DONT_MOVE)
 		no_move = false;
@@ -2048,7 +2018,6 @@ static int do_get_record_latency(struct client *client, uint32_t command, uint32
 
 static int do_create_upload_stream(struct client *client, uint32_t command, uint32_t tag, struct message *m)
 {
-	struct impl *impl = client->impl;
 	const char *name;
 	struct sample_spec ss;
 	struct channel_map map;
@@ -2096,32 +2065,18 @@ static int do_create_upload_stream(struct client *client, uint32_t command, uint
 			client->name, commands[command].name, tag,
 			name, length);
 
-	stream = calloc(1, sizeof(struct stream));
+	stream = stream_new(client, STREAM_TYPE_UPLOAD, tag, &ss, &map, &(struct buffer_attr) {
+		.maxlength = length,
+	});
 	if (stream == NULL)
 		goto error_errno;
 
-	spa_list_init(&stream->link);
-	stream->type = STREAM_TYPE_UPLOAD;
-	stream->direction = PW_DIRECTION_OUTPUT;
-	stream->impl = impl;
-	stream->client = client;
-	stream->channel = pw_map_insert_new(&client->streams, stream);
-	if (stream->channel == SPA_ID_INVALID)
-		goto error_errno;
-
-	stream->create_tag = tag;
-	stream->ss = ss;
-	stream->map = map;
 	stream->props = props;
 
-	stream->attr.maxlength = length;
-
 	stream->buffer = calloc(1, stream->attr.maxlength);
 	if (stream->buffer == NULL)
 		goto error_errno;
 
-	spa_ringbuffer_init(&stream->ring);
-
 	reply = reply_new(client, tag);
 	message_put(reply,
 		TAG_U32, stream->channel,
diff --git a/src/modules/module-protocol-pulse/stream.c b/src/modules/module-protocol-pulse/stream.c
index e38d4c13f..946edbd50 100644
--- a/src/modules/module-protocol-pulse/stream.c
+++ b/src/modules/module-protocol-pulse/stream.c
@@ -44,6 +44,51 @@
 #include "reply.h"
 #include "stream.h"
 
+struct stream *stream_new(struct client *client, enum stream_type type, uint32_t create_tag,
+			  const struct sample_spec *ss, const struct channel_map *map,
+			  const struct buffer_attr *attr)
+{
+	int res;
+
+	struct stream *stream = calloc(1, sizeof(*stream));
+	if (stream == NULL)
+		return NULL;
+
+	stream->channel = pw_map_insert_new(&client->streams, stream);
+	if (stream->channel == SPA_ID_INVALID)
+		goto error_errno;
+
+	stream->impl = client->impl;
+	stream->client = client;
+	stream->type = type;
+	stream->create_tag = create_tag;
+	stream->ss = *ss;
+	stream->map = *map;
+	stream->attr = *attr;
+	spa_ringbuffer_init(&stream->ring);
+
+	switch (type) {
+	case STREAM_TYPE_RECORD:
+		stream->direction = PW_DIRECTION_INPUT;
+		break;
+	case STREAM_TYPE_PLAYBACK:
+	case STREAM_TYPE_UPLOAD:
+		stream->direction = PW_DIRECTION_OUTPUT;
+		break;
+	default:
+		spa_assert_not_reached();
+	}
+
+	return stream;
+
+error_errno:
+	res = errno;
+	free(stream);
+	errno = res;
+
+	return NULL;
+}
+
 void stream_free(struct stream *stream)
 {
 	struct client *client = stream->client;
diff --git a/src/modules/module-protocol-pulse/stream.h b/src/modules/module-protocol-pulse/stream.h
index 4bf26f397..790e0300c 100644
--- a/src/modules/module-protocol-pulse/stream.h
+++ b/src/modules/module-protocol-pulse/stream.h
@@ -47,6 +47,12 @@ struct buffer_attr {
 	uint32_t fragsize;
 };
 
+enum stream_type {
+	STREAM_TYPE_RECORD,
+	STREAM_TYPE_PLAYBACK,
+	STREAM_TYPE_UPLOAD,
+};
+
 struct stream {
 	struct spa_list link;
 	uint32_t create_tag;
@@ -55,10 +61,7 @@ struct stream {
 
 	struct impl *impl;
 	struct client *client;
-#define STREAM_TYPE_RECORD	0
-#define STREAM_TYPE_PLAYBACK	1
-#define STREAM_TYPE_UPLOAD	2
-	uint32_t type;
+	enum stream_type type;
 	enum pw_direction direction;
 
 	struct pw_properties *props;
@@ -105,6 +108,9 @@ struct stream {
 	unsigned int killed:1;
 };
 
+struct stream *stream_new(struct client *client, enum stream_type type, uint32_t create_tag,
+			  const struct sample_spec *ss, const struct channel_map *map,
+			  const struct buffer_attr *attr);
 void stream_free(struct stream *stream);
 void stream_flush(struct stream *stream);
 uint32_t stream_pop_missing(struct stream *stream);