[Nearby] Implement outgoing payload NearbyConnections interface

chrome/services/sharing/*
- Introduced AcceptConnection, RejectConnection for accepting/rejecting
  connection with remote endpoints, AcceptConnection also pass a
  PayloadListener
- Introduced SendPayload and CancelPayload for sedning/canceling
  outgoing payloads

chrome/browser/nearby_sharing/nearby_connections_manager*
- Browser side usage of the above APIs
  NearbyConnectionsManager uses AcceptConnection to accept connection,
  and implements PadloadListener.
  For outgoing payloads, OnPayloadTransferUpdate will pass the status to
  its listeners directly
- SendPayload and CancelPayload are straight forward

Bug: 1076008
Change-Id: If32e703e1d701373df1833719214039a990fafa8
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2339670Reviewed-by: Alex Gough <ajgo@chromium.org>
Reviewed-by: Richard Knoll <knollr@chromium.org>
Commit-Queue: Alex Chau <alexchau@chromium.org>
Cr-Commit-Position: refs/heads/master@{#797170}

chrome/browser/nearby_sharing/fake_nearby_connections_manager.cc

@@ -66,8 +66,7 @@ void FakeNearbyConnectionsManager::Disconnect(const std::string& endpoint_id) {
 
 void FakeNearbyConnectionsManager::Send(const std::string& endpoint_id,
                                         PayloadPtr payload,
-                                        PayloadStatusListener* listener,
-                                        ConnectionsCallback callback) {
+                                        PayloadStatusListener* listener) {
   DCHECK(!IsShutdown());
   // TODO(alexchau): Implement.
 }
@@ -86,8 +85,7 @@ FakeNearbyConnectionsManager::GetIncomingPayload(int64_t payload_id) {
   return nullptr;
 }
 
-void FakeNearbyConnectionsManager::Cancel(int64_t payload_id,
-                                          ConnectionsCallback callback) {
+void FakeNearbyConnectionsManager::Cancel(int64_t payload_id) {
   DCHECK(!IsShutdown());
   // TODO(alexchau): Implement.
 }

chrome/browser/nearby_sharing/fake_nearby_connections_manager.h

@@ -41,12 +41,11 @@ class FakeNearbyConnectionsManager
   void Disconnect(const std::string& endpoint_id) override;
   void Send(const std::string& endpoint_id,
             PayloadPtr payload,
-            PayloadStatusListener* listener,
-            ConnectionsCallback callback) override;
+            PayloadStatusListener* listener) override;
   void RegisterPayloadStatusListener(int64_t payload_id,
                                      PayloadStatusListener* listener) override;
   Payload* GetIncomingPayload(int64_t payload_id) override;
-  void Cancel(int64_t payload_id, ConnectionsCallback callback) override;
+  void Cancel(int64_t payload_id) override;
   void ClearIncomingPayloads() override;
   base::Optional<std::vector<uint8_t>> GetRawAuthenticationToken(
       const std::string& endpoint_id) override;

chrome/browser/nearby_sharing/mock_nearby_connections.h

@@ -19,6 +19,8 @@ using EndpointDiscoveryListener =
     location::nearby::connections::mojom::EndpointDiscoveryListener;
 using ConnectionLifecycleListener =
     location::nearby::connections::mojom::ConnectionLifecycleListener;
+using PayloadListener = location::nearby::connections::mojom::PayloadListener;
+using PayloadPtr = location::nearby::connections::mojom::PayloadPtr;
 
 class MockNearbyConnections : public NearbyConnectionsMojom {
  public:
@@ -55,6 +57,27 @@ class MockNearbyConnections : public NearbyConnectionsMojom {
               DisconnectFromEndpoint,
               (const std::string& endpoint_id, DisconnectFromEndpointCallback),
               (override));
+  MOCK_METHOD(void,
+              AcceptConnection,
+              (const std::string& endpoint_id,
+               mojo::PendingRemote<PayloadListener> listener,
+               AcceptConnectionCallback callback),
+              (override));
+  MOCK_METHOD(void,
+              RejectConnection,
+              (const std::string& endpoint_id,
+               RejectConnectionCallback callback),
+              (override));
+  MOCK_METHOD(void,
+              SendPayload,
+              (const std::vector<std::string>& endpoint_ids,
+               PayloadPtr payload,
+               SendPayloadCallback callback),
+              (override));
+  MOCK_METHOD(void,
+              CancelPayload,
+              (int64_t payload_id, CancelPayloadCallback callback),
+              (override));
 };
 
 #endif  // CHROME_BROWSER_NEARBY_SHARING_MOCK_NEARBY_CONNECTIONS_H_

chrome/browser/nearby_sharing/nearby_connection_impl.cc

@@ -43,7 +43,7 @@ void NearbyConnectionImpl::Write(std::vector<uint8_t> bytes) {
   payload->content =
       PayloadContent::NewBytes(BytesPayload::New(std::move(bytes)));
   nearby_connections_manager_->Send(endpoint_id_, std::move(payload),
-                                    /*listener=*/nullptr, base::DoNothing());
+                                    /*listener=*/nullptr);
 }
 
 void NearbyConnectionImpl::Close() {

chrome/browser/nearby_sharing/nearby_connections_manager.h

@@ -54,12 +54,13 @@ class NearbyConnectionsManager {
   // A callback for tracking the status of a payload (both incoming and
   // outgoing).
   class PayloadStatusListener {
-    using PayloadTransferUpdate =
-        location::nearby::connections::mojom::PayloadTransferUpdate;
+   public:
+    using PayloadTransferUpdatePtr =
+        location::nearby::connections::mojom::PayloadTransferUpdatePtr;
 
     virtual ~PayloadStatusListener() = default;
 
-    virtual void OnStatusUpdate(PayloadTransferUpdate update) = 0;
+    virtual void OnStatusUpdate(PayloadTransferUpdatePtr update) = 0;
   };
 
   virtual ~NearbyConnectionsManager() = default;
@@ -102,8 +103,7 @@ class NearbyConnectionsManager {
   // OnStatusUpdate.
   virtual void Send(const std::string& endpoint_id,
                     PayloadPtr payload,
-                    PayloadStatusListener* listener,
-                    ConnectionsCallback callback) = 0;
+                    PayloadStatusListener* listener) = 0;
 
   // Register a |listener| with |payload_id|. Caller is expected to ensure
   // |listener| remains valid until kSuccess/kFailure/kCancelled is invoked with
@@ -116,7 +116,7 @@ class NearbyConnectionsManager {
   virtual Payload* GetIncomingPayload(int64_t payload_id) = 0;
 
   // Cancels a Payload currently in-flight to or from remote endpoints.
-  virtual void Cancel(int64_t payload_id, ConnectionsCallback callback) = 0;
+  virtual void Cancel(int64_t payload_id) = 0;
 
   // Clears all incoming payloads.
   virtual void ClearIncomingPayloads() = 0;

chrome/browser/nearby_sharing/nearby_connections_manager_impl.cc

@@ -181,23 +181,21 @@ void NearbyConnectionsManagerImpl::Disconnect(const std::string& endpoint_id) {
 
 void NearbyConnectionsManagerImpl::Send(const std::string& endpoint_id,
                                         PayloadPtr payload,
-                                        PayloadStatusListener* listener,
-                                        ConnectionsCallback callback) {
-  if (!nearby_connections_) {
-    std::move(callback).Run(ConnectionsStatus::kError);
+                                        PayloadStatusListener* listener) {
+  if (!nearby_connections_)
     return;
-  }
 
-  // TOOD(crbug/1076008): Implement.
+  if (listener)
+    RegisterPayloadStatusListener(payload->id, listener);
+
+  nearby_connections_->SendPayload({endpoint_id}, std::move(payload),
+                                   base::DoNothing());
 }
 
 void NearbyConnectionsManagerImpl::RegisterPayloadStatusListener(
     int64_t payload_id,
     PayloadStatusListener* listener) {
-  if (!nearby_connections_)
-    return;
-
-  // TOOD(crbug/1076008): Implement.
+  payload_status_listeners_.insert_or_assign(payload_id, listener);
 }
 
 NearbyConnectionsManagerImpl::Payload*
@@ -206,14 +204,20 @@ NearbyConnectionsManagerImpl::GetIncomingPayload(int64_t payload_id) {
   return nullptr;
 }
 
-void NearbyConnectionsManagerImpl::Cancel(int64_t payload_id,
-                                          ConnectionsCallback callback) {
-  if (!nearby_connections_) {
-    std::move(callback).Run(ConnectionsStatus::kError);
+void NearbyConnectionsManagerImpl::Cancel(int64_t payload_id) {
+  if (!nearby_connections_)
     return;
-  }
 
-  // TOOD(crbug/1076008): Implement.
+  auto it = payload_status_listeners_.find(payload_id);
+  if (it != payload_status_listeners_.end()) {
+    it->second->OnStatusUpdate(
+        PayloadTransferUpdate::New(payload_id, PayloadStatus::kCanceled,
+                                   /*total_bytes=*/0,
+                                   /*bytes_transferred=*/0));
+    payload_status_listeners_.erase(it);
+  }
+  nearby_connections_->CancelPayload(payload_id, base::DoNothing());
+  NS_LOG(INFO) << "Cancelling payload: " << payload_id;
 }
 
 void NearbyConnectionsManagerImpl::ClearIncomingPayloads() {
@@ -300,9 +304,13 @@ void NearbyConnectionsManagerImpl::OnConnectionInitiated(
     ConnectionInfoPtr info) {
   auto result = connection_info_map_.emplace(endpoint_id, std::move(info));
   DCHECK(result.second);
-  // TOOD(crbug/1076008): Implemnet AcceptConnection.
-  // nearby_connections_->AcceptConnection(
-  //     endpoint_id, payload_listener_.BindNewPipeAndPassRemote());
+
+  mojo::PendingRemote<PayloadListener> payload_listener;
+  payload_listeners_.Add(this,
+                         payload_listener.InitWithNewPipeAndPassReceiver());
+
+  nearby_connections_->AcceptConnection(
+      endpoint_id, std::move(payload_listener), base::DoNothing());
 }
 
 void NearbyConnectionsManagerImpl::OnConnectionAccepted(
@@ -374,6 +382,29 @@ void NearbyConnectionsManagerImpl::OnBandwidthChanged(
   // TODO(crbug/1111458): Support TransferManager.
 }
 
+void NearbyConnectionsManagerImpl::OnPayloadTransferUpdate(
+    const std::string& endpoint_id,
+    PayloadTransferUpdatePtr update) {
+  // If this is a payload we've registered for, then forward its status to the
+  // PayloadStatusListener. We don't need to do anything more with the payload.
+  auto it = payload_status_listeners_.find(update->payload_id);
+  if (it != payload_status_listeners_.end()) {
+    PayloadStatusListener* listener = it->second;
+    switch (update->status) {
+      case PayloadStatus::kInProgress:
+        break;
+      case PayloadStatus::kSuccess:
+      case PayloadStatus::kCanceled:
+      case PayloadStatus::kFailure:
+        payload_status_listeners_.erase(it);
+        break;
+    }
+    listener->OnStatusUpdate(std::move(update));
+  }
+
+  // TOOD(crbug/1076008): Handle incoming  payload transfer.
+}
+
 bool NearbyConnectionsManagerImpl::BindNearbyConnections() {
   if (!nearby_connections_) {
     nearby_connections_ =

chrome/browser/nearby_sharing/nearby_connections_manager_impl.h

@@ -24,7 +24,8 @@ class NearbyConnectionsManagerImpl
     : public NearbyConnectionsManager,
       public NearbyProcessManager::Observer,
       public location::nearby::connections::mojom::EndpointDiscoveryListener,
-      public location::nearby::connections::mojom::ConnectionLifecycleListener {
+      public location::nearby::connections::mojom::ConnectionLifecycleListener,
+      public location::nearby::connections::mojom::PayloadListener {
  public:
   NearbyConnectionsManagerImpl(NearbyProcessManager* process_manager,
                                Profile* profile);
@@ -52,12 +53,11 @@ class NearbyConnectionsManagerImpl
   void Disconnect(const std::string& endpoint_id) override;
   void Send(const std::string& endpoint_id,
             PayloadPtr payload,
-            PayloadStatusListener* listener,
-            ConnectionsCallback callback) override;
+            PayloadStatusListener* listener) override;
   void RegisterPayloadStatusListener(int64_t payload_id,
                                      PayloadStatusListener* listener) override;
   Payload* GetIncomingPayload(int64_t payload_id) override;
-  void Cancel(int64_t payload_id, ConnectionsCallback callback) override;
+  void Cancel(int64_t payload_id) override;
   void ClearIncomingPayloads() override;
   base::Optional<std::vector<uint8_t>> GetRawAuthenticationToken(
       const std::string& endpoint_id) override;
@@ -66,17 +66,23 @@ class NearbyConnectionsManagerImpl
  private:
   using AdvertisingOptions =
       location::nearby::connections::mojom::AdvertisingOptions;
-  using MediumSelection = location::nearby::connections::mojom::MediumSelection;
-  using DiscoveryOptions =
-      location::nearby::connections::mojom::DiscoveryOptions;
-  using EndpointDiscoveryListener =
-      location::nearby::connections::mojom::EndpointDiscoveryListener;
+  using ConnectionInfoPtr =
+      location::nearby::connections::mojom::ConnectionInfoPtr;
   using ConnectionLifecycleListener =
       location::nearby::connections::mojom::ConnectionLifecycleListener;
   using DiscoveredEndpointInfoPtr =
       location::nearby::connections::mojom::DiscoveredEndpointInfoPtr;
-  using ConnectionInfoPtr =
-      location::nearby::connections::mojom::ConnectionInfoPtr;
+  using DiscoveryOptions =
+      location::nearby::connections::mojom::DiscoveryOptions;
+  using EndpointDiscoveryListener =
+      location::nearby::connections::mojom::EndpointDiscoveryListener;
+  using MediumSelection = location::nearby::connections::mojom::MediumSelection;
+  using PayloadListener = location::nearby::connections::mojom::PayloadListener;
+  using PayloadTransferUpdate =
+      location::nearby::connections::mojom::PayloadTransferUpdate;
+  using PayloadStatus = location::nearby::connections::mojom::PayloadStatus;
+  using PayloadTransferUpdatePtr =
+      location::nearby::connections::mojom::PayloadTransferUpdatePtr;
   using Status = location::nearby::connections::mojom::Status;
 
   FRIEND_TEST_ALL_PREFIXES(NearbyConnectionsManagerImplTest,
@@ -87,12 +93,12 @@ class NearbyConnectionsManagerImpl
   void OnNearbyProcessStarted() override;
   void OnNearbyProcessStopped() override;
 
-  // mojom::EndpointDiscoveryListener:
+  // EndpointDiscoveryListener:
   void OnEndpointFound(const std::string& endpoint_id,
                        DiscoveredEndpointInfoPtr info) override;
   void OnEndpointLost(const std::string& endpoint_id) override;
 
-  // mojom::ConnectionLifecycleListener:
+  // ConnectionLifecycleListener:
   void OnConnectionInitiated(const std::string& endpoint_id,
                              ConnectionInfoPtr info) override;
   void OnConnectionAccepted(const std::string& endpoint_id) override;
@@ -102,6 +108,10 @@ class NearbyConnectionsManagerImpl
   void OnBandwidthChanged(const std::string& endpoint_id,
                           int32_t quality) override;
 
+  // PayloadListener:
+  void OnPayloadTransferUpdate(const std::string& endpoint_id,
+                               PayloadTransferUpdatePtr update) override;
+
   void OnConnectionRequested(const std::string& endpoint_id,
                              NearbyConnectionCallback callback,
                              ConnectionsStatus status);
@@ -121,12 +131,15 @@ class NearbyConnectionsManagerImpl
   // A map of endpoint_id to NearbyConnection.
   base::flat_map<std::string, std::unique_ptr<NearbyConnectionImpl>>
       connections_;
+  // A map of payload_id to PayloadStatusListener*.
+  base::flat_map<int64_t, PayloadStatusListener*> payload_status_listeners_;
 
   ScopedObserver<NearbyProcessManager, NearbyProcessManager::Observer>
       nearby_process_observer_{this};
   mojo::Receiver<EndpointDiscoveryListener> endpoint_discovery_listener_{this};
   mojo::ReceiverSet<ConnectionLifecycleListener>
       connection_lifecycle_listeners_;
+  mojo::ReceiverSet<PayloadListener> payload_listeners_;
 
   location::nearby::connections::mojom::NearbyConnections* nearby_connections_ =
       nullptr;

chrome/browser/nearby_sharing/nearby_connections_manager_impl_unittest.cc

@@ -4,8 +4,10 @@
 
 #include "chrome/browser/nearby_sharing/nearby_connections_manager_impl.h"
 
+#include <algorithm>
 #include <memory>
 
+#include "base/files/file_util.h"
 #include "base/run_loop.h"
 #include "base/test/bind_test_util.h"
 #include "base/test/mock_callback.h"
@@ -34,6 +36,10 @@ const char kAuthenticationToken[] = "authentication_token";
 const char kRawAuthenticationToken[] = {0x00, 0x05, 0x04, 0x03, 0x02};
 const char kBytePayload[] = {0x08, 0x09, 0x06, 0x04, 0x0f};
 const char kBytePayload2[] = {0x0a, 0x0b, 0x0c, 0x0d, 0x0e};
+const int64_t kPayloadId = 689777;
+const uint64_t kTotalSize = 5201314;
+const uint64_t kBytesTransferred = 721831;
+const char kPayload[] = {0x0f, 0x0a, 0x0c, 0x0e};
 
 }  // namespace
 
@@ -42,6 +48,12 @@ using DiscoveredEndpointInfo =
     location::nearby::connections::mojom::DiscoveredEndpointInfo;
 using ConnectionInfo = location::nearby::connections::mojom::ConnectionInfo;
 using MediumSelection = location::nearby::connections::mojom::MediumSelection;
+using PayloadContent = location::nearby::connections::mojom::PayloadContent;
+using PayloadStatus = location::nearby::connections::mojom::PayloadStatus;
+using PayloadTransferUpdate =
+    location::nearby::connections::mojom::PayloadTransferUpdate;
+using Payload = location::nearby::connections::mojom::Payload;
+using FilePayload = location::nearby::connections::mojom::FilePayload;
 
 class MockDiscoveryListener
     : public NearbyConnectionsManager::DiscoveryListener {
@@ -68,6 +80,15 @@ class MockIncomingConnectionListener
               (override));
 };
 
+class MockPayloadStatusListener
+    : public NearbyConnectionsManager::PayloadStatusListener {
+ public:
+  MOCK_METHOD(void,
+              OnStatusUpdate,
+              (PayloadTransferUpdatePtr update),
+              (override));
+};
+
 class NearbyConnectionsManagerImplTest : public testing::Test {
  public:
   void SetUp() override {
@@ -128,7 +149,8 @@ class NearbyConnectionsManagerImplTest : public testing::Test {
   enum class ConnectionResponse { kAccepted, kRejceted, kDisconnected };
 
   NearbyConnection* Connect(
-      mojo::Remote<ConnectionLifecycleListener>& listener_remote,
+      mojo::Remote<ConnectionLifecycleListener>& connection_listener_remote,
+      mojo::Remote<PayloadListener>& payload_listener_remote,
       ConnectionResponse connection_response) {
     const std::vector<uint8_t> local_endpoint_info(std::begin(kEndpointInfo),
                                                    std::end(kEndpointInfo));
@@ -146,7 +168,7 @@ class NearbyConnectionsManagerImplTest : public testing::Test {
               EXPECT_EQ(local_endpoint_info, endpoint_info);
               EXPECT_EQ(kRemoteEndpointId, endpoint_id);
 
-              listener_remote.Bind(std::move(listener));
+              connection_listener_remote.Bind(std::move(listener));
               std::move(callback).Run(Status::kSuccess);
             });
 
@@ -160,22 +182,36 @@ class NearbyConnectionsManagerImplTest : public testing::Test {
           run_loop.Quit();
         }));
 
-    listener_remote->OnConnectionInitiated(
+    base::RunLoop accept_run_loop;
+    EXPECT_CALL(nearby_connections_, AcceptConnection)
+        .WillOnce(
+            [&](const std::string& endpoint_id,
+                mojo::PendingRemote<PayloadListener> listener,
+                NearbyConnectionsMojom::AcceptConnectionCallback callback) {
+              EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+
+              payload_listener_remote.Bind(std::move(listener));
+              std::move(callback).Run(Status::kSuccess);
+              accept_run_loop.Quit();
+            });
+
+    connection_listener_remote->OnConnectionInitiated(
         kRemoteEndpointId,
         ConnectionInfo::New(kAuthenticationToken, raw_authentication_token,
                             remote_endpoint_info,
                             /*is_incoming_connection=*/false));
+    accept_run_loop.Run();
 
     switch (connection_response) {
       case ConnectionResponse::kAccepted:
-        listener_remote->OnConnectionAccepted(kRemoteEndpointId);
+        connection_listener_remote->OnConnectionAccepted(kRemoteEndpointId);
         break;
       case ConnectionResponse::kRejceted:
-        listener_remote->OnConnectionRejected(kRemoteEndpointId,
-                                              Status::kConnectionRejected);
+        connection_listener_remote->OnConnectionRejected(
+            kRemoteEndpointId, Status::kConnectionRejected);
         break;
       case ConnectionResponse::kDisconnected:
-        listener_remote->OnDisconnected(kRemoteEndpointId);
+        connection_listener_remote->OnDisconnected(kRemoteEndpointId);
         break;
     }
     run_loop.Run();
@@ -183,6 +219,47 @@ class NearbyConnectionsManagerImplTest : public testing::Test {
     return nearby_connection;
   }
 
+  void SendPayload(
+      testing::NiceMock<MockPayloadStatusListener>& payload_listener) {
+    const std::vector<uint8_t> expected_payload(std::begin(kPayload),
+                                                std::end(kPayload));
+
+    base::FilePath path;
+    ASSERT_TRUE(base::CreateTemporaryFile(&path));
+    base::File file(path, base::File::Flags::FLAG_CREATE_ALWAYS |
+                              base::File::Flags::FLAG_READ |
+                              base::File::Flags::FLAG_WRITE);
+    EXPECT_TRUE(file.WriteAndCheck(
+        /*offset=*/0, base::make_span(expected_payload)));
+
+    base::RunLoop run_loop;
+    EXPECT_CALL(nearby_connections_, SendPayload)
+        .WillOnce([&](const std::vector<std::string>& endpoint_ids,
+                      PayloadPtr payload,
+                      NearbyConnectionsMojom::SendPayloadCallback callback) {
+          ASSERT_EQ(1u, endpoint_ids.size());
+          EXPECT_EQ(kRemoteEndpointId, endpoint_ids.front());
+          ASSERT_TRUE(payload);
+          ASSERT_EQ(PayloadContent::Tag::FILE, payload->content->which());
+
+          std::vector<uint8_t> payload_bytes(
+              payload->content->get_file()->file.GetLength());
+          EXPECT_TRUE(payload->content->get_file()->file.ReadAndCheck(
+              /*offset=*/0, base::make_span(payload_bytes)));
+          EXPECT_EQ(expected_payload, payload_bytes);
+
+          std::move(callback).Run(Status::kSuccess);
+          run_loop.Quit();
+        });
+
+    nearby_connections_manager_.Send(
+        kRemoteEndpointId,
+        Payload::New(kPayloadId, PayloadContent::NewFile(
+                                     FilePayload::New(std::move(file)))),
+        &payload_listener);
+    run_loop.Run();
+  }
+
   content::BrowserTaskEnvironment task_environment_;
   TestingProfile profile_;
   std::unique_ptr<net::test::MockNetworkChangeNotifier> network_notifier_ =
@@ -288,9 +365,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectRejected) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kRejceted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kRejceted);
   EXPECT_FALSE(nearby_connection);
   EXPECT_FALSE(
       nearby_connections_manager_.GetRawAuthenticationToken(kRemoteEndpointId));
@@ -303,9 +382,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectDisconnected) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kDisconnected);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kDisconnected);
   EXPECT_FALSE(nearby_connection);
   EXPECT_FALSE(
       nearby_connections_manager_.GetRawAuthenticationToken(kRemoteEndpointId));
@@ -321,9 +402,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectAccepted) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   EXPECT_TRUE(nearby_connection);
   EXPECT_EQ(
       raw_authentication_token,
@@ -340,9 +423,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectReadBeforeAppend) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
   // Read before message is appended should also succeed.
@@ -370,9 +455,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectReadAfterAppend) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
   // Read after message is appended should succeed.
@@ -408,13 +495,30 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectWrite) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
+  base::RunLoop run_loop;
+  EXPECT_CALL(nearby_connections_, SendPayload)
+      .WillOnce([&](const std::vector<std::string>& endpoint_ids,
+                    PayloadPtr payload,
+                    NearbyConnectionsMojom::SendPayloadCallback callback) {
+        ASSERT_EQ(1u, endpoint_ids.size());
+        EXPECT_EQ(kRemoteEndpointId, endpoint_ids.front());
+        ASSERT_TRUE(payload);
+        ASSERT_EQ(PayloadContent::Tag::BYTES, payload->content->which());
+        EXPECT_EQ(byte_payload, payload->content->get_bytes()->bytes);
+
+        std::move(callback).Run(Status::kSuccess);
+        run_loop.Quit();
+      });
+
   nearby_connection->Write(byte_payload);
-  // TOOD(crbug/1076008): Veriy that nearby_connections_.SendPayload is called.
+  run_loop.Run();
 }
 
 TEST_F(NearbyConnectionsManagerImplTest, ConnectClosed) {
@@ -424,9 +528,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosed) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
   // Close should invoke disconnection callback and read callback.
@@ -462,9 +568,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosedByRemote) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
   // Remote closing should invoke disconnection callback and read callback.
@@ -478,7 +586,7 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosedByRemote) {
         read_run_loop.Quit();
       }));
 
