diff --git a/src/audio/copier/copier.c b/src/audio/copier/copier.c
index 7c8d53726d20..76d464bd78f8 100644
--- a/src/audio/copier/copier.c
+++ b/src/audio/copier/copier.c
@@ -455,15 +455,14 @@ static int copier_copy_to_sinks(struct copier_data *cd, struct comp_dev *dev,
 				struct comp_buffer *src_c,
 				struct comp_copy_limits *processed_data)
 {
-	struct list_item *sink_list;
 	struct comp_buffer *sink;
 	int ret = 0;
 
 	/* module copy, one source to multiple sink buffers */
-	list_for_item(sink_list, &dev->bsink_list) {
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		struct comp_dev *sink_dev;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		sink_dev = sink->sink;
 		processed_data->sink_bytes = 0;
 		if (sink_dev->state == COMP_STATE_ACTIVE) {
@@ -475,6 +474,8 @@ static int copier_copy_to_sinks(struct copier_data *cd, struct comp_dev *dev,
 				 dev->ipc_config.id);
 			break;
 		}
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	if (!ret) {
@@ -1066,20 +1067,22 @@ static int copier_bind(struct processing_module *mod, void *data)
 	const uint32_t src_queue_id = bu->extension.r.src_queue;
 	struct copier_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct list_item *list;
 
 	if (dev->ipc_config.id != src_id)
 		return 0; /* Another component is a data producer */
 
 	/* update sink format */
-	list_for_item(list, &dev->bsink_list) {
-		struct comp_buffer *buffer = container_of(list, struct comp_buffer, source_list);
+	struct comp_buffer *buffer = comp_dev_get_first_data_consumer(dev);
+
+	while (buffer) {
 		uint32_t id = IPC4_SRC_QUEUE_ID(buf_get_id(buffer));
 
 		if (src_queue_id == id) {
 			ipc4_update_buffer_format(buffer, &cd->out_fmt[id]);
 			return 0;
 		}
+
+		buffer = comp_dev_get_next_data_consumer(dev, buffer);
 	}
 
 	comp_err(dev, "No sink buffer found for src_queue = %u", src_queue_id);
diff --git a/src/audio/copier/copier_generic.c b/src/audio/copier/copier_generic.c
index 4cb480ebe3f9..eaea5b1d3c57 100644
--- a/src/audio/copier/copier_generic.c
+++ b/src/audio/copier/copier_generic.c
@@ -62,7 +62,6 @@ void copier_update_params(struct copier_data *cd, struct comp_dev *dev,
 			  struct sof_ipc_stream_params *params)
 {
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 
 	memset(params, 0, sizeof(*params));
 	params->direction = cd->direction;
@@ -80,14 +79,14 @@ void copier_update_params(struct copier_data *cd, struct comp_dev *dev,
 	params->no_stream_position = 1;
 
 	/* update each sink format */
-	list_for_item(sink_list, &dev->bsink_list) {
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		int j;
-
-		sink = container_of(sink_list, struct comp_buffer, source_list);
-
 		j = IPC4_SINK_QUEUE_ID(buf_get_id(sink));
 
 		ipc4_update_buffer_format(sink, &cd->out_fmt[j]);
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	/* update params for the DMA buffer */
diff --git a/src/audio/crossover/crossover.c b/src/audio/crossover/crossover.c
index 9fffda1858c5..8b8598b7a426 100644
--- a/src/audio/crossover/crossover.c
+++ b/src/audio/crossover/crossover.c
@@ -100,15 +100,14 @@ static int crossover_assign_sinks(struct processing_module *mod,
 	struct sof_crossover_config *config = cd->config;
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 	int num_sinks = 0;
 	int i;
 	int j = 0;
 
-	list_for_item(sink_list, &dev->bsink_list) {
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		unsigned int sink_id, state;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		sink_id = crossover_get_sink_id(cd, buffer_pipeline_id(sink), j);
 		state = sink->sink->state;
 		if (state != dev->state) {
@@ -145,6 +144,8 @@ static int crossover_assign_sinks(struct processing_module *mod,
 		assigned_obufs[i] = &output_buffers[j];
 		enabled[j++] = true;
 		num_sinks++;
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	return num_sinks;
@@ -529,7 +530,6 @@ static int crossover_prepare(struct processing_module *mod,
 	struct comp_data *cd =  module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *source, *sink;
-	struct list_item *sink_list;
 	int channels;
 	int ret = 0;
 
@@ -546,8 +546,8 @@ static int crossover_prepare(struct processing_module *mod,
 	channels = audio_stream_get_channels(&source->stream);
 
 	/* Validate frame format and buffer size of sinks */
-	list_for_item(sink_list, &dev->bsink_list) {
-		sink = container_of(sink_list, struct comp_buffer, source_list);
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		if (cd->source_format != audio_stream_get_frm_fmt(&sink->stream)) {
 			comp_err(dev, "crossover_prepare(): Source fmt %d and sink fmt %d are different.",
 				 cd->source_format, audio_stream_get_frm_fmt(&sink->stream));
@@ -556,6 +556,8 @@ static int crossover_prepare(struct processing_module *mod,
 
 		if (ret < 0)
 			return ret;
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	comp_info(dev, "crossover_prepare(), source_format=%d, sink_formats=%d, nch=%d",
diff --git a/src/audio/crossover/crossover_ipc3.c b/src/audio/crossover/crossover_ipc3.c
index c715f1710460..59ca268e171a 100644
--- a/src/audio/crossover/crossover_ipc3.c
+++ b/src/audio/crossover/crossover_ipc3.c
@@ -37,15 +37,14 @@ int crossover_check_sink_assign(struct processing_module *mod,
 {
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 	int num_assigned_sinks = 0;
 	uint8_t assigned_sinks[SOF_CROSSOVER_MAX_STREAMS] = {0};
 	int i;
 
-	list_for_item(sink_list, &dev->bsink_list) {
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		unsigned int pipeline_id;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		pipeline_id = buffer_pipeline_id(sink);
 
 		i = crossover_get_stream_index(mod, config, pipeline_id);
@@ -63,6 +62,8 @@ int crossover_check_sink_assign(struct processing_module *mod,
 
 		assigned_sinks[i] = true;
 		num_assigned_sinks++;
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	return num_assigned_sinks;
diff --git a/src/audio/crossover/crossover_ipc4.c b/src/audio/crossover/crossover_ipc4.c
index d0a8fe4adab2..9f73eeb0cd88 100644
--- a/src/audio/crossover/crossover_ipc4.c
+++ b/src/audio/crossover/crossover_ipc4.c
@@ -111,7 +111,6 @@ void crossover_params(struct processing_module *mod)
 {
 	struct sof_ipc_stream_params *params = mod->stream_params;
 	struct comp_buffer *sinkb, *sourceb;
-	struct list_item *sink_list;
 	struct comp_dev *dev = mod->dev;
 
 	comp_dbg(dev, "crossover_params()");
@@ -122,8 +121,10 @@ void crossover_params(struct processing_module *mod)
 	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	list_for_item(sink_list, &dev->bsink_list) {
-		sinkb = container_of(sink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
+	while (sinkb) {
 		ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
+
+		sinkb = comp_dev_get_next_data_consumer(dev, sinkb);
 	}
 }
diff --git a/src/audio/dai-zephyr.c b/src/audio/dai-zephyr.c
index d2e672e700c1..896f4f459528 100644
--- a/src/audio/dai-zephyr.c
+++ b/src/audio/dai-zephyr.c
@@ -269,18 +269,17 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 #if CONFIG_IPC_MAJOR_4
-		struct list_item *sink_list;
 		/*
 		 * copy from local buffer to all sinks that are not gateway buffers
 		 * using the right PCM converter function.
 		 */
-		list_for_item(sink_list, &dev->bsink_list) {
+		struct comp_buffer *sink;
+
+		sink = comp_dev_get_first_data_consumer(dev);
+		while (sink) {
 			struct comp_dev *sink_dev;
-			struct comp_buffer *sink;
 			int j;
 
-			sink = container_of(sink_list, struct comp_buffer, source_list);
-
 			if (sink == dd->dma_buffer)
 				continue;
 
@@ -306,6 +305,8 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 				ret = stream_copy_from_no_consume(dd->local_buffer, sink,
 								  converter[j], bytes, dd->chmap);
 			}
+
+			sink = comp_dev_get_next_data_consumer(dev, sink);
 		}
 #endif
 		ret = dma_buffer_copy_to(dd->local_buffer, dd->dma_buffer,
@@ -319,20 +320,19 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 		ret = stream_copy_from_no_consume(dd->dma_buffer, dd->local_buffer,
 						  dd->process, bytes, dd->chmap);
 #if CONFIG_IPC_MAJOR_4
-		struct list_item *sink_list;
 		/* Skip in case of endpoint DAI devices created by the copier */
 		if (converter) {
 			/*
 			 * copy from DMA buffer to all sink buffers using the right PCM converter
 			 * function
 			 */
-			list_for_item(sink_list, &dev->bsink_list) {
+			struct comp_buffer *sink;
+
+			sink = comp_dev_get_first_data_consumer(dev);
+			while (sink) {
 				struct comp_dev *sink_dev;
-				struct comp_buffer *sink;
 				int j;
 
-				sink = container_of(sink_list, struct comp_buffer, source_list);
-
 				/* this has been handled above already */
 				if (sink == dd->local_buffer)
 					continue;
@@ -359,6 +359,8 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 					ret = stream_copy_from_no_consume(dd->dma_buffer,
 									  sink, converter[j],
 									  bytes, dd->chmap);
+
+				sink = comp_dev_get_next_data_consumer(dev, sink);
 			}
 		}
 #endif
@@ -1592,17 +1594,17 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 		sink_frames = free_bytes / audio_stream_frame_bytes(&dd->dma_buffer->stream);
 		frames = MIN(src_frames, sink_frames);
 
-		struct list_item *sink_list;
 		/*
 		 * In the case of playback DAI's with multiple sink buffers, compute the
 		 * minimum number of frames based on the DMA avail_bytes and the free
 		 * samples in all active sink buffers.
 		 */
-		list_for_item(sink_list, &dev->bsink_list) {
+		struct comp_buffer *sink;
+
+		sink = comp_dev_get_first_data_consumer(dev);
+		while (sink) {
 			struct comp_dev *sink_dev;
-			struct comp_buffer *sink;
 
-			sink = container_of(sink_list, struct comp_buffer, source_list);
 			sink_dev = sink->sink;
 
 			if (sink_dev && sink_dev->state == COMP_STATE_ACTIVE &&
@@ -1610,11 +1612,11 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 				sink_frames =
 					audio_stream_get_free_frames(&sink->stream);
 				frames = MIN(frames, sink_frames);
+
+				sink = comp_dev_get_next_data_consumer(dev, sink);
 			}
 		}
 	} else {
-		struct list_item *sink_list;
-
 		src_frames = avail_bytes / audio_stream_frame_bytes(&dd->dma_buffer->stream);
 
 		/*
@@ -1630,11 +1632,12 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 			 * minimum number of samples based on the DMA avail_bytes and the free
 			 * samples in all active sink buffers.
 			 */
-			list_for_item(sink_list, &dev->bsink_list) {
+			struct comp_buffer *sink;
+
+			sink = comp_dev_get_first_data_consumer(dev);
+			while (sink) {
 				struct comp_dev *sink_dev;
-				struct comp_buffer *sink;
 
-				sink = container_of(sink_list, struct comp_buffer, source_list);
 				sink_dev = sink->sink;
 
 				if (sink_dev && sink_dev->state == COMP_STATE_ACTIVE &&
@@ -1643,6 +1646,8 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 						audio_stream_get_free_frames(&sink->stream);
 					frames = MIN(frames, sink_frames);
 				}
+
+				sink = comp_dev_get_next_data_consumer(dev, sink);
 			}
 		}
 
diff --git a/src/audio/kpb.c b/src/audio/kpb.c
index 542984e88867..002d3deaf8c3 100644
--- a/src/audio/kpb.c
+++ b/src/audio/kpb.c
@@ -328,7 +328,6 @@ static int kpb_bind(struct comp_dev *dev, void *data)
 {
 	struct comp_data *kpb = comp_get_drvdata(dev);
 	struct ipc4_module_bind_unbind *bu;
-	struct list_item *blist;
 	int buf_id;
 	int ret = 0;
 
@@ -343,9 +342,9 @@ static int kpb_bind(struct comp_dev *dev, void *data)
 	 * (Detector/MicSel has one input pin). To properly connect KPB sink
 	 * with Detector source we're looking for buffer with id=0.
 	 */
+	struct comp_buffer *sink = comp_dev_get_first_data_consumer(dev);
 
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink = container_of(blist, struct comp_buffer, source_list);
+	while (sink) {
 		int sink_buf_id;
 
 		if (!sink->sink) {
@@ -361,6 +360,8 @@ static int kpb_bind(struct comp_dev *dev, void *data)
 			else
 				kpb->host_sink = sink;
 		}
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	return ret;
@@ -858,10 +859,9 @@ static int kpb_prepare(struct comp_dev *dev)
 	 * NOTE! We assume here that channel selector component device
 	 * is connected to the KPB sinks as well as host device.
 	 */
-	struct list_item *blist;
+	struct comp_buffer *sink = comp_dev_get_first_data_consumer(dev);
 
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink = container_of(blist, struct comp_buffer, source_list);
+	while (sink) {
 		enum sof_comp_type type;
 
 		if (!sink->sink) {
@@ -882,25 +882,27 @@ static int kpb_prepare(struct comp_dev *dev)
 		default:
 			break;
 		}
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 #else
 	/* Update number of sel_sink channels.
 	 * If OBS is not equal to IBS it means that KPB will work in micselector mode.
 	 */
 	if (kpb->ipc4_cfg.base_cfg.ibs != kpb->ipc4_cfg.base_cfg.obs) {
-		struct list_item *sink_list;
 		uint32_t sink_id;
 
-		list_for_item(sink_list, &dev->bsink_list) {
-			struct comp_buffer *sink =
-				container_of(sink_list, struct comp_buffer, source_list);
+		struct comp_buffer *sink = comp_dev_get_first_data_consumer(dev);
 
+		while (sink) {
 			sink_id = buf_get_id(sink);
 
 			if (sink_id == 0)
 				audio_stream_set_channels(&sink->stream, kpb->num_of_sel_mic);
 			else
 				audio_stream_set_channels(&sink->stream, kpb->config.channels);
+
+			sink = comp_dev_get_next_data_consumer(dev, sink);
 		}
 	}
 #endif /* CONFIG_IPC_MAJOR_4 */
diff --git a/src/audio/mux/mux.c b/src/audio/mux/mux.c
index 9ac9365297e6..3c0f9a8f5b38 100644
--- a/src/audio/mux/mux.c
+++ b/src/audio/mux/mux.c
@@ -233,7 +233,6 @@ static int demux_process(struct processing_module *mod,
 {
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct list_item *clist;
 	struct comp_buffer *sink;
 	struct audio_stream *sinks_stream[MUX_MAX_STREAMS] = { NULL };
 	struct mux_look_up *look_ups[MUX_MAX_STREAMS] = { NULL };
@@ -245,8 +244,8 @@ static int demux_process(struct processing_module *mod,
 	comp_dbg(dev, "demux_process()");
 
 	/* align sink streams with their respective configurations */
-	list_for_item(clist, &dev->bsink_list) {
-		sink = container_of(clist, struct comp_buffer, source_list);
+	sink = comp_dev_get_first_data_consumer(dev);
+	while (sink) {
 		if (sink->sink->state == dev->state) {
 			i = get_stream_index(dev, cd, buffer_pipeline_id(sink));
 			/* return if index wrong */
@@ -257,6 +256,8 @@ static int demux_process(struct processing_module *mod,
 			look_ups[i] = get_lookup_table(dev, cd, buffer_pipeline_id(sink));
 			sinks_stream[i] = &sink->stream;
 		}
+
+		sink = comp_dev_get_next_data_consumer(dev, sink);
 	}
 
 	/* if there are no sinks active, then sinks[] is also empty */
diff --git a/src/audio/selector/selector.c b/src/audio/selector/selector.c
index 506853043420..b32a0f801482 100644
--- a/src/audio/selector/selector.c
+++ b/src/audio/selector/selector.c
@@ -644,7 +644,6 @@ static void set_selector_params(struct processing_module *mod,
 	const struct sof_selector_ipc4_config *sel_cfg = &cd->sel_ipc4_cfg;
 	const struct ipc4_audio_format *out_fmt = NULL;
 	struct comp_buffer *src_buf;
-	struct list_item *sink_list;
 	int i;
 
 	if (cd->sel_ipc4_cfg.init_payload_fmt == IPC4_SEL_INIT_PAYLOAD_BASE_WITH_EXT)
@@ -664,13 +663,14 @@ static void set_selector_params(struct processing_module *mod,
 		params->chmap[i] = (out_fmt->ch_map >> i * 4) & 0xf;
 
 	/* update each sink format */
-	list_for_item(sink_list, &dev->bsink_list) {
-		struct comp_buffer *sink_buf =
-			container_of(sink_list, struct comp_buffer, source_list);
+	struct comp_buffer *sink_buf = comp_dev_get_first_data_consumer(dev);
 
+	while (sink_buf) {
 		ipc4_update_buffer_format(sink_buf, out_fmt);
 		audio_stream_set_channels(&sink_buf->stream, params->channels);
 		audio_stream_set_rate(&sink_buf->stream, params->rate);
+
+		sink_buf = comp_dev_get_next_data_consumer(dev, sink_buf);
 	}
 
 	/* update the source format
diff --git a/src/probe/probe.c b/src/probe/probe.c
index 721004555636..c7795cbf0203 100644
--- a/src/probe/probe.c
+++ b/src/probe/probe.c
@@ -1061,7 +1061,7 @@ static bool probe_purpose_needs_ext_dma(uint32_t purpose)
 static struct comp_buffer *ipc4_get_buffer(struct ipc_comp_dev *dev, probe_point_id_t probe_point)
 {
 	struct comp_buffer *buf;
-	struct list_item *sink_list, *source_list;
+	struct list_item *source_list;
 	unsigned int queue_id;
 
 	switch (probe_point.fields.type) {
@@ -1075,12 +1075,14 @@ static struct comp_buffer *ipc4_get_buffer(struct ipc_comp_dev *dev, probe_point
 		}
 		break;
 	case PROBE_TYPE_OUTPUT:
-		list_for_item(sink_list, &dev->cd->bsink_list) {
-			buf = container_of(sink_list, struct comp_buffer, source_list);
+		buf = comp_dev_get_first_data_consumer(dev->cd);
+		while (buf) {
 			queue_id = IPC4_SINK_QUEUE_ID(buf_get_id(buf));
 
 			if (queue_id == probe_point.fields.index)
 				return buf;
+
+			buf = comp_dev_get_next_data_consumer(dev->cd, buf);
 		}
 	}