CRAS: Add cras_playback_rclient for playback clients

Add cras_playback_rclient which handles message from `.cras_playback`.
Clients connect from the socket can only open playback streams.

Add valid_stream_id function to check if the stream_id is valid.

BUG=chromium:937765
TEST=unittests
TEST=Modify cras_tests to test `.cras_playback`.
TEST=Use cras_test_client and USE Chrome to verify that legacy path still
     works.

Change-Id: I9a1a36d0bf7b232ad0ca1800addd7522c2bccbed
Reviewed-on: https://chromium-review.googlesource.com/1679958
Tested-by: Chih-Yang Hsia <paulhsia@chromium.org>
Commit-Ready: Chih-Yang Hsia <paulhsia@chromium.org>
Legacy-Commit-Queue: Commit Bot <commit-bot@chromium.org>
Reviewed-by: Cheng-Yi Chiang <cychiang@chromium.org>

10 files changed, 568 insertions, 18 deletions

diff --git a/cras/src/Makefile.am b/cras/src/Makefile.am
index 43a94437..6738231f 100644
--- a/cras/src/Makefile.am
+++ b/cras/src/Makefile.am
@@ -141,6 +141,7 @@ cras_server_SOURCES = \
 	server/cras_rclient.c \
 	server/cras_rclient_util.c \
 	server/cras_control_rclient.c \
+	server/cras_playback_rclient.c \
 	server/cras_rstream.c \
 	server/cras_server_metrics.c \
 	server/cras_system_state.c \
@@ -378,6 +379,7 @@ TESTS = \
 	ramp_unittest \
 	rate_estimator_unittest \
 	control_rclient_unittest \
+	playback_rclient_unittest \
 	rstream_unittest \
 	shm_unittest \
 	server_metrics_unittest \
@@ -865,6 +867,12 @@ control_rclient_unittest_CPPFLAGS = $(COMMON_CPPFLAGS) \
 	-I$(top_srcdir)/src/server $(CRAS_UT_TMPDIR_CFLAGS)
 control_rclient_unittest_LDADD = -lgtest -lpthread
 
+playback_rclient_unittest_SOURCES = tests/playback_rclient_unittest.cc
+playback_rclient_unittest_CPPFLAGS = $(COMMON_CPPFLAGS) \
+	-I$(top_srcdir)/src/common \
+	-I$(top_srcdir)/src/server $(CRAS_UT_TMPDIR_CFLAGS)
+playback_rclient_unittest_LDADD = -lgtest -lpthread
+
 rstream_unittest_SOURCES = tests/rstream_unittest.cc server/cras_rstream.c \
 	common/cras_shm.c $(CRAS_SELINUX_UNITTEST_SOURCES)
 rstream_unittest_CPPFLAGS = $(COMMON_CPPFLAGS) -I$(top_srcdir)/src/common \
diff --git a/cras/src/common/cras_config.h b/cras/src/common/cras_config.h
index c1731c73..6485df69 100644
--- a/cras/src/common/cras_config.h
+++ b/cras/src/common/cras_config.h
@@ -12,6 +12,7 @@
 #define CRAS_CLIENT_RT_THREAD_PRIORITY 10
 #define CRAS_CLIENT_NICENESS_LEVEL -10
 #define CRAS_SOCKET_FILE ".cras_socket"
+#define CRAS_PLAYBACK_SOCKET_FILE ".cras_playback"
 
 /* CRAS_CONFIG_FILE_DIR is defined as $sysconfdir/cras by the configure
    script. */
diff --git a/cras/src/common/cras_types.h b/cras/src/common/cras_types.h
index bce677ab..fbabff39 100644
--- a/cras/src/common/cras_types.h
+++ b/cras/src/common/cras_types.h
@@ -9,6 +9,7 @@
 #ifndef CRAS_TYPES_H_
 #define CRAS_TYPES_H_
 
+#include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
 