-  listener_remote->OnDisconnected(kRemoteEndpointId);
+  connection_listener_remote->OnDisconnected(kRemoteEndpointId);
   close_run_loop.Run();
   read_run_loop.Run();
 
@@ -493,9 +601,11 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosedByClient) {
   StartDiscovery(discovery_listener_remote, discovery_listener);
 
   // RequestConnection will succeed.
-  mojo::Remote<ConnectionLifecycleListener> listener_remote;
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
   NearbyConnection* nearby_connection =
-      Connect(listener_remote, ConnectionResponse::kAccepted);
+      Connect(connection_listener_remote, payload_listener_remote,
+              ConnectionResponse::kAccepted);
   ASSERT_TRUE(nearby_connection);
 
   // Remote closing should invoke disconnection callback and read callback.
@@ -524,6 +634,127 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosedByClient) {
       nearby_connections_manager_.GetRawAuthenticationToken(kRemoteEndpointId));
 }
 
+TEST_F(NearbyConnectionsManagerImplTest, ConnectPayloadStatusListener) {
+  // StartDiscovery will succeed.
+  mojo::Remote<EndpointDiscoveryListener> discovery_listener_remote;
+  testing::NiceMock<MockDiscoveryListener> discovery_listener;
+  StartDiscovery(discovery_listener_remote, discovery_listener);
+
+  // RequestConnection will succeed.
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
+  Connect(connection_listener_remote, payload_listener_remote,
+          ConnectionResponse::kAccepted);
+
+  testing::NiceMock<MockPayloadStatusListener> payload_listener;
+  nearby_connections_manager_.RegisterPayloadStatusListener(kPayloadId,
+                                                            &payload_listener);
+
+  auto expected_update = PayloadTransferUpdate::New(
+      kPayloadId, PayloadStatus::kInProgress, kTotalSize, kBytesTransferred);
+  base::RunLoop payload_run_loop;
+  EXPECT_CALL(payload_listener, OnStatusUpdate(testing::_))
+      .WillOnce(
+          [&](MockPayloadStatusListener::PayloadTransferUpdatePtr update) {
+            EXPECT_EQ(expected_update, update);
+            payload_run_loop.Quit();
+          });
+
+  payload_listener_remote->OnPayloadTransferUpdate(kRemoteEndpointId,
+                                                   expected_update.Clone());
+  payload_run_loop.Run();
+
+  // After success status.
+  base::RunLoop payload_run_loop_2;
+  EXPECT_CALL(payload_listener, OnStatusUpdate(testing::_))
+      .WillOnce([&payload_run_loop_2]() { payload_run_loop_2.Quit(); });
+
+  payload_listener_remote->OnPayloadTransferUpdate(
+      kRemoteEndpointId,
+      PayloadTransferUpdate::New(kPayloadId, PayloadStatus::kSuccess,
+                                 kTotalSize, /*bytes_transferred=*/kTotalSize));
+  payload_run_loop_2.Run();
+
+  // PayloadStatusListener will be unregistered and won't receive further
+  // updates.
+  payload_listener_remote->OnPayloadTransferUpdate(
+      kRemoteEndpointId,
+      PayloadTransferUpdate::New(kPayloadId, PayloadStatus::kSuccess,
+                                 kTotalSize, /*bytes_transferred=*/kTotalSize));
+  EXPECT_CALL(payload_listener, OnStatusUpdate(testing::_)).Times(0);
+}
+
+TEST_F(NearbyConnectionsManagerImplTest, ConnectSendPayload) {
+  // StartDiscovery will succeed.
+  mojo::Remote<EndpointDiscoveryListener> discovery_listener_remote;
+  testing::NiceMock<MockDiscoveryListener> discovery_listener;
+  StartDiscovery(discovery_listener_remote, discovery_listener);
+
+  // RequestConnection will succeed.
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
+  Connect(connection_listener_remote, payload_listener_remote,
+          ConnectionResponse::kAccepted);
+
+  testing::NiceMock<MockPayloadStatusListener> payload_listener;
+  SendPayload(payload_listener);
+
+  auto expected_update = PayloadTransferUpdate::New(
+      kPayloadId, PayloadStatus::kInProgress, kTotalSize, kBytesTransferred);
+  base::RunLoop payload_run_loop;
+  EXPECT_CALL(payload_listener, OnStatusUpdate(testing::_))
+      .WillOnce(
+          [&](MockPayloadStatusListener::PayloadTransferUpdatePtr update) {
+            EXPECT_EQ(expected_update, update);
+            payload_run_loop.Quit();
+          });
+
+  payload_listener_remote->OnPayloadTransferUpdate(kRemoteEndpointId,
+                                                   expected_update.Clone());
+  payload_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsManagerImplTest, ConnectCancelPayload) {
+  // StartDiscovery will succeed.
+  mojo::Remote<EndpointDiscoveryListener> discovery_listener_remote;
+  testing::NiceMock<MockDiscoveryListener> discovery_listener;
+  StartDiscovery(discovery_listener_remote, discovery_listener);
+
+  // RequestConnection will succeed.
+  mojo::Remote<ConnectionLifecycleListener> connection_listener_remote;
+  mojo::Remote<PayloadListener> payload_listener_remote;
+  Connect(connection_listener_remote, payload_listener_remote,
+          ConnectionResponse::kAccepted);
+
+  testing::NiceMock<MockPayloadStatusListener> payload_listener;
+  SendPayload(payload_listener);
+
+  base::RunLoop cancel_run_loop;
+  EXPECT_CALL(nearby_connections_, CancelPayload)
+      .WillOnce([&](int64_t payload_id,
+                    NearbyConnectionsMojom::CancelPayloadCallback callback) {
+        EXPECT_EQ(kPayloadId, payload_id);
+
+        std::move(callback).Run(Status::kSuccess);
+        cancel_run_loop.Quit();
+      });
+
+  base::RunLoop payload_run_loop;
+  EXPECT_CALL(payload_listener, OnStatusUpdate(testing::_))
+      .WillOnce(
+          [&](MockPayloadStatusListener::PayloadTransferUpdatePtr update) {
+            EXPECT_EQ(kPayloadId, update->payload_id);
+            EXPECT_EQ(PayloadStatus::kCanceled, update->status);
+            EXPECT_EQ(0u, update->total_bytes);
+            EXPECT_EQ(0u, update->bytes_transferred);
+            payload_run_loop.Quit();
+          });
+
+  nearby_connections_manager_.Cancel(kPayloadId);
+  payload_run_loop.Run();
+  cancel_run_loop.Run();
+}
+
 TEST_F(NearbyConnectionsManagerImplTest, StartAdvertising) {
   mojo::Remote<ConnectionLifecycleListener> listener_remote;
   testing::NiceMock<MockIncomingConnectionListener>

chrome/services/sharing/nearby/nearby_connections.cc

@@ -8,6 +8,8 @@
 #include "chrome/browser/nearby_sharing/logging/logging.h"
 #include "chrome/services/sharing/nearby/nearby_connections_conversions.h"
 #include "chrome/services/sharing/public/mojom/nearby_connections_types.mojom.h"
+#include "third_party/nearby/src/cpp/platform_v2/public/file.h"
+#include "third_party/nearby/src/cpp/platform_v2/public/single_thread_executor.h"
 
 namespace location {
 namespace nearby {
@@ -276,6 +278,77 @@ void NearbyConnections::DisconnectFromEndpoint(
                                 ResultCallbackFromMojom(std::move(callback)));
 }
 
+void NearbyConnections::AcceptConnection(
+    const std::string& endpoint_id,
+    mojo::PendingRemote<mojom::PayloadListener> listener,
+    AcceptConnectionCallback callback) {
+  mojo::SharedRemote<mojom::PayloadListener> remote(std::move(listener));
+  PayloadListener payload_listener = {
+      .payload_progress_cb = [remote](const std::string& endpoint_id,
+                                      const PayloadProgressInfo& info) {
+        if (!remote)
+          return;
+
+        DCHECK_GE(info.total_bytes, 0);
+        DCHECK_GE(info.bytes_transferred, 0);
+        remote->OnPayloadTransferUpdate(
+            endpoint_id, mojom::PayloadTransferUpdate::New(
+                             info.payload_id, PayloadStatusToMojom(info.status),
+                             info.total_bytes, info.bytes_transferred));
+      }};
+
+  core_->AcceptConnection(endpoint_id, std::move(payload_listener),
+                          ResultCallbackFromMojom(std::move(callback)));
+}
+
+void NearbyConnections::RejectConnection(const std::string& endpoint_id,
+                                         RejectConnectionCallback callback) {
+  core_->RejectConnection(endpoint_id,
+                          ResultCallbackFromMojom(std::move(callback)));
+}
+
+void NearbyConnections::SendPayload(
+    const std::vector<std::string>& endpoint_ids,
+    mojom::PayloadPtr payload,
+    SendPayloadCallback callback) {
+  std::unique_ptr<Payload> core_payload;
+  switch (payload->content->which()) {
+    case mojom::PayloadContent::Tag::BYTES:
+      core_payload = std::make_unique<Payload>(
+          payload->id,
+          ByteArrayFromMojom(payload->content->get_bytes()->bytes));
+      break;
+    case mojom::PayloadContent::Tag::FILE:
+      int64_t file_size = payload->content->get_file()->file.GetLength();
+      outgoing_file_map_.insert_or_assign(
+          payload->id, std::move(payload->content->get_file()->file));
+      core_payload = std::make_unique<Payload>(
+          payload->id, InputFile(payload->id, file_size));
+      break;
+  }
+
+  core_->SendPayload(absl::MakeSpan(endpoint_ids), std::move(*core_payload),
+                     ResultCallbackFromMojom(std::move(callback)));
+}
+
+void NearbyConnections::CancelPayload(int64_t payload_id,
+                                      CancelPayloadCallback callback) {
+  core_->CancelPayload(payload_id,
+                       ResultCallbackFromMojom(std::move(callback)));
+}
+
+base::File NearbyConnections::ExtractFileForPayload(int64_t payload_id) {
+  auto file_it = outgoing_file_map_.find(payload_id);
+  if (file_it == outgoing_file_map_.end())
+    return base::File();
+
+  base::File file = std::move(file_it->second);
+  outgoing_file_map_.erase(file_it);
+  return file;
+
+  // TOOD(crbug/1076008): Handle incoming file payload.
+}
+
 }  // namespace connections
 }  // namespace nearby
 }  // namespace location

