Add Cast app discovery mechanism

This change ports CastAppDiscoveryServiceImpl from Chromium, along with
necessary changes.  This includes Cast message request/response tracking
on top of VirtualConnectionRouter.

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

7 files changed, 788 insertions, 0 deletions

diff --git a/cast/sender/BUILD.gn b/cast/sender/BUILD.gn
index 86ecb7d6..f500f766 100644
--- a/cast/sender/BUILD.gn
+++ b/cast/sender/BUILD.gn
@@ -31,8 +31,12 @@ source_set("sender") {
   sources = [
     "cast_app_availability_tracker.cc",
     "cast_app_availability_tracker.h",
+    "cast_app_discovery_service_impl.cc",
+    "cast_app_discovery_service_impl.h",
     "cast_platform_client.cc",
     "cast_platform_client.h",
+    "public/cast_app_discovery_service.cc",
+    "public/cast_app_discovery_service.h",
     "public/cast_media_source.cc",
     "public/cast_media_source.h",
   ]
@@ -70,6 +74,7 @@ source_set("unittests") {
   testonly = true
   sources = [
     "cast_app_availability_tracker_unittest.cc",
+    "cast_app_discovery_service_impl_unittest.cc",
     "cast_platform_client_unittest.cc",
     "channel/cast_auth_util_unittest.cc",
   ]
diff --git a/cast/sender/cast_app_discovery_service_impl.cc b/cast/sender/cast_app_discovery_service_impl.cc
new file mode 100644
index 00000000..69028aac
--- /dev/null
+++ b/cast/sender/cast_app_discovery_service_impl.cc
@@ -0,0 +1,211 @@
+// 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_app_discovery_service_impl.h"
+
+#include <algorithm>
+#include <chrono>
+
+#include "cast/sender/public/cast_media_source.h"
+#include "util/logging.h"
+
+namespace openscreen {
+namespace cast {
+namespace {
+
+// The minimum time that must elapse before an app availability result can be
+// force refreshed.
+static constexpr std::chrono::minutes kRefreshThreshold =
+    std::chrono::minutes(1);
+
+}  // namespace
+
+CastAppDiscoveryServiceImpl::CastAppDiscoveryServiceImpl(
+    CastPlatformClient* platform_client,
+    ClockNowFunctionPtr clock)
+    : platform_client_(platform_client), clock_(clock), weak_factory_(this) {
+  OSP_DCHECK(platform_client_);
+  OSP_DCHECK(clock_);
+}
+
+CastAppDiscoveryServiceImpl::~CastAppDiscoveryServiceImpl() {
+  OSP_CHECK_EQ(avail_queries_.size(), 0u);
+}
+
+CastAppDiscoveryService::Subscription
+CastAppDiscoveryServiceImpl::StartObservingAvailability(
+    const CastMediaSource& source,
+    AvailabilityCallback callback) {
+  const std::string& source_id = source.source_id();
+
+  // Return cached results immediately, if available.
+  std::vector<std::string> cached_device_ids =
+      availability_tracker_.GetAvailableDevices(source);
+  if (!cached_device_ids.empty()) {
+    callback(source, GetReceiversByIds(cached_device_ids));
+  }
+
+  auto& callbacks = avail_queries_[source_id];
+  uint32_t query_id = GetNextAvailabilityQueryId();
+  callbacks.push_back({query_id, std::move(callback)});
+  if (callbacks.size() == 1) {
+    // NOTE: Even though we retain availability results for an app unregistered
+    // from the tracker, we will refresh the results when the app is
+    // re-registered.
+    std::vector<std::string> new_app_ids =
+        availability_tracker_.RegisterSource(source);
+    for (const auto& app_id : new_app_ids) {
+      for (const auto& entry : receivers_by_id_) {
+        RequestAppAvailability(entry.first, app_id);
+      }
+    }
+  }
+
+  return MakeSubscription(this, query_id);
+}
+
+void CastAppDiscoveryServiceImpl::Refresh() {
+  const auto app_ids = availability_tracker_.GetRegisteredApps();
+  for (const auto& entry : receivers_by_id_) {
+    for (const auto& app_id : app_ids) {
+      RequestAppAvailability(entry.first, app_id);
+    }
+  }
+}
+
+void CastAppDiscoveryServiceImpl::AddOrUpdateReceiver(
+    const ServiceInfo& receiver) {
+  const std::string& device_id = receiver.unique_id;
+  receivers_by_id_[device_id] = receiver;
+
+  // Any queries that currently contain this receiver should be updated.
+  UpdateAvailabilityQueries(
+      availability_tracker_.GetSupportedSources(device_id));
+
+  for (const std::string& app_id : availability_tracker_.GetRegisteredApps()) {
+    RequestAppAvailability(device_id, app_id);
+  }
+}
+
+void CastAppDiscoveryServiceImpl::RemoveReceiver(const ServiceInfo& receiver) {
+  const std::string& device_id = receiver.unique_id;
+  receivers_by_id_.erase(device_id);
+  UpdateAvailabilityQueries(
+      availability_tracker_.RemoveResultsForDevice(device_id));
+}
+
+void CastAppDiscoveryServiceImpl::RequestAppAvailability(
+    const std::string& device_id,
+    const std::string& app_id) {
+  if (ShouldRefreshAppAvailability(device_id, app_id, clock_())) {
+    platform_client_->RequestAppAvailability(
+        device_id, app_id,
+        [self = weak_factory_.GetWeakPtr(), device_id](
+            const std::string& app_id, AppAvailabilityResult availability) {
+          if (self) {
+            self->UpdateAppAvailability(device_id, app_id, availability);
+          }
+        });
+  }
+}
+
+void CastAppDiscoveryServiceImpl::UpdateAppAvailability(
+    const std::string& device_id,
+    const std::string& app_id,
+    AppAvailabilityResult availability) {
+  if (receivers_by_id_.find(device_id) == receivers_by_id_.end()) {
+    return;
+  }
+
+  OSP_DVLOG << "App " << app_id << " on receiver " << device_id << " is "
+            << ToString(availability);
+
+  UpdateAvailabilityQueries(availability_tracker_.UpdateAppAvailability(
+      device_id, app_id, {availability, clock_()}));
+}
+
+void CastAppDiscoveryServiceImpl::UpdateAvailabilityQueries(
+    const std::vector<CastMediaSource>& sources) {
+  for (const auto& source : sources) {
+    const std::string& source_id = source.source_id();
+    auto it = avail_queries_.find(source_id);
+    if (it == avail_queries_.end())
+      continue;
+    std::vector<std::string> device_ids =
+        availability_tracker_.GetAvailableDevices(source);
+    std::vector<ServiceInfo> receivers = GetReceiversByIds(device_ids);
+    for (const auto& callback : it->second) {
+      callback.callback(source, receivers);
+    }
+  }
+}
+
+std::vector<ServiceInfo> CastAppDiscoveryServiceImpl::GetReceiversByIds(
+    const std::vector<std::string>& device_ids) const {
+  std::vector<ServiceInfo> receivers;
+  for (const std::string& device_id : device_ids) {
+    auto entry = receivers_by_id_.find(device_id);
+    if (entry != receivers_by_id_.end()) {
+      receivers.push_back(entry->second);
+    }
+  }
+  return receivers;
+}
+
+bool CastAppDiscoveryServiceImpl::ShouldRefreshAppAvailability(
+    const std::string& device_id,
+    const std::string& app_id,
+    Clock::time_point now) const {
+  // TODO(btolsch): Consider an exponential backoff mechanism instead.
+  // Receivers will typically respond with "unavailable" immediately after boot
+  // and then become available 10-30 seconds later.
+  auto availability = availability_tracker_.GetAvailability(device_id, app_id);
+  switch (availability.availability) {
+    case AppAvailabilityResult::kAvailable:
+      return false;
+    case AppAvailabilityResult::kUnavailable:
+      return (now - availability.time) > kRefreshThreshold;
+    // TODO(btolsch): Should there be a background task for periodically
+    // refreshing kUnknown (or even kUnavailable) results?
+    case AppAvailabilityResult::kUnknown:
+      return true;
+  }
+
+  OSP_NOTREACHED();
+  return false;
+}
+
+uint32_t CastAppDiscoveryServiceImpl::GetNextAvailabilityQueryId() {
+  if (free_query_ids_.empty()) {
+    return next_avail_query_id_++;
+  } else {
+    uint32_t id = free_query_ids_.back();
+    free_query_ids_.pop_back();
+    return id;
+  }
+}
+
+void CastAppDiscoveryServiceImpl::RemoveAvailabilityCallback(uint32_t id) {
+  for (auto entry = avail_queries_.begin(); entry != avail_queries_.end();
+       ++entry) {
+    const std::string& source_id = entry->first;
+    auto& callbacks = entry->second;
+    auto it =
+        std::find_if(callbacks.begin(), callbacks.end(),
+                     [id](const AvailabilityCallbackEntry& callback_entry) {
+                       return callback_entry.id == id;
+                     });
+    if (it != callbacks.end()) {
+      callbacks.erase(it);
+      if (callbacks.empty()) {
+        availability_tracker_.UnregisterSource(source_id);
+        avail_queries_.erase(entry);
+      }
+      return;
+    }
+  }
+}
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/cast_app_discovery_service_impl.h b/cast/sender/cast_app_discovery_service_impl.h
new file mode 100644
index 00000000..4093311a
--- /dev/null
+++ b/cast/sender/cast_app_discovery_service_impl.h
@@ -0,0 +1,99 @@
+// 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_APP_DISCOVERY_SERVICE_IMPL_H_
+#define CAST_SENDER_CAST_APP_DISCOVERY_SERVICE_IMPL_H_
+
+#include <map>
+#include <string>
+#include <vector>
+
+#include "cast/common/public/service_info.h"
+#include "cast/sender/cast_app_availability_tracker.h"
+#include "cast/sender/cast_platform_client.h"
+#include "cast/sender/public/cast_app_discovery_service.h"
+#include "platform/api/time.h"
+#include "util/weak_ptr.h"
+
+namespace openscreen {
+namespace cast {
+
+// Keeps track of availability queries, receives receiver updates, and issues
+// app availability requests based on these signals.
+class CastAppDiscoveryServiceImpl : public CastAppDiscoveryService {
+ public:
+  // |platform_client| must outlive |this|.
+  CastAppDiscoveryServiceImpl(CastPlatformClient* platform_client,
+                              ClockNowFunctionPtr clock);
+  ~CastAppDiscoveryServiceImpl() override;
+
+  // CastAppDiscoveryService implementation.
+  Subscription StartObservingAvailability(
+      const CastMediaSource& source,
+      AvailabilityCallback callback) override;
+
+  // Reissues app availability requests for currently registered (device_id,
+  // app_id) pairs whose status is kUnavailable or kUnknown.
+  void Refresh() override;
+
+  void AddOrUpdateReceiver(const ServiceInfo& receiver);
+  void RemoveReceiver(const ServiceInfo& receiver);
+
+ private:
+  struct AvailabilityCallbackEntry {
+    uint32_t id;
+    AvailabilityCallback callback;
+  };
+
+  // Issues an app availability request for |app_id| to the receiver given by
+  // |device_id|.
+  void RequestAppAvailability(const std::string& device_id,
+                              const std::string& app_id);
+
+  // Updates the availability result for |device_id| and |app_id| with |result|,
+  // and notifies callbacks with updated availability query results.
+  void UpdateAppAvailability(const std::string& device_id,
+                             const std::string& app_id,
+                             AppAvailabilityResult result);
+
+  // Updates the availability query results for |sources|.
+  void UpdateAvailabilityQueries(const std::vector<CastMediaSource>& sources);
+
+  std::vector<ServiceInfo> GetReceiversByIds(
+      const std::vector<std::string>& device_ids) const;
+
+  // Returns true if an app availability request should be issued for
+  // |device_id| and |app_id|. |now| is used for checking whether previously
+  // cached results should be refreshed.
+  bool ShouldRefreshAppAvailability(const std::string& device_id,
+                                    const std::string& app_id,
+                                    Clock::time_point now) const;
+
+  uint32_t GetNextAvailabilityQueryId();
+
+  void RemoveAvailabilityCallback(uint32_t id) override;
+
+  std::map<std::string, ServiceInfo> receivers_by_id_;
+
+  // Registered availability queries and their associated callbacks keyed by
+  // media source IDs.
+  std::map<std::string, std::vector<AvailabilityCallbackEntry>> avail_queries_;
+
+  // Callback ID tracking.
+  uint32_t next_avail_query_id_;
+  std::vector<uint32_t> free_query_ids_;
+
+  CastPlatformClient* const platform_client_;
+
+  CastAppAvailabilityTracker availability_tracker_;
+
+  const ClockNowFunctionPtr clock_;
+
+  WeakPtrFactory<CastAppDiscoveryServiceImpl> weak_factory_;
+};
+
+}  // namespace cast
+}  // namespace openscreen
+
+#endif  // CAST_SENDER_CAST_APP_DISCOVERY_SERVICE_IMPL_H_
diff --git a/cast/sender/cast_app_discovery_service_impl_unittest.cc b/cast/sender/cast_app_discovery_service_impl_unittest.cc
new file mode 100644
index 00000000..bfe596a8
--- /dev/null
+++ b/cast/sender/cast_app_discovery_service_impl_unittest.cc
@@ -0,0 +1,348 @@
+// 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_app_discovery_service_impl.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 "gtest/gtest.h"
+#include "platform/test/fake_clock.h"
+#include "platform/test/fake_task_runner.h"
+#include "util/logging.h"
+
+namespace openscreen {
+namespace cast {
+
+using ::cast::channel::CastMessage;
+
+using ::testing::_;
+
+class CastAppDiscoveryServiceImplTest : public ::testing::Test {
+ public:
+  void SetUp() override {
+    socket_id_ = fake_cast_socket_pair_.socket->socket_id();
+    router_.TakeSocket(&mock_error_handler_,
+                       std::move(fake_cast_socket_pair_.socket));
+
+    receiver_.v4_endpoint = fake_cast_socket_pair_.remote_endpoint;
+    receiver_.unique_id = "deviceId1";
+    receiver_.friendly_name = "Some Name";
+  }
+
+ protected:
+  CastSocket& peer_socket() { return *fake_cast_socket_pair_.peer_socket; }
+  MockCastSocketClient& peer_client() {
+    return fake_cast_socket_pair_.mock_peer_client;
+  }
+
+  void AddOrUpdateReceiver(const ServiceInfo& receiver, int32_t socket_id) {
+    platform_client_.AddOrUpdateReceiver(receiver, socket_id);
+    app_discovery_service_.AddOrUpdateReceiver(receiver);
+  }
+
+  CastAppDiscoveryService::Subscription StartObservingAvailability(
+      const CastMediaSource& source,
+      std::vector<ServiceInfo>* save_receivers) {
+    return app_discovery_service_.StartObservingAvailability(
+        source, [save_receivers](const CastMediaSource& source,
+                                 const std::vector<ServiceInfo>& receivers) {
+          *save_receivers = receivers;
+        });
+  }
+
+  CastAppDiscoveryService::Subscription StartSourceA1Query(
+      std::vector<ServiceInfo>* receivers,
+      int* request_id,
+      std::string* sender_id) {
+    auto subscription = StartObservingAvailability(source_a_1_, receivers);
+
+    // Adding a receiver after app registered causes app availability request to
+    // be sent.
+    *request_id = -1;
+    *sender_id = "";
+    EXPECT_CALL(peer_client(), OnMessage(_, _))
+        .WillOnce([request_id, sender_id](CastSocket*, CastMessage message) {
+          VerifyAppAvailabilityRequest(message, "AAA", request_id, sender_id);
+        });
+
+    AddOrUpdateReceiver(receiver_, socket_id_);
+
+    return subscription;
+  }
+
+  FakeCastSocketPair fake_cast_socket_pair_;
+  int32_t socket_id_;
+  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_};
+  CastAppDiscoveryServiceImpl app_discovery_service_{&platform_client_,
+                                                     &FakeClock::now};
+
+  CastMediaSource source_a_1_{"cast:AAA?clientId=1", {"AAA"}};
+  CastMediaSource source_a_2_{"cast:AAA?clientId=2", {"AAA"}};
+  CastMediaSource source_b_1_{"cast:BBB?clientId=1", {"BBB"}};
+
+  ServiceInfo receiver_;
+};
+
+TEST_F(CastAppDiscoveryServiceImplTest, StartObservingAvailability) {
+  std::vector<ServiceInfo> receivers1;
+  int request_id;
+  std::string sender_id;
+  auto subscription1 = StartSourceA1Query(&receivers1, &request_id, &sender_id);
+
+  // Same app ID should not trigger another request.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  std::vector<ServiceInfo> receivers2;
+  auto subscription2 = StartObservingAvailability(source_a_2_, &receivers2);
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  ASSERT_EQ(receivers1.size(), 1u);
+  ASSERT_EQ(receivers2.size(), 1u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+  EXPECT_EQ(receivers2[0].unique_id, "deviceId1");
+
+  // No more updates for |source_a_1_| (i.e. |receivers1|).
+  subscription1.Reset();
+  platform_client_.RemoveReceiver(receiver_);
+  app_discovery_service_.RemoveReceiver(receiver_);
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers2.size(), 0u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest, ReAddAvailQueryUsesCachedValue) {
+  std::vector<ServiceInfo> receivers1;
+  int request_id;
+  std::string sender_id;
+  auto subscription1 = StartSourceA1Query(&receivers1, &request_id, &sender_id);
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+
+  subscription1.Reset();
+  receivers1.clear();
+
+  // Request not re-sent; cached kAvailable value is used.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  subscription1 = StartObservingAvailability(source_a_1_, &receivers1);
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest, AvailQueryUpdatedOnReceiverUpdate) {
+  std::vector<ServiceInfo> receivers1;
+  int request_id;
+  std::string sender_id;
+  auto subscription1 = StartSourceA1Query(&receivers1, &request_id, &sender_id);
+
+  // Result set now includes |receiver_|.
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+
+  // Updating |receiver_| causes |source_a_1_| query to be updated, but it's too
+  // soon for a new message to be sent.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  receiver_.friendly_name = "New Name";
+
+  AddOrUpdateReceiver(receiver_, socket_id_);
+
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers1[0].friendly_name, "New Name");
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest, Refresh) {
+  std::vector<ServiceInfo> receivers1;
+  auto subscription1 = StartObservingAvailability(source_a_1_, &receivers1);
+  std::vector<ServiceInfo> receivers2;
+  auto subscription2 = StartObservingAvailability(source_b_1_, &receivers2);
+
+  // Adding a receiver after app registered causes two separate app availability
+  // requests to be sent.
+  int request_idA = -1;
+  int request_idB = -1;
+  std::string sender_id = "";
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .Times(2)
+      .WillRepeatedly([&request_idA, &request_idB, &sender_id](
+                          CastSocket*, CastMessage message) {
+        std::string app_id;
+        int request_id = -1;
+        VerifyAppAvailabilityRequest(message, &app_id, &request_id, &sender_id);
+        if (app_id == "AAA") {
+          EXPECT_EQ(request_idA, -1);
+          request_idA = request_id;
+        } else if (app_id == "BBB") {
+          EXPECT_EQ(request_idB, -1);
+          request_idB = request_id;
+        } else {
+          EXPECT_TRUE(false);
+        }
+      });
+
+  AddOrUpdateReceiver(receiver_, socket_id_);
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_idA, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  availability_response =
+      CreateAppUnavailableResponseChecked(request_idB, sender_id, "BBB");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  ASSERT_EQ(receivers1.size(), 1u);
+  ASSERT_EQ(receivers2.size(), 0u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+
+  // Not enough time has passed for a refresh.
+  clock_.Advance(std::chrono::seconds(30));
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  app_discovery_service_.Refresh();
+
+  // Refresh will now query again for unavailable app IDs.
+  clock_.Advance(std::chrono::minutes(2));
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_idB, &sender_id](CastSocket*, CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "BBB", &request_idB, &sender_id);
+      });
+  app_discovery_service_.Refresh();
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest,
+       StartObservingAvailabilityAfterReceiverAdded) {
+  // No registered apps.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  AddOrUpdateReceiver(receiver_, socket_id_);
+
+  // Registering apps immediately sends requests to |receiver_|.
+  int request_idA = -1;
+  std::string sender_id = "";
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_idA, &sender_id](CastSocket*, CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "AAA", &request_idA, &sender_id);
+      });
+  std::vector<ServiceInfo> receivers1;
+  auto subscription1 = StartObservingAvailability(source_a_1_, &receivers1);
+
+  int request_idB = -1;
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_idB, &sender_id](CastSocket*, CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "BBB", &request_idB, &sender_id);
+      });
+  std::vector<ServiceInfo> receivers2;
+  auto subscription2 = StartObservingAvailability(source_b_1_, &receivers2);
+
+  // Add a new receiver with a corresponding socket.
+  FakeCastSocketPair fake_sockets2({{192, 168, 1, 17}, 2345},
+                                   {{192, 168, 1, 19}, 2345});
+  CastSocket* socket2 = fake_sockets2.socket.get();
+  router_.TakeSocket(&mock_error_handler_, std::move(fake_sockets2.socket));
+  ServiceInfo receiver2;
+  receiver2.unique_id = "deviceId2";
+  receiver2.v4_endpoint = fake_sockets2.remote_endpoint;
+
+  // Adding new receiver causes availability requests for both apps to be sent
+  // to the new receiver.
+  request_idA = -1;
+  request_idB = -1;
+  EXPECT_CALL(fake_sockets2.mock_peer_client, OnMessage(_, _))
+      .Times(2)
+      .WillRepeatedly([&request_idA, &request_idB, &sender_id](
+                          CastSocket*, CastMessage message) {
+        std::string app_id;
+        int request_id = -1;
+        VerifyAppAvailabilityRequest(message, &app_id, &request_id, &sender_id);
+        if (app_id == "AAA") {
+          EXPECT_EQ(request_idA, -1);
+          request_idA = request_id;
+        } else if (app_id == "BBB") {
+          EXPECT_EQ(request_idB, -1);
+          request_idB = request_id;
+        } else {
+          EXPECT_TRUE(false);
+        }
+      });
+
+  AddOrUpdateReceiver(receiver2, socket2->socket_id());
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest, StartObservingAvailabilityCachedValue) {
+  std::vector<ServiceInfo> receivers1;
+  int request_id;
+  std::string sender_id;
+  auto subscription1 = StartSourceA1Query(&receivers1, &request_id, &sender_id);
+
+  CastMessage availability_response =
+      CreateAppAvailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+  ASSERT_EQ(receivers1.size(), 1u);
+  EXPECT_EQ(receivers1[0].unique_id, "deviceId1");
+
+  // Same app ID should not trigger another request, but it should return
+  // cached value.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  std::vector<ServiceInfo> receivers2;
+  auto subscription2 = StartObservingAvailability(source_a_2_, &receivers2);
+  ASSERT_EQ(receivers2.size(), 1u);
+  EXPECT_EQ(receivers2[0].unique_id, "deviceId1");
+}
+
+TEST_F(CastAppDiscoveryServiceImplTest, AvailabilityUnknownOrUnavailable) {
+  std::vector<ServiceInfo> receivers1;
+  int request_id;
+  std::string sender_id;
+  auto subscription1 = StartSourceA1Query(&receivers1, &request_id, &sender_id);
+
+  // The request will timeout resulting in unknown app availability.
+  clock_.Advance(std::chrono::seconds(10));
+  task_runner_.RunTasksUntilIdle();
+  EXPECT_EQ(receivers1.size(), 0u);
+
+  // Receiver updated together with unknown app availability will cause a
+  // request to be sent again.
+  request_id = -1;
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_id, &sender_id](CastSocket*, CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "AAA", &request_id, &sender_id);
+      });
+  AddOrUpdateReceiver(receiver_, socket_id_);
+
+  CastMessage availability_response =
+      CreateAppUnavailableResponseChecked(request_id, sender_id, "AAA");
+  EXPECT_TRUE(peer_socket().SendMessage(availability_response).ok());
+
+  // Known availability so no request sent.
+  EXPECT_CALL(peer_client(), OnMessage(_, _)).Times(0);
+  AddOrUpdateReceiver(receiver_, socket_id_);
+
+  // Removing the receiver will also remove previous availability information.
+  // Next time the receiver is added, a new request will be sent.
+  platform_client_.RemoveReceiver(receiver_);
+  app_discovery_service_.RemoveReceiver(receiver_);
+
+  request_id = -1;
+  EXPECT_CALL(peer_client(), OnMessage(_, _))
+      .WillOnce([&request_id, &sender_id](CastSocket*, CastMessage message) {
+        VerifyAppAvailabilityRequest(message, "AAA", &request_id, &sender_id);
+      });
+
+  AddOrUpdateReceiver(receiver_, socket_id_);
+}
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/public/DEPS b/cast/sender/public/DEPS
index 3cfae918..8a61a258 100644
--- a/cast/sender/public/DEPS
+++ b/cast/sender/public/DEPS
@@ -4,4 +4,6 @@ include_rules = [
   # Dependencies on the implementation are not allowed in public/.
   '-cast/sender',
   '+cast/sender/public',
+  '-cast/common',
+  '+cast/common/public',
 ]
