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>

1 files changed, 224 insertions, 0 deletions

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