chrome/services/sharing/nearby/nearby_connections.h

@@ -5,10 +5,15 @@
 #ifndef CHROME_SERVICES_SHARING_NEARBY_NEARBY_CONNECTIONS_H_
 #define CHROME_SERVICES_SHARING_NEARBY_NEARBY_CONNECTIONS_H_
 
+#include <stdint.h>
 #include <memory>
 
 #include "base/callback_forward.h"
+#include "base/containers/flat_map.h"
+#include "base/files/file.h"
 #include "base/memory/weak_ptr.h"
+#include "base/optional.h"
+#include "base/task/post_task.h"
 #include "chrome/services/sharing/public/mojom/nearby_connections.mojom.h"
 #include "chrome/services/sharing/public/mojom/webrtc_signaling_messenger.mojom.h"
 #include "device/bluetooth/public/mojom/adapter.mojom.h"
@@ -78,6 +83,19 @@ class NearbyConnections : public mojom::NearbyConnections {
       RequestConnectionCallback callback) override;
   void DisconnectFromEndpoint(const std::string& endpoint_id,
                               DisconnectFromEndpointCallback callback) override;
+  void AcceptConnection(const std::string& endpoint_id,
+                        mojo::PendingRemote<mojom::PayloadListener> listener,
+                        AcceptConnectionCallback callback) override;
+  void RejectConnection(const std::string& endpoint_id,
+                        RejectConnectionCallback callback) override;
+  void SendPayload(const std::vector<std::string>& endpoint_ids,
+                   mojom::PayloadPtr payload,
+                   SendPayloadCallback callback) override;
+  void CancelPayload(int64_t payload_id,
+                     CancelPayloadCallback callback) override;
+
+  // Return the file associated with |payload_id|.
+  base::File ExtractFileForPayload(int64_t payload_id);
 
  private:
   void OnDisconnect();
