CRAS: rstream - Add API to flush audio messages

Add API to let cras_rstream read and handle audio messages from client.
The pending flag in shm can be cleared if client replies the request.

BUG=b:68479798
TEST=unittest to be added.
Change-Id: Ic997bbd3a789af31ba23ab028f6605e3271183aa
Reviewed-on: https://chromium-review.googlesource.com/768776
Commit-Ready: Cheng-Yi Chiang <cychiang@chromium.org>
Tested-by: Cheng-Yi Chiang <cychiang@chromium.org>
Reviewed-by: Cheng-Yi Chiang <cychiang@chromium.org>

3 files changed, 176 insertions, 15 deletions

diff --git a/cras/src/server/cras_rstream.c b/cras/src/server/cras_rstream.c
index 7c3e1c06..337536c6 100644
--- a/cras/src/server/cras_rstream.c
+++ b/cras/src/server/cras_rstream.c
@@ -136,6 +136,75 @@ static void set_pending_reply(struct cras_rstream *stream)
 	cras_shm_set_callback_pending(&stream->shm, 1);
 }
 
+/*
+ * Clearing pending reply is only needed inside this module.
+ */
+static void clear_pending_reply(struct cras_rstream *stream)
+{
+	cras_shm_set_callback_pending(&stream->shm, 0);
+}
+
+/*
+ * Reads one response of audio request from client.
+ * Args:
+ *   stream[in]: A pointer to cras_rstream.
+ *   msg[out]: A pointer to audio_message to hold the message.
+ * Returns:
+ *   Number of bytes read from the socket.
+ *   A negative error code if read fails or the message from client
+ *   has errors.
+ */
+static int get_audio_request_reply(
+	const struct cras_rstream *stream, struct audio_message *msg)
+{
+	int rc;
+
+	rc = read(stream->fd, msg, sizeof(*msg));
+	if (rc < 0)
+		return -errno;
+	if (msg->error < 0)
+		return msg->error;
+	return rc;
+}
+
+/*
+ * Reads and handles one audio message from client.
+ * Returns:
+ *   Number of bytes read from the socket.
+ *   A negative error code if read fails or the message from client
+ *   has errors.
+ */
+static int read_and_handle_client_message(struct cras_rstream *stream) {
+
+	struct audio_message msg;
+	int rc;
+
+	rc = get_audio_request_reply(stream, &msg);
+	if (rc < 0) {
+		syslog(LOG_ERR, "Got error from client: rc: %d, msg.error: %d",
+		       rc, msg.error);
+		return rc;
+	}
+
+	/*
+	 * Got client reply that data in the input stream is captured.
+	 */
+	if (stream->direction == CRAS_STREAM_INPUT &&
+	    msg.id == AUDIO_MESSAGE_DATA_CAPTURED) {
+		clear_pending_reply(stream);
+	}
+
+	/*
+	 * Got client reply that data for output stream is ready in shm.
+	 */
+	if (stream->direction == CRAS_STREAM_OUTPUT &&
+	    msg.id == AUDIO_MESSAGE_DATA_READY) {
+		clear_pending_reply(stream);
+	}
+
+	return rc;
+}
+
 /* Exported functions */
 
 int cras_rstream_create(struct cras_rstream_config *config,
@@ -264,19 +333,6 @@ int cras_rstream_audio_ready(struct cras_rstream *stream, size_t count)
 	return rc;
 }
 
-int cras_rstream_get_audio_request_reply(const struct cras_rstream *stream)
-{
-	struct audio_message msg;
-	int rc;
-
-	rc = read(stream->fd, &msg, sizeof(msg));
-	if (rc < 0)
-		return -errno;
-	if (msg.error < 0)
-		return msg.error;
-	return 0;
-}
-
 void cras_rstream_dev_attach(struct cras_rstream *rstream,
 			     unsigned int dev_id,
 			     void *dev_ptr)
@@ -384,3 +440,24 @@ int cras_rstream_is_pending_reply(const struct cras_rstream *stream)
 {
 	return cras_shm_callback_pending(&stream->shm);
 }
