Add implementation of a UDPSocket client for cast streaming.

Add a UDP transport that can send/receive UDP packets through Network
Service by connecting to a UDPSocket that implements
network::mojom::UDPSocket interface.

Bug: 734672
Change-Id: Ia2ccf456eea85394e01be919eee645684ba7625a

TBR=jam@chromium.org

Change-Id: Ia2ccf456eea85394e01be919eee645684ba7625a
Reviewed-on: https://chromium-review.googlesource.com/910127
Commit-Queue: Xiangjun Zhang <xjz@chromium.org>
Reviewed-by: Christian Dullweber <dullweber@chromium.org>
Reviewed-by: Yuri Wiitala <miu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#537895}

media/cast/BUILD.gn

@@ -104,6 +104,8 @@ source_set("net") {
     "net/rtp/rtp_sender.h",
     "net/udp_packet_pipe.cc",
     "net/udp_packet_pipe.h",
+    "net/udp_socket_client.cc",
+    "net/udp_socket_client.h",
     "net/udp_transport_impl.cc",
     "net/udp_transport_impl.h",
     "net/udp_transport_interface.h",
@@ -115,6 +117,7 @@ source_set("net") {
     "//media/mojo/common",
     "//mojo/public/cpp/system",
     "//net",
+    "//services/network/public/mojom",
   ]
 
   public_deps = [
@@ -317,6 +320,7 @@ test("cast_unittests") {
     "net/rtp/rtp_packetizer_unittest.cc",
     "net/rtp/rtp_parser_unittest.cc",
     "net/udp_packet_pipe_unittest.cc",
+    "net/udp_socket_client_unittest.cc",
     "net/udp_transport_unittest.cc",
     "receiver/audio_decoder_unittest.cc",
     "receiver/frame_receiver_unittest.cc",
@@ -345,7 +349,9 @@ test("cast_unittests") {
     "//base/test:test_support",
     "//media:test_support",
     "//media/test:run_all_unittests",
+    "//mojo/public/cpp/bindings",
     "//net",
+    "//services/network/public/mojom",
     "//testing/gmock",
     "//testing/gtest",
     "//third_party/opus",

media/cast/net/DEPS

@@ -7,8 +7,9 @@ include_rules = [
   "+media/cast/logging",
   "+media/cast/net",
   "+media/mojo/common",
-  "+mojo/public/cpp/system",
+  "+mojo/public/cpp",
   "+net",
+  "+services/network/public/mojom",
 ]
 
 specific_include_rules = {

media/cast/net/udp_socket_client.cc

+// Copyright 2018 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 "media/cast/net/udp_socket_client.h"
+
+#include "base/callback.h"
+#include "media/cast/net/udp_packet_pipe.h"
+#include "net/base/address_family.h"
+#include "net/traffic_annotation/network_traffic_annotation.h"
+
+namespace media {
+namespace cast {
+
+namespace {
+
+// The minimal number of packets asking for receiving.
+constexpr int kNumPacketsAsking = 1024;
+
+net::NetworkTrafficAnnotationTag GetNetworkTrafficAnnotationTag() {
+  return net::DefineNetworkTrafficAnnotation("cast_udp_socket", R"(
+      semantics {
+        sender: "Cast Streaming"
+        description:
+          "Media streaming protocol for LAN transport of screen mirroring "
+          "audio/video. This is also used by browser features that wish to "
+          "send browser content for remote display, and such features are "
+          "generally started/stopped from the Media Router dialog."
+        trigger:
+          "User invokes feature from the Media Router dialog (right click on "
+          "page, 'Cast...')."
+        data:
+          "Media and related protocol-level control and performance messages."
+        destination: OTHER
+        destination_other:
+          "A playback receiver, such as a Chromecast device."
+      }
+      policy {
+        cookies_allowed: NO
+        setting: "This feature cannot be disabled in settings."
+        chrome_policy {
+          EnableMediaRouter {
+            EnableMediaRouter: false
+          }
+        }
+      })");
+}
+
+}  // namespace
+
+UdpSocketClient::UdpSocketClient(const net::IPEndPoint& remote_endpoint,
+                                 network::mojom::NetworkContextPtr context,
+                                 base::OnceClosure error_callback)
+    : remote_endpoint_(remote_endpoint),
+      network_context_(std::move(context)),
+      error_callback_(std::move(error_callback)),
+      binding_(this),
+      bytes_sent_(0),
+      allow_sending_(false),
+      num_packets_pending_receive_(0),
+      weak_factory_(this) {}
+
+UdpSocketClient::~UdpSocketClient() {}
+
+bool UdpSocketClient::SendPacket(cast::PacketRef packet,
+                                 const base::RepeatingClosure& cb) {
+  DVLOG(3) << __func__;
+  DCHECK(resume_send_callback_.is_null());
+
+  bytes_sent_ += packet->data.size();
+  if (!allow_sending_) {
+    resume_send_callback_ = cb;
+    return false;
+  }
+
+  DCHECK(udp_socket_);
+  udp_socket_->Send(
+      packet->data,
+      net::MutableNetworkTrafficAnnotationTag(GetNetworkTrafficAnnotationTag()),
+      base::BindOnce(&UdpSocketClient::OnPacketSent,
+                     weak_factory_.GetWeakPtr()));
+  return true;
+}
+
+void UdpSocketClient::OnPacketSent(int result) {
+  if (result != net::OK)
+    VLOG(2) << __func__ << ": error=" << result;
+
+  // Block the further sending if too many send requests are pending.
+  if (result == net::ERR_INSUFFICIENT_RESOURCES) {
+    allow_sending_ = false;
+    return;
+  }
+
+  allow_sending_ = true;
+  if (!resume_send_callback_.is_null())
+    std::move(resume_send_callback_).Run();
+}
+
+int64_t UdpSocketClient::GetBytesSent() {
+  return bytes_sent_;
+}
+
+void UdpSocketClient::StartReceiving(
+    const cast::PacketReceiverCallbackWithStatus& packet_receiver) {
+  DVLOG(1) << __func__;
+  packet_receiver_callback_ = packet_receiver;
+  network::mojom::UDPSocketReceiverPtr udp_socket_receiver;
+  binding_.Bind(mojo::MakeRequest(&udp_socket_receiver));
+  network_context_->CreateUDPSocket(mojo::MakeRequest(&udp_socket_),
+                                    std::move(udp_socket_receiver));
+  network::mojom::UDPSocketOptionsPtr options;
+  udp_socket_->Connect(remote_endpoint_, std::move(options),
+                       base::BindOnce(&UdpSocketClient::OnSocketConnected,
+                                      weak_factory_.GetWeakPtr()));
+}
+
+void UdpSocketClient::OnSocketConnected(
+    int result,
+    const base::Optional<net::IPEndPoint>& addr) {
+  DVLOG(2) << __func__ << ": result=" << result;
+
+  if (result == net::OK) {
+    allow_sending_ = true;
+    if (!resume_send_callback_.is_null())
+      std::move(resume_send_callback_).Run();
+  } else {
+    allow_sending_ = false;
+    VLOG(1) << "Socket connect error=" << result;
+    if (!error_callback_.is_null())
+      std::move(error_callback_).Run();
+    return;
+  }
+
+  if (!packet_receiver_callback_.is_null()) {
+    udp_socket_->ReceiveMore(kNumPacketsAsking);
+    num_packets_pending_receive_ = kNumPacketsAsking;
+  }
+}
+
+void UdpSocketClient::StopReceiving() {
+  packet_receiver_callback_.Reset();
+  if (binding_.is_bound())
+    binding_.Close();
+  if (udp_socket_.is_bound())
+    udp_socket_.reset();
+  num_packets_pending_receive_ = 0;
+}
+
+void UdpSocketClient::OnReceived(
+    int32_t result,
+    const base::Optional<net::IPEndPoint>& src_addr,
+    base::Optional<base::span<const uint8_t>> data) {
+  DVLOG(3) << __func__ << ": result=" << result;
+  DCHECK_GT(num_packets_pending_receive_, 0);
+  DCHECK(!packet_receiver_callback_.is_null());
+
+  --num_packets_pending_receive_;
+  if (num_packets_pending_receive_ < kNumPacketsAsking) {
+    udp_socket_->ReceiveMore(kNumPacketsAsking);
+    num_packets_pending_receive_ += kNumPacketsAsking;
+  }
+  if (result != net::OK)
+    return;
+  std::unique_ptr<cast::Packet> packet(
+      new cast::Packet(data->begin(), data->end()));
+  packet_receiver_callback_.Run(std::move(packet));
+}
+
+}  // namespace cast
+}  // namespace media

media/cast/net/udp_socket_client.h

+// Copyright 2018 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 MEDIA_CAST_NET_UDP_SOCKET_CLIENT_H_
+#define MEDIA_CAST_NET_UDP_SOCKET_CLIENT_H_
+
+#include "base/callback.h"
+#include "media/cast/net/cast_transport_config.h"
+#include "mojo/public/cpp/bindings/binding.h"
+#include "net/base/ip_endpoint.h"
+#include "services/network/public/mojom/network_service.mojom.h"
+#include "services/network/public/mojom/udp_socket.mojom.h"
+
+namespace media {
+namespace cast {
+
+// This class implements a UDP packet transport that can send/receive UDP
+// packets to/from |remote_endpoint_| through a UDPSocket that implements
+// network::mojom::UDPSocket. The UDPSocket is created and connected to
+// |remote_endpoint_| when StartReceiving() is called. Sending/Receiving ends
+// when StopReceiving() is called or this class is destructed. |error_callback|
+// will be called if the UDPSocket is failed to be created or connected.
+class UdpSocketClient final : public PacketTransport,
+                              public network::mojom::UDPSocketReceiver {
+ public:
+  UdpSocketClient(const net::IPEndPoint& remote_endpoint,
+                  network::mojom::NetworkContextPtr context,
+                  base::OnceClosure error_callback);
+
+  ~UdpSocketClient() override;
+
+  // cast::PacketTransport implementations.
+  bool SendPacket(cast::PacketRef packet,
+                  const base::RepeatingClosure& cb) override;
+  int64_t GetBytesSent() override;
+  void StartReceiving(
+      const cast::PacketReceiverCallbackWithStatus& packet_receiver) override;
+  void StopReceiving() override;
+
+  // network::mojom::UDPSocketReceiver implementation.
+  void OnReceived(int32_t result,
+                  const base::Optional<net::IPEndPoint>& src_addr,
+                  base::Optional<base::span<const uint8_t>> data) override;
+
+ private:
+  // The callback of network::mojom::UDPSocket::Send(). Further sending is
+  // blocked if |result| indicates that pending send requests buffer is full.
+  void OnPacketSent(int result);
+
+  // The callback of network::mojom::UDPSocket::Connect(). Packets are only
+  // allowed to send after the socket is successfully connected to the
+  // |remote_endpoint_|.
+  void OnSocketConnected(int result,
+                         const base::Optional<net::IPEndPoint>& addr);
+
+  const net::IPEndPoint remote_endpoint_;
+  const network::mojom::NetworkContextPtr network_context_;
+  base::OnceClosure error_callback_;
+
+  mojo::Binding<network::mojom::UDPSocketReceiver> binding_;
+
+  // The callback to deliver the received packets to the packet parser. Set
+  // when StartReceiving() is called.
+  cast::PacketReceiverCallbackWithStatus packet_receiver_callback_;
+
+  network::mojom::UDPSocketPtr udp_socket_;
+
+  // Set by SendPacket() when the sending is not allowed. Once set, SendPacket()
+  // can only be called again when a previous sending completes successfully.
+  // TODO(xjz): Change the callback to a base::OnceClosure as well as in the
+  // cast::PacketTransport SendPacket().
+  base::RepeatingClosure resume_send_callback_;
+
+  // Total numbe of bytes written to the data pipe.
+  int64_t bytes_sent_;
+
+  // Indicates whether further sending is allowed. Set to true when the
+  // UDPSocket is successfully connected to |remote_endpoint_|, and false when
+  // too many send requests are pending.
+  bool allow_sending_;
+
+  // The number of packets pending to receive.
+  int num_packets_pending_receive_;
+
+  base::WeakPtrFactory<UdpSocketClient> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(UdpSocketClient);
+};
+
+}  // namespace cast
+}  // namespace media
+
+#endif  // MEDIA_CAST_NET_UDP_SOCKET_CLIENT_H_

media/cast/net/udp_socket_client_unittest.cc

+// Copyright 2018 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 "media/cast/net/udp_socket_client.h"
+
+#include <algorithm>
+#include <string>
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/run_loop.h"
+#include "base/test/mock_callback.h"
+#include "base/test/scoped_task_environment.h"
+#include "base/time/time.h"
+#include "media/cast/net/cast_transport_config.h"
+#include "media/cast/test/utility/net_utility.h"
+#include "mojo/public/cpp/bindings/binding.h"
+#include "net/base/ip_endpoint.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using ::testing::InvokeWithoutArgs;
+
+namespace media {
+namespace cast {
+
+namespace {
+
+class MockUdpSocket final : public network::mojom::UDPSocket {
+ public:
+  MockUdpSocket(network::mojom::UDPSocketRequest request,
+                network::mojom::UDPSocketReceiverPtr receiver)
+      : binding_(this, std::move(request)), receiver_(std::move(receiver)) {}
+
+  ~MockUdpSocket() override {}
+
+  MOCK_METHOD0(OnSend, void());
+
+  // network::mojom::UDPSocket implementation.
+  void Connect(const net::IPEndPoint& remote_addr,
+               network::mojom::UDPSocketOptionsPtr options,
+               ConnectCallback callback) override {
+    std::move(callback).Run(net::OK, test::GetFreeLocalPort());
+  }
+
+  void Bind(const net::IPEndPoint& local_addr,
+            network::mojom::UDPSocketOptionsPtr options,
+            BindCallback callback) override {}
+  void SetBroadcast(bool broadcast, SetBroadcastCallback callback) override {}
+  void JoinGroup(const net::IPAddress& group_address,
+                 JoinGroupCallback callback) override {}
+  void LeaveGroup(const net::IPAddress& group_address,
+                  LeaveGroupCallback callback) override {}
+
+  void ReceiveMore(uint32_t num_additional_datagrams) override {
+    num_ask_for_receive_ += num_additional_datagrams;
+  }
+
+  void ReceiveMoreWithBufferSize(uint32_t num_additional_datagrams,
+                                 uint32_t buffer_size) override {}
+  void SendTo(const net::IPEndPoint& dest_addr,
+              base::span<const uint8_t> data,
+              const net::MutableNetworkTrafficAnnotationTag& traffic_annotation,
+              SendToCallback callback) override {}
+
+  void Send(base::span<const uint8_t> data,
+            const net::MutableNetworkTrafficAnnotationTag& traffic_annotation,
+            SendCallback callback) override {
+    sending_packet_ = std::make_unique<Packet>(data.begin(), data.end());
+    std::move(callback).Run(net::OK);
+    OnSend();
+  }
+
+  void Close() override {}
+
+  // Simulate receiving a packet from the network.
+  void OnReceivedPacket(const Packet& packet) {
+    if (num_ask_for_receive_) {
+      receiver_->OnReceived(
+          net::OK, base::nullopt,
+          base::span<const uint8_t>(
+              reinterpret_cast<const uint8_t*>(packet.data()), packet.size()));
+      ASSERT_LT(0, num_ask_for_receive_);
+      --num_ask_for_receive_;
+    }
+  }
+
+  void VerifySendingPacket(const Packet& packet) {
+    EXPECT_TRUE(
+        std::equal(packet.begin(), packet.end(), sending_packet_->begin()));
+  }
+
+ private:
+  mojo::Binding<network::mojom::UDPSocket> binding_;
+  network::mojom::UDPSocketReceiverPtr receiver_;
+  std::unique_ptr<Packet> sending_packet_;
+  int num_ask_for_receive_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(MockUdpSocket);
+};
+
+class MockNetworkContext final : public network::mojom::NetworkContext {
+ public:
+  explicit MockNetworkContext(network::mojom::NetworkContextRequest request)
+      : binding_(this, std::move(request)) {}
+  ~MockNetworkContext() override {}
+
+  // network::mojom::NetworkContext implementation:
+  void CreateURLLoaderFactory(network::mojom::URLLoaderFactoryRequest request,
+                              uint32_t process_id) override {}
+  void HandleViewCacheRequest(
+      const GURL& url,
+      network::mojom::URLLoaderClientPtr client) override {}
+  void GetCookieManager(network::mojom::CookieManagerRequest request) override {
+  }
+  void GetRestrictedCookieManager(
+      network::mojom::RestrictedCookieManagerRequest request,
+      int32_t render_process_id,
+      int32_t render_frame_id) override {}
+  void ClearNetworkingHistorySince(
+      base::Time time,
+      base::OnceClosure completion_callback) override {}
+  void SetNetworkConditions(
+      const std::string& profile_id,
+      network::mojom::NetworkConditionsPtr conditions) override {}
+  void AddHSTSForTesting(const std::string& host,
+                         base::Time expiry,
+                         bool include_subdomains,
+                         AddHSTSForTestingCallback callback) override {}
+
+  MOCK_METHOD0(OnUDPSocketCreated, void());
+  void CreateUDPSocket(network::mojom::UDPSocketRequest request,
+                       network::mojom::UDPSocketReceiverPtr receiver) override {
+    udp_socket_ = std::make_unique<MockUdpSocket>(std::move(request),
+                                                  std::move(receiver));
+    OnUDPSocketCreated();
+  }
+
+  MockUdpSocket* udp_socket() const { return udp_socket_.get(); };
+
+ private:
+  mojo::Binding<network::mojom::NetworkContext> binding_;
+  std::unique_ptr<MockUdpSocket> udp_socket_;
+  DISALLOW_COPY_AND_ASSIGN(MockNetworkContext);
+};
+
+}  // namespace
+
+class UdpSocketClientTest : public ::testing::Test {
+ public:
+  UdpSocketClientTest() {
+    network::mojom::NetworkContextPtr network_context_ptr;
+    network_context_ = std::make_unique<MockNetworkContext>(
+        mojo::MakeRequest(&network_context_ptr));
+    udp_transport_client_ = std::make_unique<UdpSocketClient>(
+        test::GetFreeLocalPort(), std::move(network_context_ptr),
+        base::OnceClosure());
+  }
+
+  ~UdpSocketClientTest() override = default;
+
+  MOCK_METHOD0(OnReceivedPacketCall, void());
+  bool OnReceivedPacket(std::unique_ptr<Packet> packet) {
+    received_packet_ = std::move(packet);
+    OnReceivedPacketCall();
+    return true;
+  }
+
+ protected:
+  base::test::ScopedTaskEnvironment scoped_task_environment_;
+  std::unique_ptr<MockNetworkContext> network_context_;
+  std::unique_ptr<UdpSocketClient> udp_transport_client_;
+  std::unique_ptr<Packet> received_packet_;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(UdpSocketClientTest);
+};
+
+TEST_F(UdpSocketClientTest, SendAndReceive) {
+  std::string data = "Test";
+  Packet packet(data.begin(), data.end());
+
+  {
+    // Expect the UDPSocket to be created when calling StartReceiving().
+    base::RunLoop run_loop;
+    EXPECT_CALL(*network_context_, OnUDPSocketCreated())
+        .WillOnce(InvokeWithoutArgs(&run_loop, &base::RunLoop::Quit));
+    udp_transport_client_->StartReceiving(base::BindRepeating(
+        &UdpSocketClientTest::OnReceivedPacket, base::Unretained(this)));
+    run_loop.Run();
+  }
+  scoped_task_environment_.RunUntilIdle();
+
+  MockUdpSocket* socket = network_context_->udp_socket();
+
+  {
+    // Request to send one packet.
+    base::RunLoop run_loop;
+    base::RepeatingClosure cb;
+    EXPECT_CALL(*socket, OnSend())
+        .WillOnce(InvokeWithoutArgs(&run_loop, &base::RunLoop::Quit));
+    EXPECT_TRUE(udp_transport_client_->SendPacket(
+        new base::RefCountedData<Packet>(packet), cb));
+    run_loop.Run();
+  }
+
+  // Expect the packet to be sent is delivered to the UDPSocket.
+  socket->VerifySendingPacket(packet);
+
+  // Test receiving packet.
+  std::string data2 = "Hello";
+  Packet packet2(data.begin(), data.end());
+  {
+    // Simulate receiving |packet2| from the network.
+    base::RunLoop run_loop;
+    EXPECT_CALL(*this, OnReceivedPacketCall())
+        .WillOnce(InvokeWithoutArgs(&run_loop, &base::RunLoop::Quit));
+    socket->OnReceivedPacket(packet2);
+    run_loop.Run();
+  }
+
+  // The packet is expected to be received.
+  EXPECT_TRUE(
+      std::equal(packet2.begin(), packet2.end(), received_packet_->begin()));
+
+  udp_transport_client_->StopReceiving();
+}
+
+TEST_F(UdpSocketClientTest, SendBeforeConnected) {
+  std::string data = "Test";
+  Packet packet(data.begin(), data.end());
+
+  // Request to send one packet.
+  base::MockCallback<base::RepeatingClosure> resume_send_cb;
+  {
+    EXPECT_CALL(resume_send_cb, Run()).Times(0);
+    EXPECT_FALSE(udp_transport_client_->SendPacket(
+        new base::RefCountedData<Packet>(packet), resume_send_cb.Get()));
+    scoped_task_environment_.RunUntilIdle();
+  }
+  {
+    // Expect the UDPSocket to be created when calling StartReceiving().
+    base::RunLoop run_loop;
+    EXPECT_CALL(*network_context_, OnUDPSocketCreated()).Times(1);
+    EXPECT_CALL(resume_send_cb, Run())
+        .WillOnce(InvokeWithoutArgs(&run_loop, &base::RunLoop::Quit));
+    udp_transport_client_->StartReceiving(base::BindRepeating(
+        &UdpSocketClientTest::OnReceivedPacket, base::Unretained(this)));
+    run_loop.Run();
+  }
+  udp_transport_client_->StopReceiving();
+}
+
+}  // namespace cast
+}  // namespace media

tools/traffic_annotation/summary/annotations.xml

@@ -33,6 +33,7 @@ Refer to README.md for content description and update process.
  <item id="cast_keep_alive_delegate" hash_code="134755844" type="0" content_hash_code="66118796" os_list="linux,windows" file_path="components/cast_channel/keep_alive_delegate.cc"/>
  <item id="cast_message_handler" hash_code="87558948" type="0" content_hash_code="49684899" os_list="linux,windows" file_path="components/cast_channel/cast_message_handler.cc"/>
  <item id="cast_socket" hash_code="115192205" type="0" content_hash_code="63056899" os_list="linux,windows" file_path="components/cast_channel/cast_socket.cc"/>
+ <item id="cast_udp_socket" hash_code="22573197" type="0" content_hash_code="75328301" os_list="linux,windows" file_path="media/cast/net/udp_socket_client.cc"/>
  <item id="cast_udp_transport" hash_code="5576536" type="0" content_hash_code="107643273" os_list="linux,windows" file_path="media/cast/net/udp_transport_impl.cc"/>
  <item id="certificate_verifier" hash_code="113553577" type="0" content_hash_code="62346354" os_list="linux,windows" file_path="net/cert_net/cert_net_fetcher_impl.cc"/>
  <item id="chrome_apps_socket_api" hash_code="8591273" type="0" content_hash_code="70868355" os_list="linux,windows" file_path="extensions/browser/api/socket/socket.cc"/>