@@ -507,6 +508,12 @@ static inline cras_stream_id_t cras_get_stream_id(uint16_t client_id,
 	return (cras_stream_id_t)(((client_id & 0x0000ffff) << 16) |
 				  (stream_id & 0x0000ffff));
 }
+/* Verify if the stream_id fits the given client_id */
+static inline bool cras_valid_stream_id(cras_stream_id_t stream_id,
+					uint16_t client_id)
+{
+	return ((stream_id >> 16) ^ client_id) == 0;
+}
 
 enum CRAS_NODE_TYPE {
 	/* These value can be used for output nodes. */
diff --git a/cras/src/server/cras_control_rclient.c b/cras/src/server/cras_control_rclient.c
index d4931e77..7bf94e3f 100644
--- a/cras/src/server/cras_control_rclient.c
+++ b/cras/src/server/cras_control_rclient.c
@@ -44,7 +44,7 @@ static int handle_client_stream_connect(struct cras_rclient *client,
 	/* check the aud_fd is valid. */
 	if (aud_fd < 0) {
 		syslog(LOG_ERR, "Invalid fd in stream connect.\n");
-		rc = -EINVAL;
+		rc = -EBADF;
 		goto reply_err;
 	}
 	/* When full, getting an error is preferable to blocking. */
@@ -342,8 +342,6 @@ static int direction_valid(enum CRAS_STREAM_DIRECTION direction)
 	       direction != CRAS_STREAM_UNDEFINED;
 }
 
-#define MSG_LEN_VALID(msg, type) ((msg)->length >= sizeof(type))
-
 /* Entry point for handling a message from the client.  Called from the main
  * server context. */
 static int ccr_handle_message_from_client(struct cras_rclient *client,
@@ -365,7 +363,7 @@ static int ccr_handle_message_from_client(struct cras_rclient *client,
 			       "Message %d should not have fd attached.",
 			       msg->id);
 			close(fd);
-			return -1;
+			return -EINVAL;
 		}
 		break;
 	}