@@ -95,6 +113,7 @@ class NearbyConnections : public mojom::NearbyConnections {
       webrtc_signaling_messenger_;
 
   std::unique_ptr<Core> core_;
+  base::flat_map<int64_t, base::File> outgoing_file_map_;
 
   base::WeakPtrFactory<NearbyConnections> weak_ptr_factory_{this};
 };

chrome/services/sharing/nearby/nearby_connections_conversions.cc

@@ -81,6 +81,23 @@ std::vector<uint8_t> ByteArrayToMojom(const ByteArray& byte_array) {
                               byte_array.data() + byte_array.size());
 }
 
+ByteArray ByteArrayFromMojom(const std::vector<uint8_t>& byte_array) {
+  return ByteArray(std::string(byte_array.begin(), byte_array.end()));
+}
+
+mojom::PayloadStatus PayloadStatusToMojom(PayloadProgressInfo::Status status) {
+  switch (status) {
+    case PayloadProgressInfo::Status::kSuccess:
+      return mojom::PayloadStatus::kSuccess;
+    case PayloadProgressInfo::Status::kFailure:
+      return mojom::PayloadStatus::kFailure;
+    case PayloadProgressInfo::Status::kInProgress:
+      return mojom::PayloadStatus::kInProgress;
+    case PayloadProgressInfo::Status::kCanceled:
+      return mojom::PayloadStatus::kCanceled;
+  }
+}
+
 }  // namespace connections
 }  // namespace nearby
 }  // namespace location

