diff --git a/src/audio/buffers/comp_buffer.c b/src/audio/buffers/comp_buffer.c
index 3d5896343822..02b44e5dcdce 100644
--- a/src/audio/buffers/comp_buffer.c
+++ b/src/audio/buffers/comp_buffer.c
@@ -212,8 +212,8 @@ static struct comp_buffer *buffer_alloc_struct(void *stream_addr, size_t size, u
 	audio_stream_set_underrun(&buffer->stream, !!(flags & SOF_BUF_UNDERRUN_PERMITTED));
 	audio_stream_set_overrun(&buffer->stream, !!(flags & SOF_BUF_OVERRUN_PERMITTED));
 
-	list_init(&buffer->source_list);
-	list_init(&buffer->sink_list);
+	comp_buffer_reset_source_list(buffer);
+	comp_buffer_reset_sink_list(buffer);
 
 	return buffer;
 }
diff --git a/src/audio/module_adapter/module_adapter.c b/src/audio/module_adapter/module_adapter.c
index 5586d2fdfde7..f8eebb609a35 100644
--- a/src/audio/module_adapter/module_adapter.c
+++ b/src/audio/module_adapter/module_adapter.c
@@ -230,7 +230,7 @@ int module_adapter_prepare(struct comp_dev *dev)
 	/* Get period_bytes first on prepare(). At this point it is guaranteed that the stream
 	 * parameter from sink buffer is settled, and still prior to all references to period_bytes.
 	 */
-	sink = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sink = comp_dev_get_first_data_consumer(dev);
 
 	mod->period_bytes = audio_stream_period_bytes(&sink->stream, dev->frames);
 	comp_dbg(dev, "module_adapter_prepare(): got period_bytes = %u", mod->period_bytes);
@@ -621,17 +621,14 @@ static void module_adapter_process_output(struct comp_dev *dev)
 
 	/* copy from all output local buffers to sink buffers */
 	i = 0;