diff --git a/cras/src/server/cras_playback_rclient.c b/cras/src/server/cras_playback_rclient.c
new file mode 100644
index 00000000..5b5ecdfd
--- /dev/null
+++ b/cras/src/server/cras_playback_rclient.c
@@ -0,0 +1,203 @@
+/* Copyright 2019 The Chromium OS Authors. All rights reserved.
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE file.
+ */
+
+#include <syslog.h>
+
+#include "cras_iodev_list.h"
+#include "cras_messages.h"
+#include "cras_observer.h"
+#include "cras_rclient.h"
+#include "cras_rclient_util.h"
+#include "cras_rstream.h"
+#include "cras_server_metrics.h"
+#include "cras_system_state.h"
+#include "cras_types.h"
+#include "cras_util.h"
+#include "stream_list.h"
+
+/* Handles a message from the client to connect a new stream */
+static int handle_client_stream_connect(struct cras_rclient *client,
+					const struct cras_connect_message *msg,
+					int aud_fd)
+{
+	struct cras_rstream *stream;
+	struct cras_client_stream_connected stream_connected;
+	struct cras_client_message *reply;
+	struct cras_audio_format remote_fmt;
+	struct cras_rstream_config stream_config;
+	int rc, header_fd, samples_fd;
+	int stream_fds[2];
+
+	if (!cras_valid_stream_id(msg->stream_id, client->id)) {
+		syslog(LOG_ERR,
+		       "stream_connect: invalid stream_id: %x for "
+		       "client: %zx.\n",
+		       msg->stream_id, client->id);
+		rc = -EINVAL;
+		goto reply_err;
+	}
+
+	if (msg->direction != CRAS_STREAM_OUTPUT) {
+		syslog(LOG_ERR, "Invalid stream direction.\n");
+		rc = -EINVAL;
+		goto reply_err;
+	}
+
+	unpack_cras_audio_format(&remote_fmt, &msg->format);
+
+	/* check the aud_fd is valid. */
+	if (aud_fd < 0) {
+		syslog(LOG_ERR, "Invalid fd in stream connect.\n");
+		rc = -EBADF;
+		goto reply_err;
+	}
+	/* When full, getting an error is preferable to blocking. */
+	cras_make_fd_nonblocking(aud_fd);
+
+	rclient_fill_cras_rstream_config(client, msg, aud_fd, &remote_fmt,
+					 &stream_config);
+	rc = stream_list_add(cras_iodev_list_get_stream_list(), &stream_config,
+			     &stream);
+	if (rc)
+		goto reply_err;
+
+	/* Tell client about the stream setup. */
+	syslog(LOG_DEBUG, "Send connected for stream %x\n", msg->stream_id);
+	cras_fill_client_stream_connected(
+		&stream_connected, 0, /* No error. */
+		msg->stream_id, &remote_fmt,
+		cras_rstream_get_samples_shm_size(stream),
+		cras_rstream_get_effects(stream));
+	reply = &stream_connected.header;
+
+	rc = cras_rstream_get_shm_fds(stream, &header_fd, &samples_fd);
+	if (rc)
+		goto reply_err;
+
+	stream_fds[0] = header_fd;
+	stream_fds[1] = samples_fd;
+
+	rc = client->ops->send_message_to_client(client, reply, stream_fds, 2);
+	if (rc < 0) {
+		syslog(LOG_ERR, "Failed to send connected messaged\n");
+		stream_list_rm(cras_iodev_list_get_stream_list(),
+			       stream->stream_id);
+		goto reply_err;
+	}
+
+	/* Metrics logs the stream configurations. */
+	cras_server_metrics_stream_config(&stream_config);
+
+	return 0;
+
+reply_err:
+	/* Send the error code to the client. */
+	cras_fill_client_stream_connected(&stream_connected, rc, msg->stream_id,
+					  &remote_fmt, 0, msg->effects);
+	reply = &stream_connected.header;
+	client->ops->send_message_to_client(client, reply, NULL, 0);
+
+	if (aud_fd >= 0)
+		close(aud_fd);
+
+	return rc;
+}
+
+/* Handles messages from the client requesting that a stream be removed from the
+ * server. */
+static int handle_client_stream_disconnect(
+	struct cras_rclient *client,
+	const struct cras_disconnect_stream_message *msg)
+{
+	if (!cras_valid_stream_id(msg->stream_id, client->id)) {
+		syslog(LOG_ERR,
+		       "stream_disconnect: invalid stream_id: %x for "
+		       "client: %zx.\n",
+		       msg->stream_id, client->id);
+		return -EINVAL;
+	}
+	return stream_list_rm(cras_iodev_list_get_stream_list(),
+			      msg->stream_id);
+}
+
+/* Entry point for handling a message from the client.  Called from the main
+ * server context. */
+static int cpr_handle_message_from_client(struct cras_rclient *client,
+					  const struct cras_server_message *msg,
+					  int fd)
+{
+	int rc = 0;
+	assert(client && msg);
+
+	/* Most messages should not have a file descriptor. */
+	switch (msg->id) {
+	case CRAS_SERVER_CONNECT_STREAM:
+		break;
+	default:
+		if (fd != -1) {
+			syslog(LOG_ERR,
+			       "Message %d should not have fd attached.",
+			       msg->id);
+			close(fd);
+			return -EINVAL;
+		}
+		break;
+	}
+
+	switch (msg->id) {
+	case CRAS_SERVER_CONNECT_STREAM:
+		if (!MSG_LEN_VALID(msg, struct cras_connect_message))
+			return -EINVAL;
+		rc = handle_client_stream_connect(
+			client, (const struct cras_connect_message *)msg, fd);
+		break;
+	case CRAS_SERVER_DISCONNECT_STREAM:
+		if (!MSG_LEN_VALID(msg, struct cras_disconnect_stream_message))
+			return -EINVAL;
+		rc = handle_client_stream_disconnect(
+			client,
+			(const struct cras_disconnect_stream_message *)msg);
+		break;
+	default:
+		break;
+	}
+
+	return rc;
+}
+
+/* Declarations of cras_rclient operators for cras_playback_rclient. */
+static const struct cras_rclient_ops cras_playback_rclient_ops = {
+	.handle_message_from_client = cpr_handle_message_from_client,
+	.send_message_to_client = rclient_send_message_to_client,
+	.destroy = rclient_destroy,
+};
+
+/*
+ * Exported Functions.
+ */
+
+/* Creates a client structure and sends a message back informing the client that
+ * the connection has succeeded. */
+struct cras_rclient *cras_playback_rclient_create(int fd, size_t id)
+{
+	struct cras_rclient *client;
+	struct cras_client_connected msg;
+	int state_fd;
+
+	client = (struct cras_rclient *)calloc(1, sizeof(struct cras_rclient));
+	if (!client)
+		return NULL;
+
+	client->fd = fd;
+	client->id = id;
+
+	client->ops = &cras_playback_rclient_ops;
+
+	cras_fill_client_connected(&msg, client->id);
+	state_fd = cras_sys_state_shm_fd();
+	client->ops->send_message_to_client(client, &msg.header, &state_fd, 1);
+
+	return client;
+}
diff --git a/cras/src/server/cras_playback_rclient.h b/cras/src/server/cras_playback_rclient.h
new file mode 100644
index 00000000..37f03cfb
--- /dev/null
+++ b/cras/src/server/cras_playback_rclient.h
@@ -0,0 +1,20 @@
+/* Copyright 2019 The Chromium OS Authors. All rights reserved.
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE file.
+ */
+
+#ifndef CRAS_PLAYBACK_RCLIENT_H_
+#define CRAS_PLAYBACK_RCLIENT_H_
+
+struct cras_rclient;
+
+/* Creates a playback rclient structure.
+ * Args:
+ *    fd - The file descriptor used for communication with the client.
+ *    id - Unique identifier for this client.
+ * Returns:
+ *    A pointer to the newly created rclient on success, NULL on failure.
+ */
+struct cras_rclient *cras_playback_rclient_create(int fd, size_t id);
+
+#endif /* CRAS_PLAYBACK_RCLIENT_H_ */
diff --git a/cras/src/server/cras_rclient.c b/cras/src/server/cras_rclient.c
index 63e20bec..b46693be 100644
--- a/cras/src/server/cras_rclient.c
+++ b/cras/src/server/cras_rclient.c
@@ -42,8 +42,7 @@ int cras_rclient_buffer_from_client(struct cras_rclient *client,
 		return -EINVAL;
 	if (msg->length != buf_len)
 		return -EINVAL;
-	client->ops->handle_message_from_client(client, msg, fd);
-	return 0;
+	return client->ops->handle_message_from_client(client, msg, fd);
 }
 
 /* Sends a message to the client. */