chrome/services/sharing/nearby/nearby_connections_conversions.h

@@ -28,6 +28,10 @@ ResultCallback ResultCallbackFromMojom(StatusCallback callback);
 
 std::vector<uint8_t> ByteArrayToMojom(const ByteArray& byte_array);
 
+ByteArray ByteArrayFromMojom(const std::vector<uint8_t>& byte_array);
+
+mojom::PayloadStatus PayloadStatusToMojom(PayloadProgressInfo::Status status);
+
 }  // namespace connections
 }  // namespace nearby
 }  // namespace location

chrome/services/sharing/nearby/nearby_connections_unittest.cc

@@ -5,15 +5,18 @@
 #include "chrome/services/sharing/nearby/nearby_connections.h"
 
 #include <stdint.h>
+#include <algorithm>
 #include <memory>
 #include <utility>
 
 #include "base/bind.h"
 #include "base/callback.h"
+#include "base/files/file_util.h"
 #include "base/memory/ptr_util.h"
 #include "base/run_loop.h"
 #include "base/test/bind_test_util.h"
 #include "base/test/task_environment.h"
+#include "chrome/services/sharing/nearby/nearby_connections_conversions.h"
 #include "chrome/services/sharing/nearby/test_support/fake_adapter.h"
 #include "chrome/services/sharing/nearby/test_support/mock_webrtc_dependencies.h"
 #include "chrome/services/sharing/public/mojom/nearby_decoder.mojom.h"
@@ -37,6 +40,8 @@ const char kRemoteEndpointInfo[] = {0x0d, 0x07, 0x06, 0x08, 0x09};
 const char kAuthenticationToken[] = "authentication_token";
 const char kRawAuthenticationToken[] = {0x00, 0x05, 0x04, 0x03, 0x02};
 const int32_t kQuality = 5201314;
