Add CastPlatformClient for handling sender requests

This change adds a new class for handling request/response sequences
initiated by a Cast sender.  It will initially be used for app
availability requests and is modelled after Chrome's CastMessageHandler.

Bug: openscreen:60
Change-Id: Iba5ae0ee343e23a2c0214b5b17a2b59dca2cbe99
Reviewed-on: https://chromium-review.googlesource.com/c/openscreen/+/2044149
Commit-Queue: Brandon Tolsch <btolsch@chromium.org>
Reviewed-by: mark a. foltz <mfoltz@chromium.org>
Reviewed-by: Takumi Fujimoto <takumif@chromium.org>

10 files changed, 629 insertions, 0 deletions

diff --git a/cast/sender/BUILD.gn b/cast/sender/BUILD.gn
index dfbc947e..86ecb7d6 100644
--- a/cast/sender/BUILD.gn
+++ b/cast/sender/BUILD.gn
@@ -31,6 +31,8 @@ source_set("sender") {
   sources = [
     "cast_app_availability_tracker.cc",
     "cast_app_availability_tracker.h",
+    "cast_platform_client.cc",
+    "cast_platform_client.h",
     "public/cast_media_source.cc",
     "public/cast_media_source.h",
   ]
@@ -38,7 +40,29 @@ source_set("sender") {
   public_deps = [
     ":channel",
     "../../platform",
+    "../../third_party/abseil",
     "../../util",
+    "../common:channel",
+    "../common:public",
+  ]
+}
+
+source_set("test_helpers") {
+  testonly = true
+  sources = [
+    "testing/test_helpers.cc",
+    "testing/test_helpers.h",
+  ]
+
+  deps = [
+    "../../third_party/googletest:gtest",
+    "../../util",
+    "../common:channel",
+    "../receiver:channel",
+  ]
+
+  public_deps = [
+    ":channel",
   ]
 }
 
@@ -46,17 +70,21 @@ source_set("unittests") {
   testonly = true
   sources = [
     "cast_app_availability_tracker_unittest.cc",
+    "cast_platform_client_unittest.cc",
     "channel/cast_auth_util_unittest.cc",
   ]
 
   deps = [
     ":channel",
     ":sender",
+    ":test_helpers",
     "../../platform",
     "../../platform:test",
     "../../testing/util",
     "../../third_party/googletest:gmock",
     "../../third_party/googletest:gtest",
+    "../../util",
+    "../common:test_helpers",
     "../common/certificate/proto:certificate_proto",
     "../common/certificate/proto:certificate_unittest_proto",
   ]
diff --git a/cast/sender/cast_app_availability_tracker.h b/cast/sender/cast_app_availability_tracker.h
index bced8a50..c0bded96 100644
--- a/cast/sender/cast_app_availability_tracker.h
+++ b/cast/sender/cast_app_availability_tracker.h
@@ -45,6 +45,7 @@ namespace cast {
 //
 // (5b): At any time, the caller may call |GetAvailableDevices()| (even before
 // the source is registered) to determine if there are cached results available.
+// TODO(crbug.com/openscreen/112): Device -> Receiver renaming.
 class CastAppAvailabilityTracker {
  public:
   // The result of an app availability request and the time when it is obtained.
diff --git a/cast/sender/cast_platform_client.cc b/cast/sender/cast_platform_client.cc
new file mode 100644
index 00000000..6e3d13da
--- /dev/null
+++ b/cast/sender/cast_platform_client.cc
@@ -0,0 +1,224 @@
+// Copyright 2020 The Chromium 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 "cast/sender/cast_platform_client.h"
+
+#include <random>
+
+#include "absl/strings/str_cat.h"
+#include "cast/common/channel/cast_socket.h"
+#include "cast/common/channel/virtual_connection_manager.h"
+#include "cast/common/channel/virtual_connection_router.h"
+#include "cast/common/public/service_info.h"
+#include "util/json/json_serialization.h"
+#include "util/logging.h"
+#include "util/stringprintf.h"
+
+namespace openscreen {
+namespace cast {
+
+static constexpr std::chrono::seconds kRequestTimeout = std::chrono::seconds(5);
+
+namespace {
+
+std::string MakeRandomSenderId() {
+  static auto& rd = *new std::random_device();
+  static auto& gen = *new std::mt19937(rd());
+  static auto& dist = *new std::uniform_int_distribution<>(1, 1000000);
+  return absl::StrCat("sender-", dist(gen));
+}
+
+}  // namespace
+
+CastPlatformClient::CastPlatformClient(VirtualConnectionRouter* router,
+                                       VirtualConnectionManager* manager,
+                                       ClockNowFunctionPtr clock,
+                                       TaskRunner* task_runner)
+    : sender_id_(MakeRandomSenderId()),
+      virtual_conn_router_(router),
+      virtual_conn_manager_(manager),
+      clock_(clock),
+      task_runner_(task_runner) {
+  OSP_DCHECK(virtual_conn_manager_);
+  OSP_DCHECK(clock_);
+  OSP_DCHECK(task_runner_);
+  virtual_conn_router_->AddHandlerForLocalId(sender_id_, this);
+}
+
+CastPlatformClient::~CastPlatformClient() {
+  virtual_conn_router_->RemoveHandlerForLocalId(sender_id_);
+
+  for (auto& pending_requests : pending_requests_by_device_id_) {
+    for (auto& avail_request : pending_requests.second.availability) {
+      avail_request.callback(avail_request.app_id,
+                             AppAvailabilityResult::kUnknown);
+    }
+  }
+}
+
+absl::optional<int> CastPlatformClient::RequestAppAvailability(
+    const std::string& device_id,
+    const std::string& app_id,
+    AppAvailabilityCallback callback) {
+  auto entry = socket_id_by_device_id_.find(device_id);
+  if (entry == socket_id_by_device_id_.end()) {
+    callback(app_id, AppAvailabilityResult::kUnknown);
+    return absl::nullopt;
+  }
+  int socket_id = entry->second;
+
+  int request_id = GetNextRequestId();
+  ErrorOr<::cast::channel::CastMessage> message =
+      CreateAppAvailabilityRequest(sender_id_, request_id, app_id);
+  OSP_DCHECK(message);
+
+  PendingRequests& pending_requests = pending_requests_by_device_id_[device_id];
+  auto timeout = std::make_unique<Alarm>(clock_, task_runner_);
+  timeout->ScheduleFromNow(
+      [this, request_id]() { CancelAppAvailabilityRequest(request_id); },
+      kRequestTimeout);
+  pending_requests.availability.push_back(AvailabilityRequest{
+      request_id, app_id, std::move(timeout), std::move(callback)});
+
+  VirtualConnection virtual_conn{sender_id_, kPlatformReceiverId, socket_id};
+  if (!virtual_conn_manager_->GetConnectionData(virtual_conn)) {
+    virtual_conn_manager_->AddConnection(virtual_conn,
+                                         VirtualConnection::AssociatedData{});
+  }
+
+  virtual_conn_router_->SendMessage(std::move(virtual_conn),
+                                    std::move(message.value()));
+
+  return request_id;
+}
+
+void CastPlatformClient::AddOrUpdateReceiver(const ServiceInfo& device,
+                                             int socket_id) {
+  socket_id_by_device_id_[device.unique_id] = socket_id;
+}
+
+void CastPlatformClient::RemoveReceiver(const ServiceInfo& device) {
+  auto pending_requests_it =
+      pending_requests_by_device_id_.find(device.unique_id);
+  if (pending_requests_it != pending_requests_by_device_id_.end()) {
+    for (const AvailabilityRequest& availability :
+         pending_requests_it->second.availability) {
+      availability.callback(availability.app_id,
+                            AppAvailabilityResult::kUnknown);
+    }
+    pending_requests_by_device_id_.erase(pending_requests_it);
+  }
+  socket_id_by_device_id_.erase(device.unique_id);
+}
+
+void CastPlatformClient::CancelRequest(int request_id) {
+  for (auto entry = pending_requests_by_device_id_.begin();
+       entry != pending_requests_by_device_id_.end(); ++entry) {
+    auto& pending_requests = entry->second;
+    auto it = std::find_if(pending_requests.availability.begin(),
+                           pending_requests.availability.end(),
+                           [request_id](const AvailabilityRequest& request) {
+                             return request.request_id == request_id;
+                           });
+    if (it != pending_requests.availability.end()) {
+      pending_requests.availability.erase(it);
+      break;
+    }
+  }
+}
+
+void CastPlatformClient::OnMessage(VirtualConnectionRouter* router,
+                                   CastSocket* socket,
+                                   ::cast::channel::CastMessage message) {
+  if (message.payload_type() !=
+          ::cast::channel::CastMessage_PayloadType_STRING ||
+      message.namespace_() != kReceiverNamespace ||
+      message.source_id() != kPlatformReceiverId) {
+    return;
+  }
+  ErrorOr<Json::Value> dict_or_error = json::Parse(message.payload_utf8());
+  if (dict_or_error.is_error()) {
+    OSP_DVLOG << "Failed to deserialize CastMessage payload.";
+    return;
+  }
+
+  Json::Value& dict = dict_or_error.value();
+  absl::optional<int> request_id =
+      MaybeGetInt(dict, JSON_EXPAND_FIND_CONSTANT_ARGS(kMessageKeyRequestId));
+  if (request_id) {
+    auto entry = std::find_if(
+        socket_id_by_device_id_.begin(), socket_id_by_device_id_.end(),
+        [socket](const std::pair<std::string, int>& entry) {
+          return entry.second == socket->socket_id();
+        });
+    if (entry != socket_id_by_device_id_.end()) {
+      HandleResponse(entry->first, request_id.value(), dict);
+    }
+  }
+}
+
+void CastPlatformClient::HandleResponse(const std::string& device_id,
+                                        int request_id,
+                                        const Json::Value& message) {
+  auto entry = pending_requests_by_device_id_.find(device_id);
+  if (entry == pending_requests_by_device_id_.end()) {
+    return;
+  }
+  PendingRequests& pending_requests = entry->second;
+  auto it = std::find_if(pending_requests.availability.begin(),
+                         pending_requests.availability.end(),
+                         [request_id](const AvailabilityRequest& request) {
+                           return request.request_id == request_id;
+                         });
+  if (it != pending_requests.availability.end()) {
+    // TODO(btolsch): Can all of this manual parsing/checking be cleaned up into
+    // a single parsing API along with other message handling?
+    const Json::Value* maybe_availability =
+        message.find(JSON_EXPAND_FIND_CONSTANT_ARGS(kMessageKeyAvailability));
+    if (maybe_availability && maybe_availability->isObject()) {
+      absl::optional<absl::string_view> result =
+          MaybeGetString(*maybe_availability, &it->app_id[0],
+                         &it->app_id[0] + it->app_id.size());
+      if (result) {
+        AppAvailabilityResult availability_result =
+            AppAvailabilityResult::kUnknown;
+        if (result.value() == kMessageValueAppAvailable) {
+          availability_result = AppAvailabilityResult::kAvailable;
+        } else if (result.value() == kMessageValueAppUnavailable) {
+          availability_result = AppAvailabilityResult::kUnavailable;
+        } else {
+          OSP_DVLOG << "Invalid availability result: " << result.value();
+        }
+        it->callback(it->app_id, availability_result);
+      }
+    }
+    pending_requests.availability.erase(it);
+  }
+}
+
+void CastPlatformClient::CancelAppAvailabilityRequest(int request_id) {
+  for (auto& entry : pending_requests_by_device_id_) {
+    PendingRequests& pending_requests = entry.second;
+    auto it = std::find_if(pending_requests.availability.begin(),
+                           pending_requests.availability.end(),
+                           [request_id](const AvailabilityRequest& request) {
+                             return request.request_id == request_id;
+                           });
+    if (it != pending_requests.availability.end()) {
+      it->callback(it->app_id, AppAvailabilityResult::kUnknown);
+      pending_requests.availability.erase(it);
+    }
+  }
+}
+
+// static
+int CastPlatformClient::GetNextRequestId() {
+  return next_request_id_++;
+}
+
+// static
+int CastPlatformClient::next_request_id_ = 0;
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/cast_platform_client.h b/cast/sender/cast_platform_client.h
new file mode 100644
index 00000000..41ad7fc7
--- /dev/null
+++ b/cast/sender/cast_platform_client.h
@@ -0,0 +1,97 @@
+// Copyright 2020 The Chromium 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 CAST_SENDER_CAST_PLATFORM_CLIENT_H_
+#define CAST_SENDER_CAST_PLATFORM_CLIENT_H_
+
+#include <functional>
+#include <map>
+#include <string>
+
+#include "absl/types/optional.h"
+#include "cast/common/channel/cast_message_handler.h"
+#include "cast/sender/channel/message_util.h"
+#include "util/alarm.h"
+#include "util/json/json_value.h"
+
+namespace openscreen {
+namespace cast {
+
+struct ServiceInfo;
+class VirtualConnectionManager;
+class VirtualConnectionRouter;
+
+// This class handles Cast messages that generally relate to the "platform", in
+// other words not a specific app currently running (e.g. app availability,
+// receiver status).  These messages follow a request/response format, so each
+// request requires a corresponding response callback.  These requests will also
+// timeout if there is no response after a certain amount of time (currently 5
+// seconds).  The timeout callbacks will be called on the thread managed by
+// |task_runner|.
+class CastPlatformClient final : public CastMessageHandler {
+ public:
+  using AppAvailabilityCallback =
+      std::function<void(const std::string& app_id, AppAvailabilityResult)>;
+
+  CastPlatformClient(VirtualConnectionRouter* router,
+                     VirtualConnectionManager* manager,
+                     ClockNowFunctionPtr clock,
+                     TaskRunner* task_runner);
+  ~CastPlatformClient() override;
+
+  // Requests availability information for |app_id| from the receiver identified
+  // by |device_id|.  |callback| will be called exactly once with a result.
+  absl::optional<int> RequestAppAvailability(const std::string& device_id,
+                                             const std::string& app_id,
+                                             AppAvailabilityCallback callback);
+
+  // Notifies this object about general receiver connectivity or property
+  // changes.
+  void AddOrUpdateReceiver(const ServiceInfo& device, int socket_id);
+  void RemoveReceiver(const ServiceInfo& device);
+
+  void CancelRequest(int request_id);
+
+ private:
+  struct AvailabilityRequest {
+    int request_id;
+    std::string app_id;
+    std::unique_ptr<Alarm> timeout;
+    AppAvailabilityCallback callback;
+  };
+
+  struct PendingRequests {
+    std::vector<AvailabilityRequest> availability;
+  };
+
+  // CastMessageHandler overrides.
+  void OnMessage(VirtualConnectionRouter* router,
+                 CastSocket* socket,
+                 ::cast::channel::CastMessage message) override;
+
+  void HandleResponse(const std::string& device_id,
+                      int request_id,
+                      const Json::Value& message);
+
+  void CancelAppAvailabilityRequest(int request_id);
+
+  static int GetNextRequestId();
+
+  static int next_request_id_;
+
+  const std::string sender_id_;
+  VirtualConnectionRouter* const virtual_conn_router_;
+  VirtualConnectionManager* const virtual_conn_manager_;
+  std::map<std::string /* device_id */, int> socket_id_by_device_id_;
+  std::map<std::string /* device_id */, PendingRequests>
+      pending_requests_by_device_id_;
+
+  const ClockNowFunctionPtr clock_;
+  TaskRunner* const task_runner_;
+};
+
+}  // namespace cast
+}  // namespace openscreen
+
+#endif  // CAST_SENDER_CAST_PLATFORM_CLIENT_H_
diff --git a/cast/sender/cast_platform_client_unittest.cc b/cast/sender/cast_platform_client_unittest.cc
new file mode 100644
index 00000000..e4f3769b
--- /dev/null
+++ b/cast/sender/cast_platform_client_unittest.cc
@@ -0,0 +1,109 @@
+// Copyright 2020 The Chromium 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 "cast/sender/cast_platform_client.h"
+
+#include "cast/common/channel/testing/fake_cast_socket.h"
+#include "cast/common/channel/testing/mock_socket_error_handler.h"
+#include "cast/common/channel/virtual_connection_manager.h"
+#include "cast/common/channel/virtual_connection_router.h"
+#include "cast/common/public/service_info.h"
+#include "cast/sender/testing/test_helpers.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "platform/test/fake_clock.h"
+#include "platform/test/fake_task_runner.h"
+#include "util/json/json_serialization.h"
+#include "util/json/json_value.h"
+
+namespace openscreen {
+namespace cast {
+
+using ::cast::channel::CastMessage;
+
+using ::testing::_;
+
+class CastPlatformClientTest : public ::testing::Test {
+ public:
+  void SetUp() override {
+    socket_ = fake_cast_socket_pair_.socket.get();
+    router_.TakeSocket(&mock_error_handler_,
+                       std::move(fake_cast_socket_pair_.socket));
+
+    receiver_.v4_endpoint = IPEndpoint{{192, 168, 0, 17}, 4434};
+    receiver_.unique_id = "deviceId1";
+    platform_client_.AddOrUpdateReceiver(receiver_, socket_->socket_id());
+  }
+
+ protected:
+  CastSocket& peer_socket() { return *fake_cast_socket_pair_.peer_socket; }
+  MockCastSocketClient& peer_client() {
+    return fake_cast_socket_pair_.mock_peer_client;
+  }
+
+  FakeCastSocketPair fake_cast_socket_pair_;
+  CastSocket* socket_ = nullptr;
+  MockSocketErrorHandler mock_error_handler_;
+  VirtualConnectionManager manager_;
+  VirtualConnectionRouter router_{&manager_};
+  FakeClock clock_{Clock::now()};
+  FakeTaskRunner task_runner_{&clock_};
+  CastPlatformClient platform_client_{&router_, &manager_, &FakeClock::now,
+                                      &task_runner_};
+  ServiceInfo receiver_;
+};
+
+TEST_F(CastPlatformClientTest, AppAvailability) {
+  int request_id = -1;
+  std::string sender_id;
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_id, &sender_id](CastSocket* socket,
+                                          CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "AAA", &request_id, &sender_id);
+      });
+  bool ran = false;
+  platform_client_.RequestAppAvailability(
+      "deviceId1", "AAA",
+      [&ran](const std::string& app_id, AppAvailabilityResult availability) {
+        EXPECT_EQ("AAA", app_id);
+        EXPECT_EQ(availability, AppAvailabilityResult::kAvailable);
+        ran = true;
+      });
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  EXPECT_TRUE(ran);
+
+  // NOTE: Callback should only fire once, so it should not fire again here.
+  ran = false;
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  EXPECT_FALSE(ran);
+}
+
+TEST_F(CastPlatformClientTest, CancelRequest) {
+  int request_id = -1;
+  std::string sender_id;
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_id, &sender_id](CastSocket* socket,
+                                          CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "AAA", &request_id, &sender_id);
+      });
+  absl::optional<int> maybe_request_id =
+      platform_client_.RequestAppAvailability(
+          "deviceId1", "AAA",
+          [](const std::string& app_id, AppAvailabilityResult availability) {
+            EXPECT_TRUE(false);
+          });
+  ASSERT_TRUE(maybe_request_id);
+  int local_request_id = maybe_request_id.value();
+  platform_client_.CancelRequest(local_request_id);
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+}
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/channel/message_util.cc b/cast/sender/channel/message_util.cc
index 48623cee..6d96b730 100644
--- a/cast/sender/channel/message_util.cc
+++ b/cast/sender/channel/message_util.cc
@@ -5,6 +5,7 @@
 #include "cast/sender/channel/message_util.h"
 
 #include "cast/sender/channel/cast_auth_util.h"
+#include "util/json/json_serialization.h"
 
 namespace openscreen {
 namespace cast {
@@ -34,5 +35,33 @@ CastMessage CreateAuthChallengeMessage(const AuthContext& auth_context) {
   return message;
 }
 
+ErrorOr<CastMessage> CreateAppAvailabilityRequest(const std::string& sender_id,
+                                                  int request_id,
+                                                  const std::string& app_id) {
+  Json::Value dict(Json::ValueType::objectValue);
+  dict[kMessageKeyType] = Json::Value(
+      CastMessageTypeToString(CastMessageType::kGetAppAvailability));
+  Json::Value app_id_value(Json::ValueType::arrayValue);
+  app_id_value.append(Json::Value(app_id));
+  dict[kMessageKeyAppId] = std::move(app_id_value);
+  dict[kMessageKeyRequestId] = Json::Value(request_id);
+
+  CastMessage message;
+  message.set_payload_type(::cast::channel::CastMessage_PayloadType_STRING);
+  ErrorOr<std::string> serialized = json::Stringify(dict);
+  if (serialized.is_error()) {
+    return serialized.error();
+  }
+  message.set_payload_utf8(serialized.value());
+
+  message.set_protocol_version(
+      ::cast::channel::CastMessage_ProtocolVersion_CASTV2_1_0);
+  message.set_source_id(sender_id);
+  message.set_destination_id(kPlatformReceiverId);
+  message.set_namespace_(kReceiverNamespace);
+
+  return message;
+}
+
 }  // namespace cast
 }  // namespace openscreen
diff --git a/cast/sender/channel/message_util.h b/cast/sender/channel/message_util.h
index 944165bc..1a8c7717 100644
--- a/cast/sender/channel/message_util.h
+++ b/cast/sender/channel/message_util.h
@@ -7,6 +7,7 @@
 
 #include "cast/common/channel/message_util.h"
 #include "cast/common/channel/proto/cast_channel.pb.h"
+#include "platform/base/error.h"
 
 namespace openscreen {
 namespace cast {
@@ -16,6 +17,12 @@ class AuthContext;
 ::cast::channel::CastMessage CreateAuthChallengeMessage(
     const AuthContext& auth_context);
 
+// |request_id| must be unique for |sender_id|.
+ErrorOr<::cast::channel::CastMessage> CreateAppAvailabilityRequest(
+    const std::string& sender_id,
+    int request_id,
+    const std::string& app_id);
+
 }  // namespace cast
 }  // namespace openscreen
 
diff --git a/cast/sender/testing/DEPS b/cast/sender/testing/DEPS
new file mode 100644
index 00000000..99039c59
--- /dev/null
+++ b/cast/sender/testing/DEPS
@@ -0,0 +1,4 @@
+include_rules = [
+  # Sender tests can use receiver code for simulation/validation.
+  '+cast/receiver',
+]
diff --git a/cast/sender/testing/test_helpers.cc b/cast/sender/testing/test_helpers.cc
new file mode 100644
index 00000000..ff9a4575
--- /dev/null
+++ b/cast/sender/testing/test_helpers.cc
@@ -0,0 +1,87 @@
+// Copyright 2020 The Chromium 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 "cast/sender/testing/test_helpers.h"
+
+#include "cast/common/channel/message_util.h"
+#include "cast/receiver/channel/message_util.h"
+#include "cast/sender/channel/message_util.h"
+#include "gtest/gtest.h"
+#include "util/json/json_serialization.h"
+#include "util/json/json_value.h"
+#include "util/logging.h"
+
+namespace openscreen {
+namespace cast {
+
+using ::cast::channel::CastMessage;
+
+void VerifyAppAvailabilityRequest(const CastMessage& message,
+                                  const std::string& expected_app_id,
+                                  int* request_id_out,
+                                  std::string* sender_id_out) {
+  std::string app_id_out;
+  VerifyAppAvailabilityRequest(message, &app_id_out, request_id_out,
+                               sender_id_out);
+  EXPECT_EQ(app_id_out, expected_app_id);
+}
+
+void VerifyAppAvailabilityRequest(const CastMessage& message,
+                                  std::string* app_id_out,
+                                  int* request_id_out,
+                                  std::string* sender_id_out) {
+  EXPECT_EQ(message.namespace_(), kReceiverNamespace);
+  EXPECT_EQ(message.destination_id(), kPlatformReceiverId);
+  EXPECT_EQ(message.payload_type(),
+            ::cast::channel::CastMessage_PayloadType_STRING);
+  EXPECT_NE(message.source_id(), kPlatformSenderId);
+  *sender_id_out = message.source_id();
+
+  ErrorOr<Json::Value> maybe_value = json::Parse(message.payload_utf8());
+  ASSERT_TRUE(maybe_value);
+  Json::Value& value = maybe_value.value();
+
+  absl::optional<absl::string_view> maybe_type =
+      MaybeGetString(value, JSON_EXPAND_FIND_CONSTANT_ARGS(kMessageKeyType));
+  ASSERT_TRUE(maybe_type);
+  EXPECT_EQ(maybe_type.value(),
+            CastMessageTypeToString(CastMessageType::kGetAppAvailability));
+
+  absl::optional<int> maybe_id =
+      MaybeGetInt(value, JSON_EXPAND_FIND_CONSTANT_ARGS(kMessageKeyRequestId));
+  ASSERT_TRUE(maybe_id);
+  *request_id_out = maybe_id.value();
+
+  const Json::Value* maybe_app_ids =
+      value.find(JSON_EXPAND_FIND_CONSTANT_ARGS(kMessageKeyAppId));
+  ASSERT_TRUE(maybe_app_ids);
+  ASSERT_TRUE(maybe_app_ids->isArray());
+  ASSERT_EQ(maybe_app_ids->size(), 1u);
+  Json::Value app_id_value = maybe_app_ids->get(0u, Json::Value(""));
+  absl::optional<absl::string_view> maybe_app_id = MaybeGetString(app_id_value);
+  ASSERT_TRUE(maybe_app_id);
+  *app_id_out =
+      std::string(maybe_app_id.value().begin(), maybe_app_id.value().end());
+}
+
+CastMessage CreateAppAvailableResponseChecked(int request_id,
+                                              const std::string& sender_id,
+                                              const std::string& app_id) {
+  ErrorOr<CastMessage> message =
+      CreateAppAvailableResponse(request_id, sender_id, app_id);
+  OSP_CHECK(message);
+  return std::move(message.value());
+}
+
+CastMessage CreateAppUnavailableResponseChecked(int request_id,
+                                                const std::string& sender_id,
+                                                const std::string& app_id) {
+  ErrorOr<CastMessage> message =
+      CreateAppUnavailableResponse(request_id, sender_id, app_id);
+  OSP_CHECK(message);
+  return std::move(message.value());
+}
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/testing/test_helpers.h b/cast/sender/testing/test_helpers.h
new file mode 100644
index 00000000..c9a68c20
--- /dev/null
+++ b/cast/sender/testing/test_helpers.h
@@ -0,0 +1,43 @@
+// Copyright 2020 The Chromium 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 CAST_SENDER_TESTING_TEST_HELPERS_H_
+#define CAST_SENDER_TESTING_TEST_HELPERS_H_
+
+#include <cstdint>
+#include <string>
+
+#include "cast/sender/channel/message_util.h"
+
+namespace cast {
+namespace channel {
+class CastMessage;
+}  // namespace channel
+}  // namespace cast
+
+namespace openscreen {
+namespace cast {
+
+void VerifyAppAvailabilityRequest(const ::cast::channel::CastMessage& message,
+                                  const std::string& expected_app_id,
+                                  int* request_id_out,
+                                  std::string* sender_id_out);
+void VerifyAppAvailabilityRequest(const ::cast::channel::CastMessage& message,
+                                  std::string* app_id_out,
+                                  int* request_id_out,
+                                  std::string* sender_id_out);
+
+::cast::channel::CastMessage CreateAppAvailableResponseChecked(
+    int request_id,
+    const std::string& sender_id,
+    const std::string& app_id);
+::cast::channel::CastMessage CreateAppUnavailableResponseChecked(
+    int request_id,
+    const std::string& sender_id,
+    const std::string& app_id);
+
+}  // namespace cast
+}  // namespace openscreen
+
+#endif  // CAST_SENDER_TESTING_TEST_HELPERS_H_