diff --git a/cras/src/server/cras_rclient_util.h b/cras/src/server/cras_rclient_util.h
index 189b3f37..2ce08827 100644
--- a/cras/src/server/cras_rclient_util.h
+++ b/cras/src/server/cras_rclient_util.h
@@ -9,6 +9,8 @@
 #ifndef CRAS_RCLIENT_UTIL_H_
 #define CRAS_RCLIENT_UTIL_H_
 
+#define MSG_LEN_VALID(msg, type) ((msg)->length >= sizeof(type))
+
 struct cras_connect_message;
 struct cras_rclient;
 struct cras_rclient_message;
diff --git a/cras/src/server/cras_server.c b/cras/src/server/cras_server.c
index 57c9a912..27dc3732 100644
--- a/cras/src/server/cras_server.c
+++ b/cras/src/server/cras_server.c
@@ -46,6 +46,7 @@
 #include "cras_observer.h"
 #include "cras_rclient.h"
 #include "cras_control_rclient.h"
+#include "cras_playback_rclient.h"
 #include "cras_server.h"
 #include "cras_server_metrics.h"
 #include "cras_system_state.h"
@@ -192,9 +193,17 @@ static void send_client_list_to_clients(struct server_data *serv)
 	cras_system_state_update_complete();
 }
 