+const int64_t kPayloadId = 612721831;
+const char kPayload[] = {0x0f, 0x0a, 0x0c, 0x0e};
 
 mojom::AdvertisingOptionsPtr CreateAdvertisingOptions() {
   auto allowed_mediums = mojom::MediumSelection::New(/*bluetooth=*/true,
@@ -108,6 +113,20 @@ class FakeConnectionLifecycleListener
       bandwidth_changed_cb = base::DoNothing();
 };
 
+class FakePayloadListener : public mojom::PayloadListener {
+ public:
+  void OnPayloadTransferUpdate(
+      const std::string& endpoint_id,
+      mojom::PayloadTransferUpdatePtr update) override {
+    payload_progress_cb.Run(endpoint_id, std::move(update));
+  }
+
+  mojo::Receiver<mojom::PayloadListener> receiver{this};
+  base::RepeatingCallback<void(const std::string&,
+                               mojom::PayloadTransferUpdatePtr)>
+      payload_progress_cb = base::DoNothing();
+};
+
 class NearbyConnectionsTest : public testing::Test {
  public:
   NearbyConnectionsTest() {
@@ -132,6 +151,101 @@ class NearbyConnectionsTest : public testing::Test {
 
   void OnDisconnect() { disconnect_run_loop_.Quit(); }
 
+  ClientProxy* StartDiscovery(
+      FakeEndpointDiscoveryListener& fake_discovery_listener) {
+    ClientProxy* client_proxy;
+    EXPECT_CALL(*service_controller_ptr_, StartDiscovery)
+        .WillOnce([&client_proxy](ClientProxy* client,
+                                  const std::string& service_id,
+                                  const ConnectionOptions& options,
+                                  const DiscoveryListener& listener) {
+          client_proxy = client;
+          EXPECT_EQ(kServiceId, service_id);
+          EXPECT_EQ(Strategy::kP2pPointToPoint, options.strategy);
+          client->StartedDiscovery(service_id, options.strategy, listener,
+                                   /*mediums=*/{});
+          return Status{Status::kAlreadyDiscovering};
+        });
+
+    base::RunLoop start_discovery_run_loop;
+    nearby_connections_->StartDiscovery(
+        kServiceId,
+        mojom::DiscoveryOptions::New(mojom::Strategy::kP2pPointToPoint),
+        fake_discovery_listener.receiver.BindNewPipeAndPassRemote(),
+        base::BindLambdaForTesting([&](mojom::Status status) {
+          EXPECT_EQ(mojom::Status::kAlreadyDiscovering, status);
+          start_discovery_run_loop.Quit();
+        }));
+    start_discovery_run_loop.Run();
+
+    return client_proxy;
+  }
+
+  ClientProxy* RequestConnection(
+      FakeConnectionLifecycleListener& fake_connection_life_cycle_listener) {
+    ClientProxy* client_proxy;
+    EXPECT_CALL(*service_controller_ptr_, RequestConnection)
+        .WillOnce([&client_proxy](ClientProxy* client,
+                                  const std::string& endpoint_id,
+                                  const ConnectionRequestInfo& info) {
+          client_proxy = client;
+          EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+          EXPECT_EQ(
+              std::string(std::begin(kEndpointInfo), std::end(kEndpointInfo)),
+              info.name);
+          client_proxy->OnConnectionInitiated(
+              endpoint_id,
+              {.authentication_token = kAuthenticationToken,
+               .raw_authentication_token = ByteArray(
+                   kRawAuthenticationToken, sizeof(kRawAuthenticationToken)),
+               .endpoint_info =
+                   ByteArray(kRemoteEndpointInfo, sizeof(kRemoteEndpointInfo)),
+               .is_incoming_connection = false},
+              info.listener);
+          return Status{Status::kSuccess};
+        });
+
+    base::RunLoop request_connection_run_loop;
+    nearby_connections_->RequestConnection(
+        std::vector<uint8_t>(std::begin(kEndpointInfo),
+                             std::end(kEndpointInfo)),
+        kRemoteEndpointId,
+        fake_connection_life_cycle_listener.receiver.BindNewPipeAndPassRemote(),
+        base::BindLambdaForTesting([&](mojom::Status status) {
+          EXPECT_EQ(mojom::Status::kSuccess, status);
+          request_connection_run_loop.Quit();
+        }));
+    request_connection_run_loop.Run();
+
+    return client_proxy;
+  }
+
+  ClientProxy* AcceptConnection(FakePayloadListener& fake_payload_listener) {
+    ClientProxy* client_proxy;
+    EXPECT_CALL(*service_controller_ptr_, AcceptConnection)
+        .WillOnce([&client_proxy](ClientProxy* client,
+                                  const std::string& endpoint_id,
+                                  const PayloadListener& listener) {
+          client_proxy = client;
+          EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+          client_proxy->LocalEndpointAcceptedConnection(endpoint_id, listener);
+          client_proxy->OnConnectionAccepted(endpoint_id);
+          return Status{Status::kSuccess};
+        });
+
+    base::RunLoop accept_connection_run_loop;
+    nearby_connections_->AcceptConnection(
+        kRemoteEndpointId,
+        fake_payload_listener.receiver.BindNewPipeAndPassRemote(),
+        base::BindLambdaForTesting([&](mojom::Status status) {
+          EXPECT_EQ(mojom::Status::kSuccess, status);
+          accept_connection_run_loop.Quit();
+        }));
+    accept_connection_run_loop.Run();
+
+    return client_proxy;
+  }
+
  protected:
   base::test::TaskEnvironment task_environment_;
   mojo::Remote<mojom::NearbyConnections> remote_;
@@ -175,33 +289,9 @@ TEST_F(NearbyConnectionsTest, WebRtcSignalingMessengerDisconnect) {
   disconnect_run_loop_.Run();
 }
 
-TEST_F(NearbyConnectionsTest, StartStopDiscovery) {
-  ClientProxy* client_proxy;
-  EXPECT_CALL(*service_controller_ptr_, StartDiscovery)
-      .WillOnce([&client_proxy](ClientProxy* client,
-                                const std::string& service_id,
-                                const ConnectionOptions& options,
-                                const DiscoveryListener& listener) {
-        client_proxy = client;
-        EXPECT_EQ(kServiceId, service_id);
-        EXPECT_EQ(Strategy::kP2pPointToPoint, options.strategy);
-
-        client->StartedDiscovery(service_id, options.strategy, listener,
-                                 /*mediums=*/{});
-        return Status{Status::kAlreadyDiscovering};
-      });
-
-  base::RunLoop start_discovery_run_loop;
+TEST_F(NearbyConnectionsTest, StartDiscovery) {
   FakeEndpointDiscoveryListener fake_discovery_listener;
-  nearby_connections_->StartDiscovery(
-      kServiceId,
-      mojom::DiscoveryOptions::New(mojom::Strategy::kP2pPointToPoint),
-      fake_discovery_listener.receiver.BindNewPipeAndPassRemote(),
-      base::BindLambdaForTesting([&](mojom::Status status) {
-        EXPECT_EQ(mojom::Status::kAlreadyDiscovering, status);
-        start_discovery_run_loop.Quit();
-      }));
-  start_discovery_run_loop.Run();
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
 
   base::RunLoop endpoint_found_run_loop;
   fake_discovery_listener.endpoint_found_cb =
@@ -229,6 +319,11 @@ TEST_F(NearbyConnectionsTest, StartStopDiscovery) {
       });
   client_proxy->OnEndpointLost(kServiceId, kRemoteEndpointId);
   endpoint_lost_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsTest, StopDiscovery) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  StartDiscovery(fake_discovery_listener);
 
   EXPECT_CALL(*service_controller_ptr_, StopDiscovery(testing::_)).Times(1);
 
@@ -244,61 +339,16 @@ TEST_F(NearbyConnectionsTest, StartStopDiscovery) {
   EXPECT_CALL(*service_controller_ptr_, StopDiscovery(testing::_)).Times(1);
 }
 
-TEST_F(NearbyConnectionsTest, RequestConnection) {
-  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
-
-  EXPECT_CALL(*service_controller_ptr_, StartDiscovery)
-      .WillOnce([](ClientProxy* client, const std::string& service_id,
-                   const ConnectionOptions& options,
-                   const DiscoveryListener& listener) {
-        client->StartedDiscovery(service_id, options.strategy, listener,
-                                 /*mediums=*/{});
-        client->OnEndpointFound(kServiceId, kRemoteEndpointId,
+TEST_F(NearbyConnectionsTest, RequestConnectionInitiated) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
                                 std::string(std::begin(kRemoteEndpointInfo),
                                             std::end(kRemoteEndpointInfo)),
                                 /*mediums=*/{});
-        return Status{Status::kSuccess};
-      });
-
-  base::RunLoop start_discovery_run_loop;
-  FakeEndpointDiscoveryListener fake_discovery_listener;
-  nearby_connections_->StartDiscovery(
-      kServiceId,
-      mojom::DiscoveryOptions::New(mojom::Strategy::kP2pPointToPoint),
-      fake_discovery_listener.receiver.BindNewPipeAndPassRemote(),
-      base::BindLambdaForTesting([&](mojom::Status status) {
-        EXPECT_EQ(mojom::Status::kSuccess, status);
-        start_discovery_run_loop.Quit();
-      }));
-  start_discovery_run_loop.Run();
-
-  ClientProxy* client_proxy;
-  ConnectionListener connections_listener;
-  EXPECT_CALL(*service_controller_ptr_, RequestConnection)
-      .WillOnce([&client_proxy, &connections_listener](
-                    ClientProxy* client, const std::string& endpoint_id,
-                    const ConnectionRequestInfo& info) {
-        client_proxy = client;
-        connections_listener = info.listener;
-        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
-        EXPECT_EQ(
-            std::string(std::begin(kEndpointInfo), std::end(kEndpointInfo)),
-            info.name);
-        return Status{Status::kSuccess};
-      });
-
-  base::RunLoop request_connection_run_loop;
-  nearby_connections_->RequestConnection(
-      std::vector<uint8_t>(std::begin(kEndpointInfo), std::end(kEndpointInfo)),
-      kRemoteEndpointId,
-      fake_connection_life_cycle_listener.receiver.BindNewPipeAndPassRemote(),
-      base::BindLambdaForTesting([&](mojom::Status status) {
-        EXPECT_EQ(mojom::Status::kSuccess, status);
-        request_connection_run_loop.Quit();
-      }));
-  request_connection_run_loop.Run();
 
   base::RunLoop initiated_run_loop;
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
   fake_connection_life_cycle_listener.initiated_cb = base::BindLambdaForTesting(
       [&](const std::string& endpoint_id, mojom::ConnectionInfoPtr info) {
         EXPECT_EQ(kRemoteEndpointId, endpoint_id);
@@ -312,16 +362,44 @@ TEST_F(NearbyConnectionsTest, RequestConnection) {
         EXPECT_FALSE(info->is_incoming_connection);
         initiated_run_loop.Quit();
       });
-  client_proxy->OnConnectionInitiated(
-      kRemoteEndpointId,
-      {.authentication_token = kAuthenticationToken,
-       .raw_authentication_token =
-           ByteArray(kRawAuthenticationToken, sizeof(kRawAuthenticationToken)),
-       .endpoint_info =
-           ByteArray(kRemoteEndpointInfo, sizeof(kRemoteEndpointInfo)),
-       .is_incoming_connection = false},
-      connections_listener);
+
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
   initiated_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsTest, RequestConnectionAccept) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  base::RunLoop accepted_run_loop;
+  fake_connection_life_cycle_listener.accepted_cb =
+      base::BindLambdaForTesting([&](const std::string& endpoint_id) {
+        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+        accepted_run_loop.Quit();
+      });
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
+  accepted_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsTest, RequestConnectionOnRejected) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
 
   base::RunLoop rejected_run_loop;
   fake_connection_life_cycle_listener.rejected_cb = base::BindLambdaForTesting(
@@ -330,31 +408,25 @@ TEST_F(NearbyConnectionsTest, RequestConnection) {
         EXPECT_EQ(mojom::Status::kConnectionRejected, status);
         rejected_run_loop.Quit();
       });
+
   client_proxy->OnConnectionRejected(kRemoteEndpointId,
                                      {Status::kConnectionRejected});
   rejected_run_loop.Run();
+}
 
-  // Initiate connection again to test accepted flow.
-  base::RunLoop initiated_run_loop_2;
-  fake_connection_life_cycle_listener.initiated_cb = base::BindLambdaForTesting(
-      [&](const std::string& endpoint_id, mojom::ConnectionInfoPtr info) {
-        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
-        EXPECT_FALSE(info->is_incoming_connection);
-        initiated_run_loop_2.Quit();
-      });
-  client_proxy->OnConnectionInitiated(kRemoteEndpointId,
-                                      {.is_incoming_connection = false},
-                                      connections_listener);
-  initiated_run_loop_2.Run();
+TEST_F(NearbyConnectionsTest, RequestConnectionOnBandwidthUpgrade) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
 
-  base::RunLoop accepted_run_loop;
-  fake_connection_life_cycle_listener.accepted_cb =
-      base::BindLambdaForTesting([&](const std::string& endpoint_id) {
-        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
-        accepted_run_loop.Quit();
-      });
-  client_proxy->OnConnectionAccepted(kRemoteEndpointId);
-  accepted_run_loop.Run();
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
 
   base::RunLoop bandwidth_changed_run_loop;
   fake_connection_life_cycle_listener.bandwidth_changed_cb =
@@ -364,8 +436,24 @@ TEST_F(NearbyConnectionsTest, RequestConnection) {
             EXPECT_EQ(kQuality, quality);
             bandwidth_changed_run_loop.Quit();
           });
+
   client_proxy->OnBandwidthChanged(kRemoteEndpointId, kQuality);
   bandwidth_changed_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsTest, RequestConnectionOnDisconnected) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
 
   base::RunLoop disconnected_run_loop;
   fake_connection_life_cycle_listener.disconnected_cb =
@@ -373,37 +461,39 @@ TEST_F(NearbyConnectionsTest, RequestConnection) {
         EXPECT_EQ(kRemoteEndpointId, endpoint_id);
         disconnected_run_loop.Quit();
       });