-	list_for_item(blist, &dev->bsink_list) {
-		struct list_item *_blist;
+	comp_dev_for_each_consumer(dev, sink) {
 		int j = 0;
 
-		list_for_item(_blist, &mod->raw_data_buffers_list) {
+		list_for_item(blist, &mod->raw_data_buffers_list) {
 			if (i == j) {
 				struct comp_buffer *source;
 
-				sink = container_of(blist, struct comp_buffer, source_list);
-				source = container_of(_blist, struct comp_buffer, buffers_list);
-
+				source = container_of(blist, struct comp_buffer, buffers_list);
 				module_copy_samples(dev, source, sink,
 						    mod->output_buffers[i].size);
 
@@ -756,7 +753,7 @@ static int module_adapter_audio_stream_copy_1to1(struct comp_dev *dev)
 	/* Note: Source buffer state is not checked to enable mixout to generate zero
 	 * PCM codes when source is not active.
 	 */
-	if (mod->sink_comp_buffer->sink->state == dev->state)
+	if (comp_buffer_get_sink_state(mod->sink_comp_buffer) == dev->state)
 		num_output_buffers = 1;
 
 	ret = module_process_legacy(mod, mod->input_buffers, 1,
@@ -783,10 +780,11 @@ static int module_adapter_audio_stream_type_copy(struct comp_dev *dev)
 {
 	struct comp_buffer *sources[PLATFORM_MAX_STREAMS];
 	struct comp_buffer *sinks[PLATFORM_MAX_STREAMS];
+	struct comp_buffer *sink;
+	struct comp_buffer *source;
 	struct processing_module *mod = comp_mod(dev);
-	struct list_item *blist;
 	uint32_t num_input_buffers, num_output_buffers;
-	int ret, i = 0;
+	int ret, i;
 
 	/* handle special case of HOST/DAI type components */
 	if (dev->ipc_config.type == SOF_COMP_HOST || dev->ipc_config.type == SOF_COMP_DAI)
@@ -796,12 +794,9 @@ static int module_adapter_audio_stream_type_copy(struct comp_dev *dev)
 		return module_adapter_audio_stream_copy_1to1(dev);
 
 	/* acquire all sink and source buffers */
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink;
-
-		sink = container_of(blist, struct comp_buffer, source_list);
+	i = 0;
+	comp_dev_for_each_consumer(dev, sink)
 		sinks[i++] = sink;
-	}
 	num_output_buffers = i;
 	if (num_output_buffers > mod->max_sinks) {
 		comp_err(dev, "Invalid number of sinks %d\n", num_output_buffers);
@@ -809,12 +804,8 @@ static int module_adapter_audio_stream_type_copy(struct comp_dev *dev)
 	}
 
 	i = 0;
-	list_for_item(blist, &dev->bsource_list) {
-		struct comp_buffer *source;
-
-		source = container_of(blist, struct comp_buffer, sink_list);
+	comp_dev_for_each_producer(dev, source)
 		sources[i++] = source;
-	}
 	num_input_buffers = i;
 	if (num_input_buffers > mod->max_sources) {
 		comp_err(dev, "Invalid number of sources %d\n", num_input_buffers);
@@ -824,11 +815,11 @@ static int module_adapter_audio_stream_type_copy(struct comp_dev *dev)
 	/* setup active input/output buffers for processing */
 	if (num_output_buffers == 1) {
 		module_single_sink_setup(dev, sources, sinks);
-		if (sinks[0]->sink->state != dev->state)
+		if (comp_buffer_get_sink_state(sinks[0]) != dev->state)
 			num_output_buffers = 0;
 	} else if (num_input_buffers == 1) {
 		module_single_source_setup(dev, sources, sinks);
-		if (sources[0]->source->state != dev->state) {
+		if (comp_buffer_get_source_state(sources[0]) != dev->state) {
 			num_input_buffers = 0;
 		}
 	} else {
@@ -910,15 +901,13 @@ static int module_adapter_copy_ring_buffers(struct comp_dev *dev)
 	 * This is an adapter, to be removed when pipeline2.0 is ready
 	 */
 	struct processing_module *mod = comp_mod(dev);
-	struct list_item *blist;
+	struct comp_buffer *buffer;
 	int err;
 
-	list_for_item(blist, &dev->bsource_list) {
+	comp_dev_for_each_producer(dev, buffer) {
 		/* input - we need to copy data from audio_stream (as source)
 		 * to ring_buffer (as sink)
 		 */
-		struct comp_buffer *buffer =
-				container_of(blist, struct comp_buffer, sink_list);
 		err = audio_buffer_sync_secondary_buffer(&buffer->audio_buffer, UINT_MAX);
 
 		if (err) {
@@ -930,7 +919,7 @@ static int module_adapter_copy_ring_buffers(struct comp_dev *dev)
 	if (mod->dp_startup_delay)
 		return 0;
 
-	list_for_item(blist, &dev->bsink_list) {
+	comp_dev_for_each_consumer(dev, buffer) {
 		/* output - we need to copy data from ring_buffer (as source)
 		 * to audio_stream (as sink)
 		 *
@@ -943,8 +932,6 @@ static int module_adapter_copy_ring_buffers(struct comp_dev *dev)
 		 *
 		 * FIX: copy only the following module's IBS in each LL cycle
 		 */
-		struct comp_buffer *buffer =
-				container_of(blist, struct comp_buffer, source_list);
 		struct sof_source *following_mod_data_source =
 				audio_buffer_get_source(&buffer->audio_buffer);
 
@@ -1021,14 +1008,12 @@ static int module_adapter_raw_data_type_copy(struct comp_dev *dev)
 	}
 
 	/* copy source samples into input buffer */
-	list_for_item(blist, &dev->bsource_list) {
+	comp_dev_for_each_producer(dev, source) {
 		uint32_t bytes_to_process;
 		int frames, source_frame_bytes;
 
-		source = container_of(blist, struct comp_buffer, sink_list);
-
 		/* check if the source dev is in the same state as the dev */
-		if (!source->source || source->source->state != dev->state)
+		if (comp_buffer_get_source_state(source) != dev->state)
 			continue;
 
 		frames = MIN(min_free_frames,
@@ -1061,10 +1046,7 @@ static int module_adapter_raw_data_type_copy(struct comp_dev *dev)
 
 	i = 0;
 	/* consume from all input buffers */
-	list_for_item(blist, &dev->bsource_list) {
-
-		source = container_of(blist, struct comp_buffer, sink_list);
-
+	comp_dev_for_each_producer(dev, source) {
 		comp_update_buffer_consume(source, mod->input_buffers[i].consumed);
 
 		bzero((__sparse_force void *)mod->input_buffers[i].data, size);
diff --git a/src/audio/module_adapter/module_adapter_ipc3.c b/src/audio/module_adapter/module_adapter_ipc3.c
index 8657c2d41072..b54b0f41388e 100644
--- a/src/audio/module_adapter/module_adapter_ipc3.c
+++ b/src/audio/module_adapter/module_adapter_ipc3.c
@@ -121,8 +121,7 @@ void module_adapter_check_data(struct processing_module *mod, struct comp_dev *d
 	 */
 	if (IS_PROCESSING_MODE_AUDIO_STREAM(mod) && mod->num_of_sources == 1 &&
 	    mod->num_of_sinks == 1) {
-		mod->source_comp_buffer = list_first_item(&dev->bsource_list,
-							  struct comp_buffer, sink_list);
+		mod->source_comp_buffer = comp_dev_get_first_data_producer(dev);
 		mod->sink_comp_buffer = sink;
 		mod->stream_copy_single_to_single = true;
 	}
@@ -134,11 +133,11 @@ void module_adapter_set_params(struct processing_module *mod, struct sof_ipc_str
 
 static int module_source_state_count(struct comp_dev *dev, uint32_t state)
 {
-	struct list_item *blist;
 	int count = 0;
+	struct comp_buffer *source;
 
 	/* count source with state == status */
-	list_for_item(blist, &dev->bsource_list) {
+	comp_dev_for_each_producer(dev, source)
 		/*
 		 * FIXME: this is racy, state can be changed by another core.
 		 * This is implicitly protected by serialised IPCs. Even when
@@ -146,12 +145,8 @@ static int module_source_state_count(struct comp_dev *dev, uint32_t state)
 		 * not be sent until the thread has processed and replied to the
 		 * current one.
 		 */
-		struct comp_buffer *source = container_of(blist, struct comp_buffer,
-							  sink_list);
-
-		if (source->source && source->source->state == state)
+		if (comp_buffer_get_source_state(source) == state)
 			count++;
-	}
 
 	return count;
 }
@@ -324,25 +319,21 @@ int module_adapter_cmd(struct comp_dev *dev, int cmd, void *data, int max_data_s
 int module_adapter_sink_src_prepare(struct comp_dev *dev)
 {
 	struct processing_module *mod = comp_mod(dev);
-	struct list_item *blist;
+	struct comp_buffer *sink_buffer;
+	struct comp_buffer *source_buffer;
 	int ret;
 	int i;
 
 	/* acquire all sink and source buffers, get handlers to sink/source API */
 	i = 0;
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink_buffer =
-				container_of(blist, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink_buffer) {
 		mod->sinks[i] = audio_buffer_get_sink(&sink_buffer->audio_buffer);
 		i++;
 	}
 	mod->num_of_sinks = i;
 
 	i = 0;
-	list_for_item(blist, &dev->bsource_list) {
-		struct comp_buffer *source_buffer =
-				container_of(blist, struct comp_buffer, sink_list);
-
+	comp_dev_for_each_producer(dev, source_buffer) {
 		mod->sources[i] = audio_buffer_get_source(&source_buffer->audio_buffer);
 		i++;
 	}
diff --git a/src/audio/module_adapter/module_adapter_ipc4.c b/src/audio/module_adapter/module_adapter_ipc4.c
index 060cef446f73..00ba4696f639 100644
--- a/src/audio/module_adapter/module_adapter_ipc4.c
+++ b/src/audio/module_adapter/module_adapter_ipc4.c
@@ -187,24 +187,20 @@ EXPORT_SYMBOL(module_adapter_get_attribute);
 static bool module_adapter_multi_sink_source_prepare(struct comp_dev *dev)
 {
 	struct processing_module *mod = comp_mod(dev);
-	struct list_item *blist;
+	struct comp_buffer *sink_buffer;
+	struct comp_buffer *source_buffer;
 	int i;
 
 	/* acquire all sink and source buffers, get handlers to sink/source API */
 	i = 0;
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink_buffer =
-				container_of(blist, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink_buffer) {
 		mod->sinks[i] = audio_buffer_get_sink(&sink_buffer->audio_buffer);
 		i++;
 	}
 	mod->num_of_sinks = i;
 
 	i = 0;
-	list_for_item(blist, &dev->bsource_list) {
-		struct comp_buffer *source_buffer =
-				container_of(blist, struct comp_buffer, sink_list);
-
+	comp_dev_for_each_producer(dev, source_buffer) {
 		mod->sources[i] = audio_buffer_get_source(&source_buffer->audio_buffer);
 		i++;
 	}
@@ -216,9 +212,8 @@ static bool module_adapter_multi_sink_source_prepare(struct comp_dev *dev)
 		return true;
 
 	/* re-assign the source/sink modules */
-	mod->sink_comp_buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
-	mod->source_comp_buffer = list_first_item(&dev->bsource_list,
-						  struct comp_buffer, sink_list);
+	mod->sink_comp_buffer = comp_dev_get_first_data_consumer(dev);
+	mod->source_comp_buffer = comp_dev_get_first_data_producer(dev);
 
 	return false;
 }
diff --git a/src/audio/pipeline/pipeline-graph.c b/src/audio/pipeline/pipeline-graph.c
index f30731fa4874..15ae9d84c2eb 100644
--- a/src/audio/pipeline/pipeline-graph.c
+++ b/src/audio/pipeline/pipeline-graph.c
@@ -167,9 +167,9 @@ static void buffer_set_comp(struct comp_buffer *buffer, struct comp_dev *comp,
 			    int dir)
 {
 	if (dir == PPL_CONN_DIR_COMP_TO_BUFFER)
-		buffer->source = comp;
+		comp_buffer_set_source_component(buffer, comp);
 	else
-		buffer->sink = comp;
+		comp_buffer_set_sink_component(buffer, comp);
 }
 
 int pipeline_connect(struct comp_dev *comp, struct comp_buffer *buffer,
diff --git a/src/ipc/ipc-helper.c b/src/ipc/ipc-helper.c
index fe2ec7541b81..54ebdf77bcbe 100644
--- a/src/ipc/ipc-helper.c
+++ b/src/ipc/ipc-helper.c
@@ -115,10 +115,8 @@ static void comp_update_params(uint32_t flag,
 int comp_verify_params(struct comp_dev *dev, uint32_t flag,
 		       struct sof_ipc_stream_params *params)
 {
-	struct list_item *buffer_list;
 	struct list_item *source_list;
 	struct list_item *sink_list;
-	struct list_item *clist;
 	struct comp_buffer *sinkb;
 	struct comp_buffer *buf;
 	int dir = dev->direction;
@@ -136,13 +134,9 @@ int comp_verify_params(struct comp_dev *dev, uint32_t flag,
 	 */
 	if (list_is_empty(source_list) != list_is_empty(sink_list)) {
 		if (list_is_empty(sink_list))
-			buf = list_first_item(source_list,
-					      struct comp_buffer,
-					      sink_list);
+			buf = comp_dev_get_first_data_producer(dev);
 		else
-			buf = list_first_item(sink_list,
-					      struct comp_buffer,
-					      source_list);
+			buf = comp_dev_get_first_data_consumer(dev);
 
 		/* update specific pcm parameter with buffer parameter if
 		 * specific flag is set.
@@ -160,18 +154,22 @@ int comp_verify_params(struct comp_dev *dev, uint32_t flag,
 		/* for other components we iterate over all downstream buffers
 		 * (for playback) or upstream buffers (for capture).
 		 */
-		buffer_list = comp_buffer_list(dev, dir);
-
-		list_for_item(clist, buffer_list) {
-			buf = buffer_from_list(clist, dir);
-			comp_update_params(flag, params, buf);
-			buffer_set_params(buf, params, BUFFER_UPDATE_FORCE);
+		if (dir == PPL_DIR_DOWNSTREAM) {
+			comp_dev_for_each_consumer(dev, buf) {
+				comp_update_params(flag, params, buf);
+				buffer_set_params(buf, params,
+						  BUFFER_UPDATE_FORCE);
+			}
+		} else {
+			comp_dev_for_each_producer(dev, buf) {
+				comp_update_params(flag, params, buf);
+				buffer_set_params(buf, params,
+						  BUFFER_UPDATE_FORCE);
+			}
 		}
 
 		/* fetch sink buffer in order to calculate period frames */
-		sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-					source_list);
-
+		sinkb = comp_dev_get_first_data_consumer(dev);
 		component_set_nearest_period_frames(dev, audio_stream_get_rate(&sinkb->stream));
 	}
 
@@ -263,7 +261,8 @@ int ipc_pipeline_complete(struct ipc *ipc, uint32_t comp_id)
 int ipc_comp_free(struct ipc *ipc, uint32_t comp_id)
 {
 	struct ipc_comp_dev *icd;
-	struct list_item *clist, *tmp;
+	struct comp_buffer *buffer;
+	struct comp_buffer *safe;
 	uint32_t flags;
 
 	/* check whether component exists */
@@ -305,21 +304,20 @@ int ipc_comp_free(struct ipc *ipc, uint32_t comp_id)
 	}
 
 	irq_local_disable(flags);
-	list_for_item_safe(clist, tmp, &icd->cd->bsource_list) {
-		struct comp_buffer *buffer = container_of(clist, struct comp_buffer, sink_list);
-
-		buffer->sink = NULL;
+	comp_dev_for_each_producer_safe(icd->cd, buffer, safe)
+	{
+		comp_buffer_set_sink_component(buffer, NULL);
 		/* This breaks the list, but we anyway delete all buffers */
-		list_init(clist);
+		comp_buffer_reset_sink_list(buffer);
 	}
 
-	list_for_item_safe(clist, tmp, &icd->cd->bsink_list) {
-		struct comp_buffer *buffer = container_of(clist, struct comp_buffer, source_list);
-
-		buffer->source = NULL;
+	comp_dev_for_each_consumer_safe(icd->cd, buffer, safe)
+	{
+		comp_buffer_set_source_component(buffer, NULL);
 		/* This breaks the list, but we anyway delete all buffers */
-		list_init(clist);
+		comp_buffer_reset_source_list(buffer);
 	}
+
 	irq_local_enable(flags);
 
 	/* free component and remove from list */
diff --git a/src/ipc/ipc3/handler.c b/src/ipc/ipc3/handler.c
index 12f2ba3a34a4..eb812e58fce2 100644
--- a/src/ipc/ipc3/handler.c
+++ b/src/ipc/ipc3/handler.c
@@ -137,7 +137,7 @@ struct ipc_cmd_hdr *mailbox_validate(void)
 /* check if a pipeline is hostless when walking downstream */
 static bool is_hostless_downstream(struct comp_dev *current)
 {
-	struct list_item *clist;
+	struct comp_buffer *buffer;
 
 	/* check if current is a HOST comp */
 	if (current->ipc_config.type == SOF_COMP_HOST ||
@@ -145,22 +145,18 @@ static bool is_hostless_downstream(struct comp_dev *current)
 		return false;
 
 	/* check if the pipeline has a HOST comp downstream */
-	list_for_item(clist, &current->bsink_list) {
-		struct comp_buffer *buffer;
-
-		buffer = container_of(clist, struct comp_buffer, source_list);
-
+	comp_dev_for_each_consumer(current, buffer) {
 		/* don't go downstream if this component is not connected */
-		if (!buffer->sink)
+		if (!comp_buffer_get_sink_component(buffer))
 			continue;
 
 		/* dont go downstream if this comp belongs to another pipe */
-		if (buffer->sink->ipc_config.pipeline_id !=
+		if (comp_buffer_get_sink_component(buffer)->ipc_config.pipeline_id !=
 			current->ipc_config.pipeline_id)
 			continue;
 
 		/* return if there's a host comp downstream */
-		if (!is_hostless_downstream(buffer->sink))
+		if (!is_hostless_downstream(comp_buffer_get_sink_component(buffer)))
 			return false;
 	}
 
@@ -170,7 +166,7 @@ static bool is_hostless_downstream(struct comp_dev *current)
 /* check if a pipeline is hostless when walking upstream */
 static bool is_hostless_upstream(struct comp_dev *current)
 {
-	struct list_item *clist;
+	struct comp_buffer *buffer;
 
 	/* check if current is a HOST comp */
 	if (current->ipc_config.type == SOF_COMP_HOST ||
@@ -178,22 +174,18 @@ static bool is_hostless_upstream(struct comp_dev *current)
 		return false;
 
 	/* check if the pipeline has a HOST comp upstream */
-	list_for_item(clist, &current->bsource_list) {
-		struct comp_buffer *buffer;
-
-		buffer = container_of(clist, struct comp_buffer, sink_list);
-
+	comp_dev_for_each_producer(current, buffer) {
 		/* don't go upstream if this component is not connected */
-		if (!buffer->source)
+		if (!comp_buffer_get_source_component(buffer))
 			continue;
 
 		/* dont go upstream if this comp belongs to another pipeline */
-		if (buffer->source->ipc_config.pipeline_id !=
+		if (comp_buffer_get_source_component(buffer)->ipc_config.pipeline_id !=
 		    current->ipc_config.pipeline_id)
 			continue;
 
 		/* return if there is a host comp upstream */
-		if (!is_hostless_upstream(buffer->source))
+		if (!is_hostless_upstream(comp_buffer_get_source_component(buffer)))
 			return false;
 	}
 
diff --git a/src/ipc/ipc3/helper.c b/src/ipc/ipc3/helper.c
index 678d3306e4b1..adb4d7ec1f16 100644
--- a/src/ipc/ipc3/helper.c
+++ b/src/ipc/ipc3/helper.c
@@ -535,15 +535,15 @@ int ipc_buffer_free(struct ipc *ipc, uint32_t buffer_id)
 			continue;
 
 		/* check comp state if sink and source are valid */
-		if (ibd->cb->sink == icd->cd) {
-			sink = ibd->cb->sink;
-			if (ibd->cb->sink->state != COMP_STATE_READY)
+		if (comp_buffer_get_sink_component(ibd->cb) == icd->cd) {
+			sink = comp_buffer_get_sink_component(ibd->cb);
+			if (comp_buffer_get_sink_state(ibd->cb) != COMP_STATE_READY)
 				sink_active = true;
 		}
 
-		if (ibd->cb->source == icd->cd) {
-			source = ibd->cb->source;
-			if (ibd->cb->source->state != COMP_STATE_READY)
+		if (comp_buffer_get_source_component(ibd->cb) == icd->cd) {
+			source = comp_buffer_get_source_component(ibd->cb);
+			if (comp_buffer_get_source_state(ibd->cb) != COMP_STATE_READY)
 				source_active = true;
 		}
 	}
diff --git a/src/ipc/ipc4/helper.c b/src/ipc/ipc4/helper.c
index 0dfb4bc73fc2..18309a39a1a3 100644
--- a/src/ipc/ipc4/helper.c
+++ b/src/ipc/ipc4/helper.c
@@ -296,16 +296,13 @@ static int ipc_pipeline_module_free(uint32_t pipeline_id)
 
 	icd = ipc_get_comp_by_ppl_id(ipc, COMP_TYPE_COMPONENT, pipeline_id, IPC_COMP_ALL);
 	while (icd) {
-		struct list_item *list, *_list;
 		struct comp_buffer *buffer;
+		struct comp_buffer *safe;
 
 		/* free sink buffer allocated by current component in bind function */
-		list_for_item_safe(list, _list, &icd->cd->bsink_list) {
-			struct comp_dev *sink;
-
-			buffer = container_of(list, struct comp_buffer, source_list);
-			pipeline_disconnect(icd->cd, buffer, PPL_CONN_DIR_COMP_TO_BUFFER);
-			sink = buffer->sink;
+		comp_dev_for_each_consumer_safe(icd->cd, buffer, safe)
+		{
+			struct comp_dev *sink = comp_buffer_get_sink_component(buffer);
 
 			/* free the buffer only when the sink module has also been disconnected */
 			if (!sink)
@@ -313,12 +310,10 @@ static int ipc_pipeline_module_free(uint32_t pipeline_id)
 		}
 
 		/* free source buffer allocated by current component in bind function */
-		list_for_item_safe(list, _list, &icd->cd->bsource_list) {
-			struct comp_dev *source;
-
-			buffer = container_of(list, struct comp_buffer, sink_list);
+		comp_dev_for_each_producer_safe(icd->cd, buffer, safe)
+		{
 			pipeline_disconnect(icd->cd, buffer, PPL_CONN_DIR_BUFFER_TO_COMP);
-			source = buffer->source;
+			struct comp_dev *source = comp_buffer_get_source_component(buffer);
 
 			/* free the buffer only when the source module has also been disconnected */
 			if (!source)
@@ -677,8 +672,8 @@ int ipc_comp_disconnect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 {
 	struct ipc4_module_bind_unbind *bu;
 	struct comp_buffer *buffer = NULL;
+	struct comp_buffer *buf;
 	struct comp_dev *src, *sink;
-	struct list_item *sink_list;
 	uint32_t src_id, sink_id, buffer_id;
 	uint32_t flags = 0;
 	int ret, ret1;
@@ -708,10 +703,8 @@ int ipc_comp_disconnect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 		return ipc4_process_on_core(src->ipc_config.core, false);
 
 	buffer_id = IPC4_COMP_ID(bu->extension.r.src_queue, bu->extension.r.dst_queue);
-	list_for_item(sink_list, &src->bsink_list) {
-		struct comp_buffer *buf = container_of(sink_list, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(src, buf) {
 		bool found = buf_get_id(buf) == buffer_id;
-
 		if (found) {
 			buffer = buf;
 			break;
@@ -851,9 +844,9 @@ static int ipc4_update_comps_direction(struct ipc *ipc, uint32_t ppl_id)
 		if (list_is_empty(&icd->cd->bsource_list))
 			continue;
 
-		src_buf = list_first_item(&icd->cd->bsource_list, struct comp_buffer, sink_list);
-		if (src_buf->source->direction_set) {
-			icd->cd->direction = src_buf->source->direction;
+		src_buf = comp_dev_get_first_data_producer(icd->cd);
+		if (comp_buffer_get_source_component(src_buf)->direction_set) {
+			icd->cd->direction = comp_buffer_get_source_component(src_buf)->direction;
 			icd->cd->direction_set = true;
 			continue;
 		}