+/* CRAS client connection types. */
+enum CRAS_CONNECTION_TYPE {
+	CRAS_CONTROL, // For legacy client.
+	CRAS_PLAYBACK, // For playback client.
+	CRAS_NUM_CONN_TYPE,
+};
+
 /* Handles requests from a client to attach to the server.  Create a local
  * structure to track the client, assign it a unique id and let it attach */
-static void handle_new_connection(struct sockaddr_un *address, int fd)
+static void handle_new_connection(struct sockaddr_un *address, int fd,
+				  enum CRAS_CONNECTION_TYPE type)
 {
 	int connection_fd;
 	struct attached_client *poll_client;
@@ -232,13 +241,22 @@ static void handle_new_connection(struct sockaddr_un *address, int fd)
 	poll_client->next = NULL;
 	poll_client->pollfd = NULL;
 	fill_client_info(poll_client);
-	poll_client->client =
-		cras_control_rclient_create(connection_fd, poll_client->id);
+	switch (type) {
+	case CRAS_CONTROL:
+		poll_client->client = cras_control_rclient_create(
+			connection_fd, poll_client->id);
+		break;
+	case CRAS_PLAYBACK:
+		poll_client->client = cras_playback_rclient_create(
+			connection_fd, poll_client->id);
+		break;
+	default:
+		syslog(LOG_ERR, "unsupported connection type");
+		goto error;
+	}
 	if (poll_client->client == NULL) {
 		syslog(LOG_ERR, "failed to create client");
-		close(connection_fd);
-		free(poll_client);
-		return;
+		goto error;
 	}
 
 	DL_APPEND(server_instance.clients_head, poll_client);
@@ -246,6 +264,11 @@ static void handle_new_connection(struct sockaddr_un *address, int fd)
 	/* Send a current list of available inputs and outputs. */
 	cras_iodev_list_update_device_list();
 	send_client_list_to_clients(&server_instance);
+	return;
+error:
+	close(connection_fd);
+	free(poll_client);
+	return;
 }
 
 /* Add a file descriptor to be passed to select in the main loop. This is
@@ -512,8 +535,10 @@ int cras_server_run(unsigned int profile_disable_mask)
 	DBusConnection *dbus_conn;
 #endif
 	int control_fd = -1;
+	int playback_fd = -1;
 	int rc = 0;
 	struct sockaddr_un control_addr;
+	struct sockaddr_un playback_addr;
 	struct attached_client *elm;
 	struct client_callback *client_cb;
 	struct system_task *tasks;
@@ -564,6 +589,11 @@ int cras_server_run(unsigned int profile_disable_mask)
 	if (control_fd < 0)
 		goto bail;
 
+	playback_fd = create_and_listen_server_socket(CRAS_PLAYBACK_SOCKET_FILE,
+						      &playback_addr);
+	if (playback_fd < 0)
+		goto bail;
+
 	tm = cras_system_state_get_tm();
 	if (!tm) {
 		syslog(LOG_ERR, "Getting timer manager.");
@@ -576,7 +606,8 @@ int cras_server_run(unsigned int profile_disable_mask)
 
 	/* Main server loop - client callbacks are run from this context. */
 	while (1) {
-		poll_size_needed = 1 + server_instance.num_clients +
+		poll_size_needed = CRAS_NUM_CONN_TYPE +
+				   server_instance.num_clients +
 				   server_instance.num_client_callbacks;
 		if (poll_size_needed > pollfds_size) {
 			pollfds_size = 2 * poll_size_needed;
@@ -584,9 +615,12 @@ int cras_server_run(unsigned int profile_disable_mask)
 					  sizeof(*pollfds) * pollfds_size);
 		}
 
