diff --git a/router/dm.c b/router/dm.c
index 9c20b0c..b669bef 100644
--- a/router/dm.c
+++ b/router/dm.c
@@ -163,6 +163,20 @@ int dm_recv(int fd, void* data)
 		return dm_recv_raw(dm);
 }
 
+static ssize_t dm_send_flow(struct diag_client *dm, const void *ptr, size_t len,
+			    struct watch_flow *flow)
+{
+	if (!dm->enabled)
+		return 0;
+
+	if (dm->hdlc_encoded)
+		hdlc_enqueue_flow(&dm->outq, ptr, len, flow);
+	else
+		queue_push_flow(&dm->outq, ptr, len, flow);
+
+	return 0;
+}
+
 /**
  * dm_send() - enqueue message to DM
  * @dm:		dm to be receiving the message
@@ -171,23 +185,16 @@ int dm_recv(int fd, void* data)
  */
 ssize_t dm_send(struct diag_client *dm, const void *ptr, size_t len)
 {
-	if (!dm->enabled)
-		return 0;
-
-	if (dm->hdlc_encoded)
-		hdlc_enqueue(&dm->outq, ptr, len);
-	else
-		queue_push(&dm->outq, ptr, len);
-
-	return 0;
+	return dm_send_flow(dm, ptr, len, NULL);
 }
 
 /**
  * dm_broadcast() - send message to all registered DMs
  * @ptr:	pointer to raw message to be sent
  * @len:	length of message
+ * @flow:	flow control context for the peripheral
  */
-void dm_broadcast(const void *ptr, size_t len)
+void dm_broadcast(const void *ptr, size_t len, struct watch_flow *flow)
 {
 	struct diag_client *dm;
 	struct list_head *item;
@@ -195,7 +202,7 @@ void dm_broadcast(const void *ptr, size_t len)
 	list_for_each(item, &diag_clients) {
 		dm = container_of(item, struct diag_client, node);
 
-		dm_send(dm, ptr, len);
+		dm_send_flow(dm, ptr, len, flow);
 	}
 }
 
diff --git a/router/dm.h b/router/dm.h
index 17964c1..9c2195c 100644
--- a/router/dm.h
+++ b/router/dm.h
@@ -39,7 +39,7 @@ struct diag_client;
 struct diag_client *dm_add(const char *name, int in_fd, int out_fd, bool hdlc_encoded);
 int dm_recv(int fd, void* data);
 ssize_t dm_send(struct diag_client *dm, const void *ptr, size_t len);
-void dm_broadcast(const void *ptr, size_t len);
+void dm_broadcast(const void *ptr, size_t len, struct watch_flow *flow);
 void dm_enable(struct diag_client *dm);
 void dm_disable(struct diag_client *dm);
 
diff --git a/router/peripheral-qrtr.c b/router/peripheral-qrtr.c
index c2ce588..7a425dd 100644
--- a/router/peripheral-qrtr.c
+++ b/router/peripheral-qrtr.c
@@ -165,7 +165,7 @@ static int qrtr_cmd_recv(int fd, void *data)
 			break;
 		}
 
-		dm_broadcast(frame->payload, frame->length);
+		dm_broadcast(frame->payload, frame->length, NULL);
 		break;
 	case QRTR_TYPE_NEW_SERVER:
 		if (pkt.node == 0 && pkt.port == 0)
@@ -243,7 +243,7 @@ static int qrtr_data_recv(int fd, void *data)
 			fprintf(stderr, "non-HDLC frame is not truncated\n");
 			break;
 		}
-		dm_broadcast(frame->payload, frame->length);
+		dm_broadcast(frame->payload, frame->length, NULL);
 		break;
 	case QRTR_TYPE_BYE:
 		watch_remove_writeq(perif->data_fd);
diff --git a/router/peripheral-rpmsg.c b/router/peripheral-rpmsg.c
index b21d4d9..220fc89 100644
--- a/router/peripheral-rpmsg.c
+++ b/router/peripheral-rpmsg.c
@@ -104,7 +104,7 @@ static int diag_cmd_recv(int fd, void *data)
 		return 0;
 	}
 
-	dm_broadcast(frame->payload, frame->length);
+	dm_broadcast(frame->payload, frame->length, NULL);
 
 	return 0;
 }
@@ -127,7 +127,7 @@ static int diag_data_recv_hdlc(int fd, struct peripheral *peripheral)
 			if (!msg)
 				break;
 
-			dm_broadcast(msg, msglen);
+			dm_broadcast(msg, msglen, NULL);
 		}
 	}
 
@@ -144,7 +144,7 @@ static int diag_data_recv_raw(int fd, struct peripheral *peripheral)
 		if (n < 0)
 			return -errno;
 
-		dm_broadcast(buf, n);
+		dm_broadcast(buf, n, NULL);
 	}
 
 	/* Not reached */