+
+int cras_rstream_flush_old_audio_messages(struct cras_rstream *stream)
+{
+	struct pollfd pollfd;
+	int err;
+
+	if (!stream->fd)
+		return 0;
+
+	pollfd.fd = stream->fd;
+	pollfd.events = POLLIN;
+
+	do {
+		err = poll(&pollfd, 1, 0);
+		if (pollfd.revents & POLLIN) {
+			err = read_and_handle_client_message(stream);
+		}
+	} while (err > 0);
+
+	return 0;
+}
diff --git a/cras/src/server/cras_rstream.h b/cras/src/server/cras_rstream.h
index b76d6088..749738d7 100644
--- a/cras/src/server/cras_rstream.h
+++ b/cras/src/server/cras_rstream.h
@@ -270,8 +270,6 @@ int cras_rstream_request_audio(struct cras_rstream *stream,
 
 /* Tells a capture client that count frames are ready. */
 int cras_rstream_audio_ready(struct cras_rstream *stream, size_t count);
-/* Waits for the response to a request for audio. */
-int cras_rstream_get_audio_request_reply(const struct cras_rstream *stream);
 
 /* Let the rstream know when a device is added or removed. */
 void cras_rstream_dev_attach(struct cras_rstream *rstream,
@@ -332,4 +330,9 @@ int cras_rstream_get_mute(const struct cras_rstream *rstream);
  */
 int cras_rstream_is_pending_reply(const struct cras_rstream *stream);
 
+/*
+ * Reads any pending audio message from the socket.
+ */
+int cras_rstream_flush_old_audio_messages(struct cras_rstream *stream);
+
 #endif /* CRAS_RSTREAM_H_ */
diff --git a/cras/src/tests/rstream_unittest.cc b/cras/src/tests/rstream_unittest.cc
index 20ad117f..baf53efa 100644
--- a/cras/src/tests/rstream_unittest.cc
+++ b/cras/src/tests/rstream_unittest.cc
@@ -57,6 +57,20 @@ class RstreamTestSuite : public testing::Test {
           fmt1->num_channels == fmt2->num_channels;
     }
 
+    void stub_client_reply(enum CRAS_AUDIO_MESSAGE_ID id, int frames, int err) {
+      int rc;
+      struct audio_message aud_msg;
+      // Create a message.
+      aud_msg.id = id;
+      aud_msg.frames = frames;
+      aud_msg.error = err;
+
+      // Use socket fd to stub message from client.
+      rc = write(client_fd_, &aud_msg, sizeof(aud_msg));
+      EXPECT_EQ(sizeof(aud_msg), rc);
+      return;
+    }
+
     struct cras_audio_format fmt_;
     struct cras_rstream_config config_;
     int client_fd_;
@@ -232,6 +246,39 @@ TEST_F(RstreamTestSuite, OutputStreamIsPendingReply) {
   cras_rstream_destroy(s);
 }
 
+TEST_F(RstreamTestSuite, OutputStreamFlushMessages) {
+  struct cras_rstream *s;
+  int rc;
+  struct timespec ts;
+
+  rc = cras_rstream_create(&config_, &s);
+  EXPECT_EQ(0, rc);
+
+  // Not pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(0, rc);
+
+  // Request some data from client.
+  rc = cras_rstream_request_audio(s, &ts);
+  EXPECT_GT(rc, 0);
+
+  // Pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(1, rc);
+
+  // Client replies that data is ready.
+  stub_client_reply(AUDIO_MESSAGE_DATA_READY, 10, 0);
+
+  // Read messages.
+  cras_rstream_flush_old_audio_messages(s);
+
+  // NOT Pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(0, rc);
+
+  cras_rstream_destroy(s);
+}
+
 TEST_F(RstreamTestSuite, InputStreamIsPendingReply) {
   struct cras_rstream *s;
   int rc;
@@ -256,6 +303,40 @@ TEST_F(RstreamTestSuite, InputStreamIsPendingReply) {
   cras_rstream_destroy(s);
 }
 
+TEST_F(RstreamTestSuite, InputStreamFlushMessages) {
+  struct cras_rstream *s;
+  int rc;
+
+  config_.direction = CRAS_STREAM_INPUT;
+
+  rc = cras_rstream_create(&config_, &s);
+  EXPECT_EQ(0, rc);
+
+  // Not pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(0, rc);
+
+  // Some data is ready. Sends it to client.
+  rc = cras_rstream_audio_ready(s, 10);
+  EXPECT_GT(rc, 0);
+
+  // Pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(1, rc);
+
+  // Client replies that data is captured.
+  stub_client_reply(AUDIO_MESSAGE_DATA_CAPTURED, 10, 0);
+
+  // Read messages.
+  cras_rstream_flush_old_audio_messages(s);
+
+  // NOT Pending reply.
+  rc = cras_rstream_is_pending_reply(s);
+  EXPECT_EQ(0, rc);
+
+  cras_rstream_destroy(s);
+}
+
 }  //  namespace
 
 int main(int argc, char **argv) {