-		pollfds[0].fd = control_fd;
-		pollfds[0].events = POLLIN;
-		num_pollfds = 1;
+		pollfds[CRAS_CONTROL].fd = control_fd;
+		pollfds[CRAS_CONTROL].events = POLLIN;
+
+		pollfds[CRAS_PLAYBACK].fd = playback_fd;
+		pollfds[CRAS_PLAYBACK].events = POLLIN;
+		num_pollfds = CRAS_NUM_CONN_TYPE;
 
 		DL_FOREACH (server_instance.clients_head, elm) {
 			pollfds[num_pollfds].fd = elm->fd;
@@ -629,8 +663,12 @@ int cras_server_run(unsigned int profile_disable_mask)
 		cras_tm_call_callbacks(tm);
 
 		/* Check for new connections. */
-		if (pollfds[0].revents & POLLIN)
-			handle_new_connection(&control_addr, control_fd);
+		if (pollfds[CRAS_CONTROL].revents & POLLIN)
+			handle_new_connection(&control_addr, control_fd,
+					      CRAS_CONTROL);
+		if (pollfds[CRAS_PLAYBACK].revents & POLLIN)
+			handle_new_connection(&playback_addr, playback_fd,
+					      CRAS_PLAYBACK);
 		/* Check if there are messages pending for any clients. */
 		DL_FOREACH (server_instance.clients_head, elm)
 			if (elm->pollfd && elm->pollfd->revents & POLLIN)
@@ -656,6 +694,10 @@ bail:
 		close(control_fd);
 		unlink(control_addr.sun_path);
 	}
+	if (playback_fd >= 0) {
+		close(playback_fd);
+		unlink(playback_addr.sun_path);
+	}
 	free(pollfds);
 	cras_observer_server_free();
 	return rc;