+
   client_proxy->OnDisconnected(kRemoteEndpointId, /*notify=*/true);
   disconnected_run_loop.Run();
+}
 
-  // Initiate and accept connection again to test DisconnectFromEndpoint.
-  base::RunLoop initiated_run_loop_3;
-  fake_connection_life_cycle_listener.initiated_cb = base::BindLambdaForTesting(
-      [&](const std::string& endpoint_id, mojom::ConnectionInfoPtr info) {
-        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
-        EXPECT_FALSE(info->is_incoming_connection);
-        initiated_run_loop_3.Quit();
-      });
-  client_proxy->OnConnectionInitiated(kRemoteEndpointId,
-                                      {.is_incoming_connection = false},
-                                      connections_listener);
-  initiated_run_loop_3.Run();
+TEST_F(NearbyConnectionsTest, RequestConnectionDisconnect) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
 
-  base::RunLoop accepted_run_loop_2;
-  fake_connection_life_cycle_listener.accepted_cb =
-      base::BindLambdaForTesting([&](const std::string& endpoint_id) {
-        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
-        accepted_run_loop_2.Quit();
-      });
-  client_proxy->OnConnectionAccepted(kRemoteEndpointId);
-  accepted_run_loop_2.Run();
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
 
   EXPECT_CALL(*service_controller_ptr_, DisconnectFromEndpoint)
       .WillOnce([](ClientProxy* client, const std::string& endpoint_id) {
         EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+        client->OnDisconnected(endpoint_id, /*notify=*/true);
         return Status{Status::kSuccess};
       });
 
+  base::RunLoop disconnected_run_loop;
+  fake_connection_life_cycle_listener.disconnected_cb =
+      base::BindLambdaForTesting([&](const std::string& endpoint_id) {
+        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+        disconnected_run_loop.Quit();
+      });
+
   base::RunLoop disconnect_from_endpoint_run_loop;
   nearby_connections_->DisconnectFromEndpoint(
       kRemoteEndpointId, base::BindLambdaForTesting([&](mojom::Status status) {
@@ -411,9 +501,128 @@ TEST_F(NearbyConnectionsTest, RequestConnection) {
         disconnect_from_endpoint_run_loop.Quit();
       }));
   disconnect_from_endpoint_run_loop.Run();
+  disconnected_run_loop.Run();
+}
+
+TEST_F(NearbyConnectionsTest, OnPayloadTransferUpdate) {
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
+
+  base::RunLoop payload_progress_run_loop;
+  fake_payload_listener.payload_progress_cb =
+      base::BindLambdaForTesting([&](const std::string& endpoint_id,
+                                     mojom::PayloadTransferUpdatePtr info) {
+        EXPECT_EQ(kRemoteEndpointId, endpoint_id);
+        payload_progress_run_loop.Quit();
+      });
+
+  client_proxy->OnPayloadProgress(kRemoteEndpointId, {});
+  payload_progress_run_loop.Run();
+}
+
+// TOOD(crbug/1076008): Re-enable test after upprev NearbyConnections.
+TEST_F(NearbyConnectionsTest, DISABLED_SendBytesPayload) {
+  const std::vector<uint8_t> expected_payload(std::begin(kPayload),
+                                              std::end(kPayload));
+
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
+
+  EXPECT_CALL(*service_controller_ptr_, SendPayload)
+      .WillOnce([&](ClientProxy* client,
+                    const std::vector<std::string>& endpoint_ids,
+                    Payload payload) {
+        ASSERT_EQ(1u, endpoint_ids.size());
+        EXPECT_EQ(kRemoteEndpointId, endpoint_ids.front());
+        EXPECT_EQ(Payload::Type::kBytes, payload.GetType());
+        std::string payload_bytes(payload.AsBytes());
+        EXPECT_EQ(expected_payload, ByteArrayToMojom(payload.AsBytes()));
+      });
+
+  base::RunLoop send_payload_run_loop;
+  nearby_connections_->SendPayload(
+      {kRemoteEndpointId},
+      mojom::Payload::New(kPayloadId,
+                          mojom::PayloadContent::NewBytes(
+                              mojom::BytesPayload::New(expected_payload))),
+      base::BindLambdaForTesting([&](mojom::Status status) {
+        EXPECT_EQ(mojom::Status::kSuccess, status);
+        send_payload_run_loop.Quit();
+      }));
+  send_payload_run_loop.Run();
+}
 