diff --git a/cast/sender/public/cast_app_discovery_service.cc b/cast/sender/public/cast_app_discovery_service.cc
new file mode 100644
index 00000000..c561b386
--- /dev/null
+++ b/cast/sender/public/cast_app_discovery_service.cc
@@ -0,0 +1,48 @@
+// 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/public/cast_app_discovery_service.h"
+
+namespace openscreen {
+namespace cast {
+
+CastAppDiscoveryService::Subscription::Subscription(
+    CastAppDiscoveryService* discovery_service,
+    uint32_t id)
+    : discovery_service_(discovery_service), id_(id) {}
+
+CastAppDiscoveryService::Subscription::Subscription(Subscription&& other)
+    : discovery_service_(other.discovery_service_), id_(other.id_) {
+  other.discovery_service_ = nullptr;
+}
+
+CastAppDiscoveryService::Subscription::~Subscription() {
+  Reset();
+}
+
+CastAppDiscoveryService::Subscription& CastAppDiscoveryService::Subscription::
+operator=(Subscription other) {
+  Swap(other);
+  return *this;
+}
+
+void CastAppDiscoveryService::Subscription::Reset() {
+  if (discovery_service_) {
+    discovery_service_->RemoveAvailabilityCallback(id_);
+  }
+  discovery_service_ = nullptr;
+}
+
+void CastAppDiscoveryService::Subscription::Swap(Subscription& other) {
+  CastAppDiscoveryService* service = other.discovery_service_;
+  other.discovery_service_ = discovery_service_;
+  discovery_service_ = service;
+
+  uint32_t id = other.id_;
+  other.id_ = id_;
+  id_ = id;
+}
+
+}  // namespace cast
+}  // namespace openscreen
diff --git a/cast/sender/public/cast_app_discovery_service.h b/cast/sender/public/cast_app_discovery_service.h
new file mode 100644
index 00000000..ccb586ff
--- /dev/null
+++ b/cast/sender/public/cast_app_discovery_service.h
@@ -0,0 +1,75 @@
+// 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_PUBLIC_CAST_APP_DISCOVERY_SERVICE_H_
+#define CAST_SENDER_PUBLIC_CAST_APP_DISCOVERY_SERVICE_H_
+
+#include <vector>
+
+#include "cast/common/public/service_info.h"
+
+namespace openscreen {
+namespace cast {
+
+class CastMediaSource;
+
+// Interface for app discovery for Cast devices.
+class CastAppDiscoveryService {
+ public:
+  using AvailabilityCallback =
+      std::function<void(const CastMediaSource& source,
+                         const std::vector<ServiceInfo>& devices)>;
+
+  class Subscription {
+   public:
+    Subscription(Subscription&&);
+    ~Subscription();
+    Subscription& operator=(Subscription);
+
+    void Reset();
+
+   private:
+    friend class CastAppDiscoveryService;
+
+    Subscription(CastAppDiscoveryService* discovery_service, uint32_t id);
+
+    void Swap(Subscription& other);
+
+    CastAppDiscoveryService* discovery_service_;
+    uint32_t id_;
+  };
+
+  virtual ~CastAppDiscoveryService() = default;
+
+  // Adds an availability query for |source|. Results will be continuously
+  // returned via |callback| until the returned Subscription is destroyed by the
+  // caller.  If there are cached results available, |callback| will be invoked
+  // before this method returns.  |callback| may be invoked with an empty list
+  // if all devices respond to the respective queries with "unavailable" or
+  // don't respond before a timeout.  |callback| may be invoked successively
+  // with the same list.
+  virtual Subscription StartObservingAvailability(
+      const CastMediaSource& source,
+      AvailabilityCallback callback) = 0;
+
+  // Refreshes the state of app discovery in the service. It is suitable to call
+  // this method when the user initiates a user gesture.
+  virtual void Refresh() = 0;
+
+ protected:
+  Subscription MakeSubscription(CastAppDiscoveryService* discovery_service,
+                                uint32_t id) {
+    return Subscription(discovery_service, id);
+  }
+
+ private:
+  friend class Subscription;
+
+  virtual void RemoveAvailabilityCallback(uint32_t id) = 0;
+};
+
+}  // namespace cast
+}  // namespace openscreen
+
+#endif  // CAST_SENDER_PUBLIC_CAST_APP_DISCOVERY_SERVICE_H_