diff --git a/cras/src/tests/playback_rclient_unittest.cc b/cras/src/tests/playback_rclient_unittest.cc
new file mode 100644
index 00000000..936dbd77
--- /dev/null
+++ b/cras/src/tests/playback_rclient_unittest.cc
@@ -0,0 +1,270 @@
+// Copyright 2019 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <gtest/gtest.h>
+#include <stdio.h>
+#include <unistd.h>
+
+extern "C" {
+#include "audio_thread.h"
+#include "cras_bt_log.h"
+#include "cras_messages.h"
+#include "cras_rclient.h"
+#include "cras_rstream.h"
+#include "cras_system_state.h"
+
+// Access to data structures and static functions.
+#include "cras_playback_rclient.c"
+#include "cras_rclient_util.c"
+}
+static unsigned int cras_make_fd_nonblocking_called;
+static unsigned int cras_observer_remove_called;
+static unsigned int cras_server_metrics_stream_config_called;
+static int stream_list_add_called;
+static int stream_list_add_return;
+static unsigned int stream_list_rm_called;
+static struct cras_audio_shm dummy_shm;
+static struct cras_rstream dummy_rstream;
+
+void ResetStubData() {
+  cras_make_fd_nonblocking_called = 0;
+  cras_observer_remove_called = 0;
+  cras_server_metrics_stream_config_called = 0;
+  stream_list_add_called = 0;
+  stream_list_add_return = 0;
+  stream_list_rm_called = 0;
+}
+
+namespace {
+
+TEST(RClientSuite, CreateSendMessage) {
+  struct cras_rclient* rclient;
+  int rc;
+  struct cras_client_connected msg;
+  int pipe_fds[2];
+
+  ResetStubData();
+
+  rc = pipe(pipe_fds);
+  ASSERT_EQ(0, rc);
+
+  rclient = cras_playback_rclient_create(pipe_fds[1], 800);
+  ASSERT_NE((void*)NULL, rclient);
+  EXPECT_EQ(800, rclient->id);
+
+  rc = read(pipe_fds[0], &msg, sizeof(msg));
+  EXPECT_EQ(sizeof(msg), rc);
+  EXPECT_EQ(CRAS_CLIENT_CONNECTED, msg.header.id);
+
+  rclient->ops->destroy(rclient);
+  EXPECT_EQ(1, cras_observer_remove_called);
+  close(pipe_fds[0]);
+  close(pipe_fds[1]);
+}
+
+class CPRMessageSuite : public testing::Test {
+ protected:
+  virtual void SetUp() {
+    int rc;
+    struct cras_client_connected msg;
+
+    rc = pipe(pipe_fds_);
+    if (rc < 0)
+      return;
+
+    rclient_ = cras_playback_rclient_create(pipe_fds_[1], 1);
+    rc = read(pipe_fds_[0], &msg, sizeof(msg));
+    if (rc < 0)
+      return;
+
+    ResetStubData();
+  }
+
+  virtual void TearDown() {
+    rclient_->ops->destroy(rclient_);
+    close(pipe_fds_[0]);
+    close(pipe_fds_[1]);
+  }
+
+  struct cras_rclient* rclient_;
+  int pipe_fds_[2];
+};
+
+TEST_F(CPRMessageSuite, StreamConnectMessage) {
+  struct cras_client_stream_connected out_msg;
+  int rc;
+
+  struct cras_connect_message msg;
+  cras_stream_id_t stream_id = 0x10002;
+  struct cras_audio_format fmt = {
+      .frame_rate = 48000,
+      .num_channels = 2,
+      .format = SND_PCM_FORMAT_S16_LE,
+  };
+  cras_fill_connect_message(&msg, CRAS_STREAM_OUTPUT, stream_id,
+                            CRAS_STREAM_TYPE_DEFAULT, 480, 240,
+                            /*flags=*/0, /*effects=*/0, fmt, NO_DEVICE);
+  ASSERT_EQ(stream_id, msg.stream_id);
+
+  rclient_->ops->handle_message_from_client(rclient_, &msg.header, 100);
+  EXPECT_EQ(1, cras_make_fd_nonblocking_called);
+  EXPECT_EQ(1, stream_list_add_called);
+  EXPECT_EQ(0, stream_list_rm_called);
+
+  rc = read(pipe_fds_[0], &out_msg, sizeof(out_msg));
+  EXPECT_EQ(sizeof(out_msg), rc);
+  EXPECT_EQ(stream_id, out_msg.stream_id);
+}
+
+TEST_F(CPRMessageSuite, StreamConnectMessageInvalidDirection) {
+  struct cras_client_stream_connected out_msg;
+  int rc;
+
+  struct cras_connect_message msg;
+  cras_stream_id_t stream_id = 0x10002;
+  struct cras_audio_format fmt = {
+      .frame_rate = 48000,
+      .num_channels = 2,
+      .format = SND_PCM_FORMAT_S16_LE,
+  };
+  cras_fill_connect_message(&msg, CRAS_STREAM_INPUT, stream_id,
+                            CRAS_STREAM_TYPE_DEFAULT, 480, 240,
+                            /*flags=*/0, /*effects=*/0, fmt, NO_DEVICE);
+  ASSERT_EQ(stream_id, msg.stream_id);
+
+  rc = rclient_->ops->handle_message_from_client(rclient_, &msg.header, 100);
+  EXPECT_EQ(-EINVAL, rc);
+  EXPECT_EQ(0, cras_make_fd_nonblocking_called);
+  EXPECT_EQ(0, stream_list_add_called);
+  EXPECT_EQ(0, stream_list_rm_called);
+
+  rc = read(pipe_fds_[0], &out_msg, sizeof(out_msg));
+  EXPECT_EQ(sizeof(out_msg), rc);
+  EXPECT_EQ(-EINVAL, out_msg.err);
+  EXPECT_EQ(stream_id, out_msg.stream_id);
+}
+
+TEST_F(CPRMessageSuite, StreamConnectMessageInvalidClientId) {
+  struct cras_client_stream_connected out_msg;
+  int rc;
+
+  struct cras_connect_message msg;
+  cras_stream_id_t stream_id = 0x20002;  // stream_id with invalid client_id
+  struct cras_audio_format fmt = {
+      .frame_rate = 48000,
+      .num_channels = 2,
+      .format = SND_PCM_FORMAT_S16_LE,
+  };
+  cras_fill_connect_message(&msg, CRAS_STREAM_OUTPUT, stream_id,
+                            CRAS_STREAM_TYPE_DEFAULT, 480, 240,
+                            /*flags=*/0, /*effects=*/0, fmt, NO_DEVICE);
+  ASSERT_EQ(stream_id, msg.stream_id);
+
+  rc = rclient_->ops->handle_message_from_client(rclient_, &msg.header, 100);
+  EXPECT_EQ(-EINVAL, rc);
+  EXPECT_EQ(0, cras_make_fd_nonblocking_called);
+  EXPECT_EQ(0, stream_list_add_called);
+  EXPECT_EQ(0, stream_list_rm_called);
+
+  rc = read(pipe_fds_[0], &out_msg, sizeof(out_msg));
+  EXPECT_EQ(sizeof(out_msg), rc);
+  EXPECT_EQ(-EINVAL, out_msg.err);
+  EXPECT_EQ(stream_id, out_msg.stream_id);
+}
+
+TEST_F(CPRMessageSuite, StreamDisconnectMessage) {
+  struct cras_disconnect_stream_message msg;
+  cras_stream_id_t stream_id = 0x10002;
+  cras_fill_disconnect_stream_message(&msg, stream_id);
+
+  rclient_->ops->handle_message_from_client(rclient_, &msg.header, -1);
+  EXPECT_EQ(0, stream_list_add_called);
+  EXPECT_EQ(1, stream_list_rm_called);
+}
+
+TEST_F(CPRMessageSuite, StreamDisconnectMessageInvalidClientId) {
+  struct cras_disconnect_stream_message msg;
+  cras_stream_id_t stream_id = 0x20002;  // stream_id with invalid client_id
+  cras_fill_disconnect_stream_message(&msg, stream_id);
+
+  rclient_->ops->handle_message_from_client(rclient_, &msg.header, -1);
+  EXPECT_EQ(0, stream_list_add_called);
+  EXPECT_EQ(0, stream_list_rm_called);
+}
+}  // namespace
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+/* stubs */
+extern "C" {
+
+struct stream_list* cras_iodev_list_get_stream_list() {
+  return NULL;
+}
+
+int cras_make_fd_nonblocking(int fd) {
+  cras_make_fd_nonblocking_called++;
+  return 0;
+}
+
+void cras_observer_remove(struct cras_observer_client* client) {
+  cras_observer_remove_called++;
+}
+
+unsigned int cras_rstream_get_effects(const struct cras_rstream* stream) {
+  return 0;
+}
+
+int cras_server_metrics_stream_config(struct cras_rstream_config* config) {
+  cras_server_metrics_stream_config_called++;
+  return 0;
+}
+
+int cras_send_with_fds(int sockfd,
+                       const void* buf,
+                       size_t len,
+                       int* fd,
+                       unsigned int num_fds) {
+  return write(sockfd, buf, len);
+}
+
+key_t cras_sys_state_shm_fd() {
+  return 1;
+}
+
+void cras_system_set_suspended(int suspended) {}
+
+int stream_list_rm_all_client_streams(struct stream_list* list,
+                                      struct cras_rclient* rclient) {
+  return 0;
+}
+
+int stream_list_rm(struct stream_list* list, cras_stream_id_t id) {
+  stream_list_rm_called++;
+  return 0;
+}
+
+int stream_list_add(struct stream_list* list,
+                    struct cras_rstream_config* config,
+                    struct cras_rstream** stream) {
+  int ret;
+
+  *stream = &dummy_rstream;
+
+  stream_list_add_called++;
+  ret = stream_list_add_return;
+  if (ret)
+    stream_list_add_return = -EINVAL;
+
+  dummy_rstream.shm = &dummy_shm;
+  dummy_rstream.direction = config->direction;
+  dummy_rstream.stream_id = config->stream_id;
+
+  return ret;
+}
+
+}  // extern "C"