-  // DisconnectFromEndpoint is also called when Core is destroyed.
-  EXPECT_CALL(*service_controller_ptr_, DisconnectFromEndpoint).Times(1);
+// TOOD(crbug/1076008): Re-enable test after upprev NearbyConnections.
+TEST_F(NearbyConnectionsTest, DISABLED_SendFilePayload) {
+  const std::vector<uint8_t> expected_payload(std::begin(kPayload),
+                                              std::end(kPayload));
+
+  FakeEndpointDiscoveryListener fake_discovery_listener;
+  ClientProxy* client_proxy = StartDiscovery(fake_discovery_listener);
+  client_proxy->OnEndpointFound(kServiceId, kRemoteEndpointId,
+                                std::string(std::begin(kRemoteEndpointInfo),
+                                            std::end(kRemoteEndpointInfo)),
+                                /*mediums=*/{});
+
+  FakeConnectionLifecycleListener fake_connection_life_cycle_listener;
+  client_proxy = RequestConnection(fake_connection_life_cycle_listener);
+
+  FakePayloadListener fake_payload_listener;
+  client_proxy = AcceptConnection(fake_payload_listener);
+
+  EXPECT_CALL(*service_controller_ptr_, SendPayload)
+      .WillOnce([&](ClientProxy* client,
+                    const std::vector<std::string>& endpoint_ids,
+                    Payload payload) {
+        ASSERT_EQ(1u, endpoint_ids.size());
+        EXPECT_EQ(kRemoteEndpointId, endpoint_ids.front());
+        EXPECT_EQ(Payload::Type::kFile, payload.GetType());
+        InputFile* file = payload.AsFile();
+        ASSERT_TRUE(file);
+        ExceptionOr<ByteArray> bytes = file->Read(file->GetTotalSize());
+        ASSERT_TRUE(bytes.ok());
+        EXPECT_EQ(expected_payload, ByteArrayToMojom(bytes.result()));
+      });
+
+  base::FilePath path;
+  ASSERT_TRUE(base::CreateTemporaryFile(&path));
+  base::File file(path, base::File::Flags::FLAG_CREATE_ALWAYS |
+                            base::File::Flags::FLAG_READ |
+                            base::File::Flags::FLAG_WRITE);
+  EXPECT_TRUE(file.WriteAndCheck(
+      /*offset=*/0, base::make_span(expected_payload)));
+
+  base::RunLoop send_payload_run_loop;
+  nearby_connections_->SendPayload(
+      {kRemoteEndpointId},
+      mojom::Payload::New(kPayloadId,
+                          mojom::PayloadContent::NewFile(
+                              mojom::FilePayload::New(std::move(file)))),
+      base::BindLambdaForTesting([&](mojom::Status status) {
+        EXPECT_EQ(mojom::Status::kSuccess, status);
+        send_payload_run_loop.Quit();
+      }));
+  send_payload_run_loop.Run();
 }
 
 TEST_F(NearbyConnectionsTest, StartAdvertising) {

chrome/services/sharing/nearby/platform_v2/BUILD.gn

@@ -19,6 +19,8 @@ source_set("platform_v2") {
     "count_down_latch.cc",
     "count_down_latch.h",
     "crypto.cc",
+    "input_file.cc",
+    "input_file.h",
     "log_message.cc",
     "log_message.h",
     "mutex.cc",

chrome/services/sharing/nearby/platform_v2/input_file.cc

+// 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 "chrome/services/sharing/nearby/platform_v2/input_file.h"
+
+#include "base/logging.h"
+
+namespace location {
+namespace nearby {
+namespace chrome {
+
+InputFile::InputFile(base::File file) : file_(std::move(file)) {
+  DCHECK(file_.IsValid());
+  seek_succeeded_ = file_.Seek(base::File::FROM_BEGIN, 0) == 0;
+}
+
+InputFile::~InputFile() = default;
+
+std::string InputFile::GetFilePath() const {
+  // File path is not supported.
+  return std::string();
+}
+
+std::int64_t InputFile::GetTotalSize() const {
+  return file_.GetLength();
+}
+
+ExceptionOr<ByteArray> InputFile::Read(std::int64_t size) {
+  if (!seek_succeeded_)
+    return Exception::kFailed;
+
+  ByteArray bytes(size);
+  int bytes_read = file_.ReadAtCurrentPos(bytes.data(), bytes.size());
+  if (bytes_read != size)
+    return Exception::kFailed;
+
+  return ExceptionOr<ByteArray>(std::move(bytes));
+}
+
+Exception InputFile::Close() {
+  file_.Close();
+  return {Exception::kSuccess};
+}
+
+}  // namespace chrome
+}  // namespace nearby
+}  // namespace location

chrome/services/sharing/nearby/platform_v2/input_file.h

+// 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 CHROME_SERVICES_SHARING_NEARBY_PLATFORM_V2_INPUT_FILE_H_
+#define CHROME_SERVICES_SHARING_NEARBY_PLATFORM_V2_INPUT_FILE_H_
+
+#include "base/files/file.h"
+#include "third_party/nearby/src/cpp/platform_v2/api/input_file.h"
+
+namespace location {
+namespace nearby {
+namespace chrome {
+
+// Concrete InputFile implementation.
+class InputFile : public api::InputFile {
+ public:
+  explicit InputFile(base::File file);
+  ~InputFile() override;
+
+  InputFile(const InputFile&) = delete;
+  InputFile& operator=(const InputFile&) = delete;
+
+  // api::InputFile:
+  std::string GetFilePath() const override;
+  std::int64_t GetTotalSize() const override;
+  ExceptionOr<ByteArray> Read(std::int64_t size) override;
+  Exception Close() override;
+
+ private:
+  // File::GetLength is not const but api::InputFile::GetTotalSize is const.
+  mutable base::File file_;
+  bool seek_succeeded_;
+};
+
+}  // namespace chrome
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CHROME_SERVICES_SHARING_NEARBY_PLATFORM_V2_INPUT_FILE_H_

chrome/services/sharing/nearby/platform_v2/platform.cc

@@ -13,6 +13,7 @@
 #include "chrome/services/sharing/nearby/platform_v2/bluetooth_classic_medium.h"
 #include "chrome/services/sharing/nearby/platform_v2/condition_variable.h"
 #include "chrome/services/sharing/nearby/platform_v2/count_down_latch.h"
+#include "chrome/services/sharing/nearby/platform_v2/input_file.h"
 #include "chrome/services/sharing/nearby/platform_v2/log_message.h"
 #include "chrome/services/sharing/nearby/platform_v2/mutex.h"
 #include "chrome/services/sharing/nearby/platform_v2/recursive_mutex.h"
@@ -104,8 +105,12 @@ std::unique_ptr<AtomicBoolean> ImplementationPlatform::CreateAtomicBoolean(
 std::unique_ptr<InputFile> ImplementationPlatform::CreateInputFile(
     std::int64_t payload_id,
     std::int64_t total_size) {
-  return std::make_unique<shared::InputFile>(GetPayloadPath(payload_id),
-                                             total_size);
+  auto& connections = connections::NearbyConnections::GetInstance();
+  auto file = connections.ExtractFileForPayload(payload_id);
+  if (!file.IsValid())
+    return nullptr;
+
+  return std::make_unique<chrome::InputFile>(std::move(file));
 }
 
 std::unique_ptr<OutputFile> ImplementationPlatform::CreateOutputFile(

chrome/services/sharing/public/mojom/nearby_connections.mojom

@@ -91,6 +91,20 @@ interface ConnectionLifecycleListener {
   OnBandwidthChanged(string endpoint_id, int32 quality);
 };
 
+// Listener for payload status. Methods in this interface are called from
+// utiltiiy process, and is used by the browser process to listen for payload
+// status associated with remote endpoints.
+interface PayloadListener {
+  // Called with progress information about an active Payload transfer, either
+  // incoming or outgoing.
+  //
+  // endpoint_id - The identifier for the remote endpoint that is sending or
+  //               receiving this payload.
+  // update      - The PayloadTransferUpdate structure describing the status of
+  //               the transfer.
+  OnPayloadTransferUpdate(string endpoint_id, PayloadTransferUpdate update);
+};
+
 // Main interface to control the NearbyConnections library. Implemented in a
 // sandboxed process. This interface is used by the browser process to connect
 // to remote devices and send / receive raw data packets. Parsing of those
@@ -179,6 +193,37 @@ interface NearbyConnections {
                    pending_remote<ConnectionLifecycleListener> listener)
       => (Status status);
 
+  // Accepts a connection to a remote endpoint. This method must be called
+  // before Payloads can be exchanged with the remote endpoint.
+  //
+  // endpoint_id - The identifier for the remote endpoint. Should match the
+  //               value provided in a call to
+  //               ConnectionsLifecycleListener::OnConnectionInitiated().
+  // listener    - An interface for payloads exchanged with the remote endpoint.
+  // Possible return values include:
+  //   Status::kSuccess if the connection request was accepted.
+  //   Status::kAlreadyConnectedToEndpoint if the app already as a connection
+  //       to the specified endpoint.
+  //   Status::kOutOfOrderApiCall if
+  //       ConnectionsLifecycleListener::OnConnectionInitiated() has not been
+  //       called for the |endpoint_id|.
+  AcceptConnection(string endpoint_id, pending_remote<PayloadListener> listener)
+      => (Status status);
+
+  // Rejects a connection to a remote endpoint.
+  //
+  // endpoint_id - The identifier for the remote endpoint. Should match the
+  //               value provided in a call to
+  //               ConnectionsLifecycleListener::OnConnectionInitiated().
+  // Possible return values include:
+  //   Status::kSuccess if the connection request was rejected.
+  //   Status::kAlreadyConnectedToEndpoint if the app already has a connection
+  //       to the specified endpoint.
+  //   Status::kOutOfOrderApiCall if
+  //       ConnectionsLifecycleListener::OnConnectionInitiated() has not been
+  //       called for the |endpoint_id|.
+  RejectConnection(string endpoint_id) => (Status status);
+
   // Disconnects from a remote endpoint. Payloads can no longer be sent
   // to or received from the endpoint after this method is called.
   //
@@ -186,6 +231,33 @@ interface NearbyConnections {
   // Possible return values include:
   //   Status::kSuccess disconnected successfully.
   DisconnectFromEndpoint(string endpoint_id) => (Status status);
+
+  // Sends a Payload to a list of remote endpoints. Payloads can only be sent to
+  // remote endpoints once a notice of connection acceptance has been delivered
+  // via ConnectionsLifecycleListener::OnConnectionResult().
+  //
+  // endpoint_ids - Array of remote endpoint identifiers for to which the
+  //                payload should be sent.
+  // payload      - The Payload to be sent.
+  // Possible return values include:
+  //   Status::kOutOfOrderApiCall if the device has not first performed
+  //       advertisement or discovery (to set the Strategy).
+  //   Status::kEndpointUnknown if there's no active (or pending) connection to
+  //       the remote endpoint.
+  //   Status::kSuccess if none of the above errors occurred. Note that this
+  //       indicates that Nearby Connections will attempt to send the Payload,
+  //       but not that the send has successfully completed yet. Errors might
+  //       still occur during transmission (and at different times for different
+  //       endpoints), and will be delivered via
+  //       PayloadListener::OnPayloadTransferUpdate().
+  SendPayload(array<string> endpoint_ids, Payload payload) => (Status status);
+
+  // Cancels a Payload currently in-flight to or from remote endpoint(s).
+  //
+  // payload_id - The identifier for the Payload to be canceled.
+  // Possible return values include:
+  //   Status::kSuccess if the payload got canceled.
+  CancelPayload(int64 payload_id) => (Status status);
 };
 
 // Provide all the dependencies that NearbyConnections library requires.