diff --git a/sharing/BUILD b/sharing/BUILD
index 66773d5636..ce0fb7a2c8 100644
--- a/sharing/BUILD
+++ b/sharing/BUILD
@@ -167,7 +167,6 @@ cc_library(
     hdrs = ["paired_key_verification_runner.h"],
     deps = [
         ":incoming_frame_reader",
-        ":types",
         "//internal/platform:types",
         "//proto:sharing_enums_cc_proto",
         "//sharing/certificates",
@@ -175,6 +174,7 @@ cc_library(
         "//sharing/proto:enums_cc_proto",
         "//sharing/proto:share_cc_proto",
         "//sharing/proto:wire_format_cc_proto",
+        "@com_google_absl//absl/functional:any_invocable",
         "@com_google_absl//absl/time",
     ],
 )
@@ -229,6 +229,7 @@ cc_library(
         "//sharing/proto:wire_format_cc_proto",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/functional:any_invocable",
+        "@com_google_absl//absl/functional:bind_front",
         "@com_google_absl//absl/strings:str_format",
         "@com_google_absl//absl/strings:string_view",
         "@com_google_absl//absl/time",
diff --git a/sharing/fake_nearby_connections_manager.cc b/sharing/fake_nearby_connections_manager.cc
index cc63c9df11..46abd75874 100644
--- a/sharing/fake_nearby_connections_manager.cc
+++ b/sharing/fake_nearby_connections_manager.cc
@@ -36,8 +36,7 @@
 #include "sharing/nearby_connections_types.h"
 #include "sharing/proto/enums.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
 using DataUsage = ::nearby::sharing::proto::DataUsage;
 
@@ -119,7 +118,7 @@ void FakeNearbyConnectionsManager::Connect(
     absl::MutexLock lock(&endpoints_mutex_);
     connection_endpoint_infos_.emplace(endpoint_id, std::move(endpoint_info));
   }
-  std::move(callback)(connection_, Status::kUnknown);
+  std::move(callback)(endpoint_id, connection_, Status::kUnknown);
 }
 
 void FakeNearbyConnectionsManager::AcceptConnection(
@@ -335,5 +334,4 @@ void FakeNearbyConnectionsManager::AddUnknownFilePathsToDeleteForTesting(
 
 std::string FakeNearbyConnectionsManager::Dump() const { return ""; }
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/incoming_frames_reader_test.cc b/sharing/incoming_frames_reader_test.cc
index c7d7754519..0e68342932 100644
--- a/sharing/incoming_frames_reader_test.cc
+++ b/sharing/incoming_frames_reader_test.cc
@@ -26,13 +26,11 @@
 #include "internal/test/fake_clock.h"
 #include "internal/test/fake_device_info.h"
 #include "internal/test/fake_task_runner.h"
-#include "sharing/fake_nearby_connections_manager.h"
 #include "sharing/internal/public/logging.h"
 #include "sharing/nearby_connection_impl.h"
 #include "sharing/proto/wire_format.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 
 using ::nearby::sharing::service::proto::ConnectionResponseFrame;
@@ -103,7 +101,7 @@ class IncomingFramesReaderTest : public testing::Test {
  public:
   IncomingFramesReaderTest() {
     nearby_connection_ = std::make_unique<NearbyConnectionImpl>(
-        fake_device_info_, &fake_nearby_connections_manager_, "endpoint_id");
+        fake_device_info_, "endpoint_id");
   }
   ~IncomingFramesReaderTest() override = default;
 
@@ -113,7 +111,7 @@ class IncomingFramesReaderTest : public testing::Test {
   }
 
   NearbyConnectionImpl& connection() {
-    NL_CHECK(nearby_connection_);
+    CHECK(nearby_connection_);
     return *nearby_connection_;
   }
 
@@ -136,7 +134,6 @@ class IncomingFramesReaderTest : public testing::Test {
   FakeClock fake_clock_;
   FakeTaskRunner fake_task_runner_ {&fake_clock_, 1};
   FakeDeviceInfo fake_device_info_;
-  FakeNearbyConnectionsManager fake_nearby_connections_manager_;
   std::unique_ptr<NearbyConnectionImpl> nearby_connection_;
   std::shared_ptr<IncomingFramesReader> frames_reader_ = nullptr;
 };
@@ -369,5 +366,4 @@ TEST_F(IncomingFramesReaderTest, SkipInvalidFrame) {
 }
 
 }  // namespace
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/incoming_share_session_test.cc b/sharing/incoming_share_session_test.cc
index 95a0da2135..88ac55e114 100644
--- a/sharing/incoming_share_session_test.cc
+++ b/sharing/incoming_share_session_test.cc
@@ -116,7 +116,7 @@ std::unique_ptr<Payload> CreateWifiCredentialsPayload(
 class IncomingShareSessionTest : public ::testing::Test {
  protected:
   IncomingShareSessionTest()
-      : connection_(device_info_, &connections_manager_, kEndpointId),
+      : connection_(device_info_, kEndpointId),
         session_(&clock_, task_runner_, &connections_manager_,
                  analytics_recorder_, std::string(kEndpointId), share_target_,
                  transfer_metadata_callback_.AsStdFunction()) {
diff --git a/sharing/nearby_connection.h b/sharing/nearby_connection.h
index 0988e00b5e..8212f4b8e9 100644
--- a/sharing/nearby_connection.h
+++ b/sharing/nearby_connection.h
@@ -20,8 +20,7 @@
 #include <optional>
 #include <vector>
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
 // A socket-like wrapper around Nearby Connections that allows for asynchronous
 // reads and writes.
@@ -38,21 +37,12 @@ class NearbyConnection {
       std::function<void(std::optional<std::vector<uint8_t>> bytes)>
           callback) = 0;
 
-  // Writes an outgoing stream of bytes to the remote device asynchronously.
-  // Must not be used on an already closed connection.
-  virtual void Write(std::vector<uint8_t> bytes) = 0;
-
-  // Closes the socket and disconnects from the remote device. This object will
-  // be invalidated after `listener` in SetDisconnectionListener is invoked.
-  virtual void Close() = 0;
-
   // Listens to the socket being closed. Invoke `listener` when the socket is
   // closed. This object will be invalidated after `listener` is invoked.
   // Previously set listener will be replaced by `listener`.
   virtual void SetDisconnectionListener(std::function<void()> listener) = 0;
 };
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
 
 #endif  // THIRD_PARTY_NEARBY_SHARING_NEARBY_CONNECTION_H_
diff --git a/sharing/nearby_connection_impl.cc b/sharing/nearby_connection_impl.cc
index b498930fa8..fc92b11e56 100644
--- a/sharing/nearby_connection_impl.cc
+++ b/sharing/nearby_connection_impl.cc
@@ -16,7 +16,6 @@
 
 #include <cstdint>
 #include <functional>
-#include <memory>
 #include <optional>
 #include <queue>
 #include <string>
@@ -27,21 +26,15 @@
 #include "absl/synchronization/mutex.h"
 #include "internal/platform/device_info.h"
 #include "sharing/internal/public/logging.h"
-#include "sharing/nearby_connections_manager.h"
-#include "sharing/nearby_connections_types.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
-NearbyConnectionImpl::NearbyConnectionImpl(
-    nearby::DeviceInfo& device_info,
-    NearbyConnectionsManager* nearby_connections_manager,
-    absl::string_view endpoint_id)
+NearbyConnectionImpl::NearbyConnectionImpl(nearby::DeviceInfo& device_info,
+                                           absl::string_view endpoint_id)
     : device_info_(device_info),
-      nearby_connections_manager_(nearby_connections_manager),
       endpoint_id_(endpoint_id) {
   if (!device_info_.PreventSleep()) {
-    NL_LOG(WARNING) << __func__ << ":Failed to prevent device sleep.";
+    LOG(WARNING) << __func__ << ":Failed to prevent device sleep.";
   }
 }
 
@@ -51,7 +44,7 @@ NearbyConnectionImpl::~NearbyConnectionImpl() {
   {
     absl::MutexLock lock(&mutex_);
     if (!device_info_.AllowSleep()) {
-      NL_LOG(ERROR) << __func__ << ":Failed to allow device sleep.";
+      LOG(ERROR) << __func__ << ":Failed to allow device sleep.";
     }
     disconnect_listener = std::move(disconnect_listener_);
     read_callback = std::move(read_callback_);
@@ -81,19 +74,6 @@ void NearbyConnectionImpl::Read(
   std::move(callback)(std::move(bytes));
 }
 
-void NearbyConnectionImpl::Write(std::vector<uint8_t> bytes) {
-  nearby_connections_manager_->Send(
-      endpoint_id_, std::make_unique<Payload>(bytes),
-      /*listener=*/
-      std::weak_ptr<NearbyConnectionsManager::PayloadStatusListener>());
-}
-
-void NearbyConnectionImpl::Close() {
-  // As [this] therefore endpoint_id_ will be destroyed in Disconnect, make a
-  // copy of [endpoint_id] as the parameter is a const ref.
-  nearby_connections_manager_->Disconnect(endpoint_id_);
-}
-
 void NearbyConnectionImpl::SetDisconnectionListener(
     std::function<void()> listener) {
   absl::MutexLock lock(&mutex_);
@@ -116,5 +96,4 @@ void NearbyConnectionImpl::WriteMessage(std::vector<uint8_t> bytes) {
   }
 }
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/nearby_connection_impl.h b/sharing/nearby_connection_impl.h
index b5eb0d4159..370b508897 100644
--- a/sharing/nearby_connection_impl.h
+++ b/sharing/nearby_connection_impl.h
@@ -28,15 +28,13 @@
 #include "internal/platform/device_info.h"
 #include "sharing/nearby_connection.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
 class NearbyConnectionsManager;
 
 class NearbyConnectionImpl : public NearbyConnection {
  public:
   NearbyConnectionImpl(nearby::DeviceInfo& device_info,
-                       NearbyConnectionsManager* nearby_connections_manager,
                        absl::string_view endpoint_id);
   ~NearbyConnectionImpl() override;
 
@@ -44,8 +42,6 @@ class NearbyConnectionImpl : public NearbyConnection {
   void Read(
       std::function<void(std::optional<std::vector<uint8_t>> bytes)> callback)
       ABSL_LOCKS_EXCLUDED(mutex_) override;
-  void Write(std::vector<uint8_t> bytes) override;
-  void Close() override;
   void SetDisconnectionListener(std::function<void()> listener)
       ABSL_LOCKS_EXCLUDED(mutex_) override;
 
@@ -54,7 +50,6 @@ class NearbyConnectionImpl : public NearbyConnection {
 
  private:
   nearby::DeviceInfo& device_info_;
-  NearbyConnectionsManager* const nearby_connections_manager_;
   const std::string endpoint_id_;
 
   absl::Mutex mutex_;
@@ -67,7 +62,6 @@ class NearbyConnectionImpl : public NearbyConnection {
   std::queue<std::vector<uint8_t>> reads_ ABSL_GUARDED_BY(mutex_);
 };
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
 
 #endif  // THIRD_PARTY_NEARBY_SHARING_NEARBY_CONNECTION_IMPL_H_
diff --git a/sharing/nearby_connection_impl_test.cc b/sharing/nearby_connection_impl_test.cc
index f9bca8d4cd..69b409947a 100644
--- a/sharing/nearby_connection_impl_test.cc
+++ b/sharing/nearby_connection_impl_test.cc
@@ -23,23 +23,19 @@
 #include "internal/test/fake_clock.h"
 #include "internal/test/fake_device_info.h"
 #include "internal/test/fake_task_runner.h"
-#include "sharing/fake_nearby_connections_manager.h"
 #include "sharing/incoming_frames_reader.h"
 #include "sharing/proto/wire_format.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 
 TEST(NearbyConnectionImpl, DestructorBeforeReaderDestructor) {
-  FakeNearbyConnectionsManager connection_manager;
   FakeClock fake_clock;
   FakeTaskRunner fake_task_runner(&fake_clock, 1);
   FakeDeviceInfo device_info;
   bool called = false;
 
-  auto connection = std::make_unique<NearbyConnectionImpl>(
-      device_info, &connection_manager, "test");
+  auto connection = std::make_unique<NearbyConnectionImpl>(device_info, "test");
   auto frames_reader = std::make_shared<IncomingFramesReader>(fake_task_runner,
                                                               connection.get());
 
@@ -56,14 +52,12 @@ TEST(NearbyConnectionImpl, DestructorBeforeReaderDestructor) {
 }
 
 TEST(NearbyConnectionImpl, DestructorAfterReaderDestructor) {
-  FakeNearbyConnectionsManager connection_manager;
   FakeClock fake_clock;
   FakeTaskRunner fake_task_runner(&fake_clock, 1);
   FakeDeviceInfo device_info;
   std::optional<nearby::sharing::service::proto::V1Frame> frame_result;
 
-  auto connection = std::make_unique<NearbyConnectionImpl>(
-      device_info, &connection_manager, "test");
+  auto connection = std::make_unique<NearbyConnectionImpl>(device_info, "test");
   auto frames_reader = std::make_shared<IncomingFramesReader>(fake_task_runner,
                                                               connection.get());
 
@@ -82,5 +76,4 @@ TEST(NearbyConnectionImpl, DestructorAfterReaderDestructor) {
 }
 
 }  // namespace
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/nearby_connections_manager.h b/sharing/nearby_connections_manager.h
index 9adb954f96..e47ea42e47 100644
--- a/sharing/nearby_connections_manager.h
+++ b/sharing/nearby_connections_manager.h
@@ -28,22 +28,19 @@
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
 #include "sharing/common/nearby_share_enums.h"
+#include "sharing/nearby_connection.h"
 #include "sharing/nearby_connections_types.h"
 #include "sharing/proto/enums.pb.h"
 
-namespace nearby {
-namespace sharing {
-
-class NearbyConnection;
-class PayloadTransferUpdatePtr;
+namespace nearby::sharing {
 
 using ConnectionsStatus = nearby::sharing::Status;
 
 class NearbyConnectionsManager {
  public:
   using ConnectionsCallback = std::function<void(ConnectionsStatus status)>;
-  using NearbyConnectionCallback =
-      std::function<void(NearbyConnection*, Status status)>;
+  using NearbyConnectionCallback = std::function<void(
+      absl::string_view endpoint_id, NearbyConnection*, Status status)>;
 
   // A callback for handling incoming connections while advertising.
   class IncomingConnectionListener {
@@ -166,7 +163,6 @@ class NearbyConnectionsManager {
   virtual std::string Dump() const = 0;
 };
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
 
 #endif  // THIRD_PARTY_NEARBY_SHARING_NEARBY_CONNECTIONS_MANAGER_H_
diff --git a/sharing/nearby_connections_manager_impl.cc b/sharing/nearby_connections_manager_impl.cc
index 6b63240506..e5a7a19d60 100644
--- a/sharing/nearby_connections_manager_impl.cc
+++ b/sharing/nearby_connections_manager_impl.cc
@@ -50,8 +50,7 @@
 #include "sharing/nearby_connections_types.h"
 #include "sharing/transfer_manager.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 using ::nearby::sharing::proto::DataUsage;
 
@@ -334,7 +333,7 @@ void NearbyConnectionsManagerImpl::Connect(
     NearbyConnectionCallback callback) {
   MutexLock lock(&mutex_);
   if (!nearby_connections_service_) {
-    callback(nullptr, Status::kError);
+    callback(endpoint_id, nullptr, Status::kError);
     return;
   }
 
@@ -690,9 +689,9 @@ void NearbyConnectionsManagerImpl::OnConnectionAccepted(
       return;
     }
 
-    auto result = connections_.emplace(std::string(endpoint_id),
-                                       std::make_unique<NearbyConnectionImpl>(
-                                           device_info_, this, endpoint_id));
+    auto result = connections_.emplace(
+        std::string(endpoint_id),
+        std::make_unique<NearbyConnectionImpl>(device_info_, endpoint_id));
     DCHECK(result.second);
     incoming_connection_listener_->OnIncomingConnection(
         endpoint_id, it->second.endpoint_info, result.first->second.get());
@@ -704,10 +703,11 @@ void NearbyConnectionsManagerImpl::OnConnectionAccepted(
     }
 
     auto result = connections_.emplace(
-        endpoint_id, std::make_unique<NearbyConnectionImpl>(device_info_, this,
-                                                            endpoint_id));
+        endpoint_id,
+        std::make_unique<NearbyConnectionImpl>(device_info_, endpoint_id));
     DCHECK(result.second);
-    std::move(it->second)(result.first->second.get(), Status::kSuccess);
+    std::move(it->second)(endpoint_id, result.first->second.get(),
+                          Status::kSuccess);
     pending_outgoing_connections_.erase(it);
     connect_timeout_timers_.erase(endpoint_id);
   }
@@ -720,7 +720,7 @@ void NearbyConnectionsManagerImpl::OnConnectionRejected(
 
   auto it = pending_outgoing_connections_.find(endpoint_id);
   if (it != pending_outgoing_connections_.end()) {
-    std::move(it->second)(nullptr, status);
+    std::move(it->second)(endpoint_id, nullptr, status);
     pending_outgoing_connections_.erase(it);
     connect_timeout_timers_.erase(endpoint_id);
   }
@@ -744,7 +744,7 @@ void NearbyConnectionsManagerImpl::OnDisconnected(
 
   auto it = pending_outgoing_connections_.find(endpoint_id);
   if (it != pending_outgoing_connections_.end()) {
-    std::move(it->second)(nullptr, connection_layer_status);
+    std::move(it->second)(endpoint_id, nullptr, connection_layer_status);
     pending_outgoing_connections_.erase(it);
     connect_timeout_timers_.erase(endpoint_id);
   }
@@ -932,7 +932,8 @@ void NearbyConnectionsManagerImpl::Reset() {
   transfer_managers_.clear();
 
   for (auto& entry : pending_outgoing_connections_)
-    std::move(entry.second)(/*connection=*/nullptr, Status::kReset);
+    std::move(entry.second)(entry.first, /*connection=*/nullptr,
+                            Status::kReset);
 
   pending_outgoing_connections_.clear();
 }
@@ -1003,5 +1004,4 @@ std::string NearbyConnectionsManagerImpl::Dump() const {
   return nearby_connections_service_->Dump();
 }
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/nearby_connections_manager_impl_test.cc b/sharing/nearby_connections_manager_impl_test.cc
index c214fd8d0f..9a45c35b0c 100644
--- a/sharing/nearby_connections_manager_impl_test.cc
+++ b/sharing/nearby_connections_manager_impl_test.cc
@@ -53,8 +53,7 @@
 #include "sharing/proto/enums.pb.h"
 #include "sharing/transfer_manager.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 
 using ::nearby::sharing::proto::DataUsage;
@@ -279,7 +278,8 @@ class NearbyConnectionsManagerImplTest : public testing::Test {
         local_endpoint_info, kRemoteEndpointId,
         /*bluetooth_mac_address=*/std::nullopt, DataUsage::OFFLINE_DATA_USAGE,
         TransportType::kHighQuality,
-        [&](NearbyConnection* connection, Status status) {
+        [&](absl::string_view endpoint_id, NearbyConnection* connection,
+            Status status) {
           nearby_connection = connection;
           notification.Notify();
         });
@@ -927,7 +927,10 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectWrite) {
         notification.Notify();
       });
 
-  nearby_connection->Write(byte_payload);
+  nearby_connections_manager_->Send(
+      kRemoteEndpointId, std::make_unique<Payload>(byte_payload),
+      /*listener=*/
+      std::weak_ptr<NearbyConnectionsManager::PayloadStatusListener>());
   Sync();
   EXPECT_TRUE(
       notification.WaitForNotificationWithTimeout(kSynchronizationTimeOut));
@@ -969,7 +972,7 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectClosed) {
         disconnect_notification.Notify();
       });
 
-  nearby_connection->Close();
+  nearby_connections_manager_->Disconnect(kRemoteEndpointId);
   Sync();
 
   EXPECT_TRUE(close_notification.WaitForNotificationWithTimeout(
@@ -1252,7 +1255,8 @@ TEST_F(NearbyConnectionsManagerImplTest, ConnectTimeout) {
       local_endpoint_info, kRemoteEndpointId,
       /*bluetooth_mac_address=*/std::nullopt, DataUsage::OFFLINE_DATA_USAGE,
       TransportType::kHighQuality,
-      [&](NearbyConnection* connection, Status status) {
+      [&](absl::string_view endpoint_id, NearbyConnection* connection,
+          Status status) {
         nearby_connection = connection;
         run_notification.Notify();
       });
@@ -2009,5 +2013,4 @@ TEST_F(NearbyConnectionsManagerImplTest, ProcessUnknownFilePathsToDelete) {
 }
 
 }  // namespace NearbyConnectionsManagerUnitTests
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/nearby_sharing_service_impl.cc b/sharing/nearby_sharing_service_impl.cc
index 89e50d4c17..c0afcab66d 100644
--- a/sharing/nearby_sharing_service_impl.cc
+++ b/sharing/nearby_sharing_service_impl.cc
@@ -964,7 +964,7 @@ void NearbySharingServiceImpl::DoCancel(
 
       session->WriteCancelFrame();
     } else {
-      session->connection()->Close();
+      session->Disconnect();
     }
   } else {
     LOG(INFO) << "Disconnect endpoint id:" << session->endpoint_id();
@@ -2389,14 +2389,13 @@ void NearbySharingServiceImpl::OnTransferStarted(bool is_incoming) {
 }
 
 void NearbySharingServiceImpl::OnOutgoingConnection(
-    int64_t share_target_id, NearbyConnection* connection, Status status) {
+    int64_t share_target_id, absl::string_view endpoint_id,
+    NearbyConnection* connection, Status status) {
   OutgoingShareSession* session = GetOutgoingShareSession(share_target_id);
   if (session == nullptr) {
     LOG(WARNING) << "Nearby connection connected, but share target "
                  << share_target_id << " already disconnected.";
-    if (connection != nullptr) {
-      connection->Close();
-    }
+    nearby_connections_manager_->Disconnect(endpoint_id);
     return;
   }
   if (connection != nullptr) {
@@ -2652,7 +2651,7 @@ void NearbySharingServiceImpl::OnOutgoingTransferUpdate(
 void NearbySharingServiceImpl::CloseConnection(int64_t share_target_id) {
   ShareSession* session = GetShareSession(share_target_id);
   if (session != nullptr && session->IsConnected()) {
-    session->connection()->Close();
+    session->Disconnect();
     return;
   }
   LOG(WARNING) << __func__ << ": Invalid connection for target - "
diff --git a/sharing/nearby_sharing_service_impl.h b/sharing/nearby_sharing_service_impl.h
index 8faf088de4..f723568ad3 100644
--- a/sharing/nearby_sharing_service_impl.h
+++ b/sharing/nearby_sharing_service_impl.h
@@ -310,6 +310,7 @@ class NearbySharingServiceImpl
   void OnTransferStarted(bool is_incoming);
 
   void OnOutgoingConnection(int64_t share_target_id,
+                            absl::string_view endpoint_id,
                             NearbyConnection* connection, Status status);
 
   void CreatePayloads(
diff --git a/sharing/nearby_sharing_service_impl_test.cc b/sharing/nearby_sharing_service_impl_test.cc
index fa066751a5..5133a38e7e 100644
--- a/sharing/nearby_sharing_service_impl_test.cc
+++ b/sharing/nearby_sharing_service_impl_test.cc
@@ -97,8 +97,7 @@
 #include "sharing/wifi_credentials_attachment.h"
 #include "google/protobuf/repeated_ptr_field.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 
 using ConnectionType = ::nearby::ConnectivityManager::ConnectionType;
@@ -428,8 +427,8 @@ class NearbySharingServiceImplTest : public testing::Test {
         std::make_unique<FakeTaskRunner>(fake_context_.fake_clock(), 1);
     sharing_service_task_runner_ = fake_task_runner.get();
     fake_nearby_connections_manager_ = new FakeNearbyConnectionsManager();
-    connection_ = std::make_unique<NearbyConnectionImpl>(
-        fake_device_info_, fake_nearby_connections_manager_, kEndpointId);
+    connection_ =
+        std::make_unique<NearbyConnectionImpl>(fake_device_info_, kEndpointId);
     fake_nearby_connections_manager_->set_send_payload_callback(
         [this](std::unique_ptr<Payload> payload,
                std::weak_ptr<NearbyConnectionsManager::PayloadStatusListener>
@@ -2498,8 +2497,9 @@ TEST_F(NearbySharingServiceImplTest,
   ScopedReceiveSurface r(service_.get(), &callback);
   EXPECT_CALL(*mock_app_info_, SetActiveFlag());
   StartIncomingConnection();
-  sharing_service_task_runner_->PostTask([this]() {
-    connection_->Close();
+  sharing_service_task_runner_->PostTask(
+      [this]() {
+    fake_nearby_connections_manager_->Disconnect(kEndpointId);
     // FakeNearbyConnectionsManager does not delete the connection on close.
     connection_.reset();
   });
@@ -2531,7 +2531,7 @@ TEST_F(NearbySharingServiceImplTest,
   ProcessLatestPublicCertificateDecryption(/*expected_num_calls=*/1,
                                            /*success=*/true);
   sharing_service_task_runner_->PostTask([this]() {
-    connection_->Close();
+    fake_nearby_connections_manager_->Disconnect(kEndpointId);
     // FakeNearbyConnectionsManager does not delete the connection on close.
     connection_.reset();
   });
@@ -2657,8 +2657,9 @@ TEST_F(NearbySharingServiceImplTest,
         EXPECT_EQ(metadata.status(), TransferMetadata::Status::kFailed);
       }));
 
-  sharing_service_task_runner_->PostTask([this]() {
-    connection_->Close();
+  sharing_service_task_runner_->PostTask(
+      [this]() {
+    fake_nearby_connections_manager_->Disconnect(kEndpointId);
     // FakeNearbyConnectionsManager does not delete the connection on close.
     connection_.reset();
   });
@@ -3600,8 +3601,9 @@ TEST_F(NearbySharingServiceImplTest, SendTextSuccessClosedConnection) {
           .has_value());
 
   // Call disconnect on the connection early before the timeout has passed.
-  sharing_service_task_runner_->PostTask([this]() {
-    connection_->Close();
+  sharing_service_task_runner_->PostTask(
+      [this]() {
+    fake_nearby_connections_manager_->Disconnect(kEndpointId);
     // FakeNearbyConnectionsManager does not destroy the connection.
     connection_.reset();
   });
@@ -5067,5 +5069,4 @@ TEST_F(NearbySharingServiceImplTest, NotifyLogoutSucceededWithCredentialError) {
 }
 
 }  // namespace NearbySharingServiceUnitTests
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/outgoing_share_session.cc b/sharing/outgoing_share_session.cc
index 767b5bf328..77ea8b5e8a 100644
--- a/sharing/outgoing_share_session.cc
+++ b/sharing/outgoing_share_session.cc
@@ -543,7 +543,9 @@ void OutgoingShareSession::Connect(
     std::vector<uint8_t> endpoint_info,
     std::optional<std::vector<uint8_t>> bluetooth_mac_address,
     DataUsage data_usage, bool disable_wifi_hotspot,
-    std::function<void(NearbyConnection* connection, Status status)> callback) {
+    std::function<void(absl::string_view endpoint_id,
+                       NearbyConnection* connection, Status status)>
+        callback) {
   connection_start_time_ = clock().Now();
   connections_manager().Connect(
       std::move(endpoint_info), endpoint_id(), std::move(bluetooth_mac_address),
diff --git a/sharing/outgoing_share_session.h b/sharing/outgoing_share_session.h
index 4b729bc8f3..b8d58febcb 100644
--- a/sharing/outgoing_share_session.h
+++ b/sharing/outgoing_share_session.h
@@ -148,7 +148,8 @@ class OutgoingShareSession : public ShareSession {
                std::optional<std::vector<uint8_t>> bluetooth_mac_address,
                nearby::sharing::proto::DataUsage data_usage,
                bool disable_wifi_hotspot,
-               std::function<void(NearbyConnection* connection, Status status)>
+               std::function<void(absl::string_view endpoint_id,
+                                  NearbyConnection* connection, Status status)>
                    callback);
 
   // Called to process the result of a connection attempt.
diff --git a/sharing/outgoing_share_session_test.cc b/sharing/outgoing_share_session_test.cc
index 9caac39071..4e20834dd4 100644
--- a/sharing/outgoing_share_session_test.cc
+++ b/sharing/outgoing_share_session_test.cc
@@ -123,6 +123,11 @@ class OutgoingShareSessionTest : public ::testing::Test {
                 Log(Matcher<const SharingLog&>(
                     AllOf((HasCategory(EventCategory::SENDING_EVENT),
                            HasEventType(EventType::ESTABLISH_CONNECTION))))));
+    connections_manager_.set_nearby_connection(connection);
+    session_.Connect({}, {}, proto::DataUsage::ONLINE_DATA_USAGE,
+                     /*disable_wifi_hotspot=*/false,
+                     [](absl::string_view endpoint_id,
+                        NearbyConnection* connection, Status status) {});
     EXPECT_THAT(session_.OnConnectResult(connection, Status::kSuccess),
                 IsTrue());
   }
@@ -286,15 +291,16 @@ TEST_F(OutgoingShareSessionTest, ConnectNoDisableWifiHotspot) {
   std::vector<uint8_t> bluetooth_mac_address = {5, 6, 7, 8};
   file1_.set_size(1000000);  // 1MB
   InitSendAttachments(CreateDefaultAttachmentContainer());
-  NearbyConnectionImpl nearby_connection(device_info_, &connections_manager_,
-                                         kEndpointId);
+  NearbyConnectionImpl nearby_connection(device_info_, kEndpointId);
   connections_manager_.set_nearby_connection(&nearby_connection);
 
   session_.Connect(
       endpoint_info, bluetooth_mac_address,
       nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
       /*disable_wifi_hotspot=*/false,
-      [&nearby_connection](NearbyConnection* connection, Status status) {
+      [&nearby_connection](absl::string_view endpoint_id,
+                           NearbyConnection* connection, Status status) {
+        EXPECT_THAT(endpoint_id, Eq(kEndpointId));
         EXPECT_THAT(connection, Eq(&nearby_connection));
       });
 
@@ -313,15 +319,16 @@ TEST_F(OutgoingShareSessionTest, ConnectDisableWifiHotspot) {
   std::vector<uint8_t> bluetooth_mac_address = {5, 6, 7, 8};
   file1_.set_size(1000000);  // 1MB
   InitSendAttachments(CreateDefaultAttachmentContainer());
-  NearbyConnectionImpl nearby_connection(device_info_, &connections_manager_,
-                                         kEndpointId);
+  NearbyConnectionImpl nearby_connection(device_info_, kEndpointId);
   connections_manager_.set_nearby_connection(&nearby_connection);
 
   session_.Connect(
       endpoint_info, bluetooth_mac_address,
       nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
       /*disable_wifi_hotspot=*/true,
-      [&nearby_connection](NearbyConnection* connection, Status status) {
+      [&nearby_connection](absl::string_view endpoint_id,
+                           NearbyConnection* connection, Status status) {
+        EXPECT_THAT(endpoint_id, Eq(kEndpointId));
         EXPECT_THAT(connection, Eq(&nearby_connection));
       });
 
@@ -340,14 +347,15 @@ TEST_F(OutgoingShareSessionTest, OnConnectResultSuccessLogsSessionDuration) {
   session_.set_session_id(1234);
   std::vector<uint8_t> endpoint_info = {1, 2, 3, 4};
   std::vector<uint8_t> bluetooth_mac_address = {5, 6, 7, 8};
-  NearbyConnectionImpl nearby_connection(device_info_, &connections_manager_,
-                                         kEndpointId);
+  NearbyConnectionImpl nearby_connection(device_info_, kEndpointId);
   connections_manager_.set_nearby_connection(&nearby_connection);
   session_.Connect(
       endpoint_info, bluetooth_mac_address,
       nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
       /*disable_wifi_hotspot=*/false,
-      [&nearby_connection](NearbyConnection* connection, Status status) {
+      [&nearby_connection](absl::string_view endpoint_id,
+                           NearbyConnection* connection, Status status) {
+        EXPECT_THAT(endpoint_id, Eq(kEndpointId));
         EXPECT_THAT(connection, Eq(&nearby_connection));
       });
   fake_clock_.FastForward(absl::Seconds(10));
@@ -375,7 +383,8 @@ TEST_F(OutgoingShareSessionTest, OnConnectResultFailureLogsSessionDuration) {
   session_.Connect(endpoint_info, bluetooth_mac_address,
                    nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
                    /*disable_wifi_hotspot=*/false,
-                   [](NearbyConnection* connection, Status status) {});
+                   [](absl::string_view endpoint_id,
+                      NearbyConnection* connection, Status status) {});
   fake_clock_.FastForward(absl::Seconds(10));
   EXPECT_CALL(
       mock_event_logger_,
@@ -405,8 +414,7 @@ TEST_F(OutgoingShareSessionTest, SendIntroductionWithoutPayloads) {
 TEST_F(OutgoingShareSessionTest, SendIntroductionSuccess) {
   InitSendAttachments(CreateDefaultAttachmentContainer());
   session_.set_session_id(1234);
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
   std::vector<NearbyFileHandler::FileInfo> file_infos;
   file_infos.push_back({
@@ -486,8 +494,7 @@ TEST_F(OutgoingShareSessionTest, SendIntroductionTimeout) {
       std::vector<WifiCredentialsAttachment>{});
   InitSendAttachments(std::move(container));
   session_.set_session_id(1234);
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
   session_.CreateTextPayloads();
   EXPECT_CALL(
@@ -514,8 +521,7 @@ TEST_F(OutgoingShareSessionTest, SendIntroductionTimeoutCancelled) {
       std::vector<WifiCredentialsAttachment>{});
   InitSendAttachments(std::move(container));
   session_.set_session_id(1234);
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
   session_.CreateTextPayloads();
   EXPECT_CALL(
@@ -551,8 +557,7 @@ TEST_F(OutgoingShareSessionTest, AcceptTransferNotConnected) {
 }
 
 TEST_F(OutgoingShareSessionTest, AcceptTransferNotReady) {
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
 
@@ -567,8 +572,7 @@ TEST_F(OutgoingShareSessionTest, AcceptTransferSuccess) {
       std::vector<WifiCredentialsAttachment>{});
   InitSendAttachments(std::move(container));
   session_.set_session_id(1234);
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
   session_.CreateTextPayloads();
   EXPECT_CALL(
@@ -659,8 +663,7 @@ TEST_F(OutgoingShareSessionTest, HandleConnectionResponseTimeoutResponse) {
 TEST_F(OutgoingShareSessionTest, HandleConnectionResponseAcceptResponse) {
   ConnectionResponseFrame response;
   response.set_status(ConnectionResponseFrame::ACCEPT);
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
   EXPECT_CALL(transfer_metadata_callback_,
@@ -721,8 +724,7 @@ TEST_F(OutgoingShareSessionTest, SendPayloadsDisableCancellationOptimization) {
                          HasEventType(EventType::SEND_ATTACHMENTS_START),
                          Property(&SharingLog::send_attachments_start,
                                   HasSessionId(1234)))))));
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
 
   session_.SendPayloads(
@@ -765,8 +767,7 @@ TEST_F(OutgoingShareSessionTest, SendPayloadsEnableCancellationOptimization) {
                          HasEventType(EventType::SEND_ATTACHMENTS_START),
                          Property(&SharingLog::send_attachments_start,
                                   HasSessionId(1234)))))));
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
 
   session_.SendPayloads(
@@ -810,8 +811,7 @@ TEST_F(OutgoingShareSessionTest, SendNextPayload) {
                          HasEventType(EventType::SEND_ATTACHMENTS_START),
                          Property(&SharingLog::send_attachments_start,
                                   HasSessionId(1234)))))));
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   ConnectionSuccess(&connection);
 
   session_.SendPayloads(
@@ -848,8 +848,7 @@ TEST_F(OutgoingShareSessionTest, SendNextPayload) {
 }
 
 TEST_F(OutgoingShareSessionTest, ProcessKeyVerificationResultFail) {
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
   session_.SetTokenForTests("1234");
@@ -865,8 +864,7 @@ TEST_F(OutgoingShareSessionTest, ProcessKeyVerificationResultFail) {
 }
 
 TEST_F(OutgoingShareSessionTest, ProcessKeyVerificationResultSuccess) {
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
   session_.SetTokenForTests("1234");
@@ -882,8 +880,7 @@ TEST_F(OutgoingShareSessionTest, ProcessKeyVerificationResultSuccess) {
 }
 
 TEST_F(OutgoingShareSessionTest, DelayCompleteMetadataReceiverDisconnect) {
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
   TransferMetadata complete_metadata =
@@ -901,16 +898,15 @@ TEST_F(OutgoingShareSessionTest, DelayCompleteMetadataReceiverDisconnect) {
 }
 
 TEST_F(OutgoingShareSessionTest, DelayCompleteMetadataDisconnectTimeout) {
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   std::vector<uint8_t> endpoint_info = {1, 2, 3, 4};
   std::vector<uint8_t> bluetooth_mac_address = {5, 6, 7, 8};
-  session_.Connect(
-      endpoint_info, bluetooth_mac_address,
-      nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
-      /*disable_wifi_hotspot=*/false,
-      [&](NearbyConnection* connection, Status status) {});
+  session_.Connect(endpoint_info, bluetooth_mac_address,
+                   nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
+                   /*disable_wifi_hotspot=*/false,
+                   [&](absl::string_view endpoint_id,
+                       NearbyConnection* connection, Status status) {});
   ConnectionSuccess(&connection);
   EXPECT_THAT(
       connections_manager_.connection_endpoint_info(kEndpointId).has_value(),
@@ -961,8 +957,7 @@ TEST_F(OutgoingShareSessionTest, UpdateSessionForDedupWithoutCertificate) {
 
 TEST_F(OutgoingShareSessionTest, UpdateSessionForDedupConnectedIsNoOp) {
   auto share_target_org = session_.share_target();
-  NearbyConnectionImpl connection(device_info_, &connections_manager_,
-                                  kEndpointId);
+  NearbyConnectionImpl connection(device_info_, kEndpointId);
   session_.set_session_id(1234);
   ConnectionSuccess(&connection);
   ShareTarget share_target2{
diff --git a/sharing/paired_key_verification_runner.cc b/sharing/paired_key_verification_runner.cc
index fdddfa5af5..7952d47c10 100644
--- a/sharing/paired_key_verification_runner.cc
+++ b/sharing/paired_key_verification_runner.cc
@@ -26,6 +26,7 @@
 #include <utility>
 #include <vector>
 
+#include "absl/functional/any_invocable.h"
 #include "absl/time/time.h"
 #include "internal/platform/clock.h"
 #include "proto/sharing_enums.pb.h"
@@ -35,14 +36,12 @@
 #include "sharing/certificates/nearby_share_decrypted_public_certificate.h"
 #include "sharing/incoming_frames_reader.h"
 #include "sharing/internal/public/logging.h"
-#include "sharing/nearby_connection.h"
 #include "sharing/proto/enums.pb.h"
 #include "sharing/proto/rpc_resources.pb.h"
 #include "sharing/proto/timestamp.pb.h"
 #include "sharing/proto/wire_format.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
 using ::location::nearby::proto::sharing::OSType;
 using ::nearby::sharing::proto::DeviceVisibility;
@@ -94,22 +93,22 @@ std::ostream& operator<<(
 PairedKeyVerificationRunner::PairedKeyVerificationRunner(
     Clock* clock, OSType os_type, bool share_target_is_incoming,
     const VisibilityHistory& visibility_history,
-    const std::vector<uint8_t>& token, NearbyConnection* connection,
+    const std::vector<uint8_t>& token,
+    absl::AnyInvocable<void(const Frame& frame)> frame_writer,
     const std::optional<NearbyShareDecryptedPublicCertificate>& certificate,
     NearbyShareCertificateManager* certificate_manager,
     IncomingFramesReader* frames_reader, absl::Duration read_frame_timeout)
     : clock_(clock),
       os_type_(os_type),
       raw_token_(token),
-      connection_(connection),
+      frame_writer_(std::move(frame_writer)),
       certificate_(certificate),
       certificate_manager_(certificate_manager),
       frames_reader_(frames_reader),
       read_frame_timeout_(read_frame_timeout) {
-  NL_DCHECK(clock_);
-  NL_DCHECK(connection);
-  NL_DCHECK(certificate_manager);
-  NL_DCHECK(frames_reader);
+  DCHECK(clock_);
+  DCHECK(certificate_manager);
+  DCHECK(frames_reader);
 
   if (share_target_is_incoming) {
     local_prefix_ = kNearbyShareReceiverVerificationPrefix;
@@ -131,7 +130,7 @@ PairedKeyVerificationRunner::~PairedKeyVerificationRunner() = default;
 
 void PairedKeyVerificationRunner::Run(
     std::function<void(PairedKeyVerificationResult, OSType)> callback) {
-  NL_DCHECK(!callback_);
+  DCHECK(!callback_);
   callback_ = std::move(callback);
   verification_result_ = PairedKeyVerificationResult::kSuccess;
 
@@ -141,7 +140,7 @@ void PairedKeyVerificationRunner::Run(
       [&, runner = GetWeakPtr()](std::optional<V1Frame> frame) {
         auto verification_runner = runner.lock();
         if (verification_runner == nullptr) {
-          NL_LOG(WARNING) << "PairedKeyVerificationRunner is released before.";
+          LOG(WARNING) << "PairedKeyVerificationRunner is released before.";
           return;
         }
         OnReadPairedKeyEncryptionFrame(std::move(frame));
@@ -152,8 +151,7 @@ void PairedKeyVerificationRunner::Run(
 void PairedKeyVerificationRunner::OnReadPairedKeyEncryptionFrame(
     std::optional<V1Frame> frame) {
   if (!frame.has_value()) {
-    NL_LOG(WARNING) << __func__
-                    << ": Failed to read remote paired key encryption";
+    LOG(WARNING) << __func__ << ": Failed to read remote paired key encryption";
     std::move(callback_)(PairedKeyVerificationResult::kFail,
                          OSType::UNKNOWN_OS_TYPE);
     return;
@@ -171,14 +169,14 @@ void PairedKeyVerificationRunner::OnReadPairedKeyEncryptionFrame(
   }
 
   ApplyResult(auth_token_hash_result);
-  NL_VLOG(1) << __func__ << ": Remote public certificate verification result "
-             << auth_token_hash_result;
+  VLOG(1) << __func__ << ": Remote public certificate verification result "
+          << auth_token_hash_result;
 
   PairedKeyVerificationResult local_result =
       VerifyPairedKeyEncryptionFrame(*frame);
   ApplyResult(local_result);
-  NL_VLOG(1) << __func__ << ": Paired key encryption verification result "
-             << local_result;
+  VLOG(1) << __func__ << ": Paired key encryption verification result "
+          << local_result;
 
   SendPairedKeyResultFrame(local_result);
 
@@ -187,7 +185,7 @@ void PairedKeyVerificationRunner::OnReadPairedKeyEncryptionFrame(
       [this, runner = GetWeakPtr()](std::optional<V1Frame> frame) {
         auto verification_runner = runner.lock();
         if (verification_runner == nullptr) {
-          NL_LOG(WARNING) << "PairedKeyVerificationRunner is released before.";
+          LOG(WARNING) << "PairedKeyVerificationRunner is released before.";
           return;
         }
         OnReadPairedKeyResultFrame(std::move(frame));
@@ -198,7 +196,7 @@ void PairedKeyVerificationRunner::OnReadPairedKeyEncryptionFrame(
 void PairedKeyVerificationRunner::OnReadPairedKeyResultFrame(
     std::optional<V1Frame> frame) {
   if (!frame.has_value()) {
-    NL_LOG(WARNING) << __func__ << ": Failed to read remote paired key result";
+    LOG(WARNING) << __func__ << ": Failed to read remote paired key result";
     std::move(callback_)(PairedKeyVerificationResult::kFail,
                          OSType::UNKNOWN_OS_TYPE);
     return;
@@ -207,11 +205,10 @@ void PairedKeyVerificationRunner::OnReadPairedKeyResultFrame(
   PairedKeyVerificationResult remote_result =
       Convert(frame->paired_key_result().status());
   ApplyResult(remote_result);
-  NL_VLOG(1) << __func__ << ": Paired key result frame result "
-             << remote_result;
+  VLOG(1) << __func__ << ": Paired key result frame result " << remote_result;
 
-  NL_VLOG(1) << __func__ << ": Combined verification result "
-             << verification_result_;
+  VLOG(1) << __func__ << ": Combined verification result "
+          << verification_result_;
 
   OSType os_type = OSType::UNKNOWN_OS_TYPE;
   if (frame->paired_key_result().has_os_type()) {
@@ -250,10 +247,7 @@ void PairedKeyVerificationRunner::SendPairedKeyResultFrame(
   // Set OS type to allow remote device knowns the paring device OS type.
   result_frame->set_os_type(os_type_);
 
-  std::vector<uint8_t> data(frame.ByteSizeLong());
-  frame.SerializeToArray(data.data(), frame.ByteSizeLong());
-
-  connection_->Write(std::move(data));
+  frame_writer_(frame);
 }
 
 void PairedKeyVerificationRunner::SendPairedKeyEncryptionFrame() {
@@ -281,7 +275,7 @@ void PairedKeyVerificationRunner::SendPairedKeyEncryptionFrame() {
       v1_frame->mutable_paired_key_encryption();
   encryption_frame->set_signed_data(signature->data(), signature->size());
   if (IsVisibilityRecentlyUpdated()) {
-    NL_LOG(INFO)
+    LOG(INFO)
         << "Attempts to sign authentication token with a previous private key.";
     std::optional<std::vector<uint8_t>> optional_signature =
         certificate_manager_->SignWithPrivateCertificate(
@@ -295,10 +289,8 @@ void PairedKeyVerificationRunner::SendPairedKeyEncryptionFrame() {
   }
   encryption_frame->set_secret_id_hash(certificate_id_hash.data(),
                                        certificate_id_hash.size());
-  std::vector<uint8_t> data(frame.ByteSizeLong());
-  frame.SerializeToArray(data.data(), frame.ByteSizeLong());
 
-  connection_->Write(std::move(data));
+  frame_writer_(frame);
 }
 
 PairedKeyVerificationRunner::PairedKeyVerificationResult
@@ -314,12 +306,11 @@ PairedKeyVerificationRunner::VerifyAuthTokenHashWithPrivateCertificate(
   std::vector<uint8_t> frame_hash_data{frame_hash.begin(), frame_hash.end()};
 
   if (hash.has_value() && *hash == frame_hash_data) {
-    NL_VLOG(1) << __func__
-               << ": Successfully verified remote public certificate.";
+    VLOG(1) << __func__ << ": Successfully verified remote public certificate.";
     return PairedKeyVerificationResult::kSuccess;
   }
 
-  NL_VLOG(1) << __func__ << ": Unable to verify remote public certificate.";
+  VLOG(1) << __func__ << ": Unable to verify remote public certificate.";
   return PairedKeyVerificationResult::kUnable;
 }
 
@@ -327,9 +318,9 @@ PairedKeyVerificationRunner::PairedKeyVerificationResult
 PairedKeyVerificationRunner::VerifyPairedKeyEncryptionFrame(
     const V1Frame& frame) {
   if (!certificate_) {
-    NL_VLOG(1) << __func__
-               << ": Unable to verify remote paired key encryption frame. "
-                  "Remote side is not a known share target.";
+    VLOG(1) << __func__
+            << ": Unable to verify remote paired key encryption frame. "
+               "Remote side is not a known share target.";
     return PairedKeyVerificationResult::kUnable;
   }
 
@@ -338,10 +329,9 @@ PairedKeyVerificationRunner::VerifyPairedKeyEncryptionFrame(
   if (!certificate_->VerifySignature(PadPrefix(remote_prefix_, raw_token_),
                                      data)) {
     if (!frame.paired_key_encryption().has_optional_signed_data()) {
-      NL_LOG(WARNING)
-          << __func__
-          << ": Unable to verify remote paired key encryption frame. "
-             "no optional signed data.";
+      LOG(WARNING) << __func__
+                   << ": Unable to verify remote paired key encryption frame. "
+                      "no optional signed data.";
       return PairedKeyVerificationResult::kFail;
     }
     // Verify optional signed data.
@@ -351,18 +341,17 @@ PairedKeyVerificationRunner::VerifyPairedKeyEncryptionFrame(
                                        optional_signed_data.end());
     if (certificate_->VerifySignature(PadPrefix(remote_prefix_, raw_token_),
                                       optional_data)) {
-      NL_LOG(INFO) << "Successfully verified remote paired key encryption "
-                      "frame with the optional signed data.";
+      LOG(INFO) << "Successfully verified remote paired key encryption "
+                   "frame with the optional signed data.";
     } else {
-      NL_LOG(WARNING)
-          << __func__
-          << ": Unable to verify remote paired key encryption frame.";
+      LOG(WARNING) << __func__
+                   << ": Unable to verify remote paired key encryption frame.";
       return PairedKeyVerificationResult::kFail;
     }
   }
 
-  NL_VLOG(1) << __func__
-             << ": Successfully verified remote paired key encryption frame.";
+  VLOG(1) << __func__
+          << ": Successfully verified remote paired key encryption frame.";
   return PairedKeyVerificationResult::kSuccess;
 }
 
@@ -397,5 +386,4 @@ bool PairedKeyVerificationRunner::IsVisibilityRecentlyUpdated() const {
           kRelaxAfterSetVisibilityTimeout);
 }
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/paired_key_verification_runner.h b/sharing/paired_key_verification_runner.h
index 131f241fa1..6bfc80269b 100644
--- a/sharing/paired_key_verification_runner.h
+++ b/sharing/paired_key_verification_runner.h
@@ -22,18 +22,17 @@
 #include <optional>
 #include <vector>
 
+#include "absl/functional/any_invocable.h"
 #include "absl/time/time.h"
 #include "internal/platform/clock.h"
 #include "proto/sharing_enums.pb.h"
 #include "sharing/certificates/nearby_share_certificate_manager.h"
 #include "sharing/certificates/nearby_share_decrypted_public_certificate.h"
 #include "sharing/incoming_frames_reader.h"
-#include "sharing/nearby_connection.h"
 #include "sharing/proto/enums.pb.h"
 #include "sharing/proto/wire_format.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 
 class PairedKeyVerificationRunner
     : public std::enable_shared_from_this<PairedKeyVerificationRunner> {
@@ -59,7 +58,10 @@ class PairedKeyVerificationRunner
       Clock* clock, location::nearby::proto::sharing::OSType os_type,
       bool share_target_is_incoming,
       const VisibilityHistory& visibility_history,
-      const std::vector<uint8_t>& token, NearbyConnection* connection,
+      const std::vector<uint8_t>& token,
+      absl::AnyInvocable<
+          void(const nearby::sharing::service::proto::Frame& frame)>
+          frame_writer,
       const std::optional<NearbyShareDecryptedPublicCertificate>& certificate,
       NearbyShareCertificateManager* certificate_manager,
       IncomingFramesReader* frames_reader, absl::Duration read_frame_timeout);
@@ -97,7 +99,8 @@ class PairedKeyVerificationRunner
   const location::nearby::proto::sharing::OSType os_type_;
   VisibilityHistory visibility_history_;
   std::vector<uint8_t> raw_token_;
-  NearbyConnection* connection_;
+  absl::AnyInvocable<void(const nearby::sharing::service::proto::Frame& frame)>
+      frame_writer_;
   std::optional<NearbyShareDecryptedPublicCertificate> certificate_;
   NearbyShareCertificateManager* certificate_manager_;
   IncomingFramesReader* frames_reader_;
@@ -110,7 +113,6 @@ class PairedKeyVerificationRunner
   char remote_prefix_;
 };
 
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
 
 #endif  // THIRD_PARTY_NEARBY_SHARING_PAIRED_KEY_VERIFICATION_RUNNER_H_
diff --git a/sharing/paired_key_verification_runner_test.cc b/sharing/paired_key_verification_runner_test.cc
index 9e1527b494..54899f7e2b 100644
--- a/sharing/paired_key_verification_runner_test.cc
+++ b/sharing/paired_key_verification_runner_test.cc
@@ -47,8 +47,7 @@
 #include "sharing/proto/rpc_resources.pb.h"
 #include "sharing/proto/wire_format.pb.h"
 
-namespace nearby {
-namespace sharing {
+namespace nearby::sharing {
 namespace {
 
 using V1Frame = ::nearby::sharing::service::proto::V1Frame;
@@ -188,8 +187,7 @@ class PairedKeyVerificationRunnerTest : public testing::Test {
   };
 
   PairedKeyVerificationRunnerTest()
-      : connection_(fake_device_info_, &fake_connections_manager_,
-                    "test_enpoint_id"),
+      : connection_(fake_device_info_, "test_enpoint_id"),
         frames_reader_(fake_task_runner_, &connection_) {
     fake_connections_manager_.set_send_payload_callback(
         [this](std::unique_ptr<Payload> payload,
@@ -220,8 +218,11 @@ class PairedKeyVerificationRunnerTest : public testing::Test {
 
     auto runner = std::make_shared<PairedKeyVerificationRunner>(
         &fake_clock_, OSType::WINDOWS, is_incoming, visibility_history,
-        GetAuthToken(), &connection_, std::move(public_certificate),
-        &certificate_manager_, &frames_reader_, kTimeout);
+        GetAuthToken(), [this](const Frame& frame) {
+          frames_data_.push(std::make_unique<Frame>(frame));
+        },
+        std::move(public_certificate), &certificate_manager_, &frames_reader_,
+        kTimeout);
 
     runner->Run(
         [&, expected_result, expected_os_type](
@@ -432,17 +433,15 @@ TEST_P(ParameterisedPairedKeyVerificationRunnerTest,
   PairedKeyVerificationRunner::PairedKeyVerificationResult expected_result =
       Merge(params.result, result_frame.status());
 
-  NL_LOG(ERROR) << "ValidEncryptionFrame_ValidResultFrame: " << "is_incoming="
-                << params.is_incoming
-                << ", has_valid_cert=" << params.has_valid_certificate
-                << ", encryption_frame_type="
-                << (int)params.encryption_frame_type
-                << ", result=" << (int)params.result
-                << ", expected_result=" << (int)expected_result
-                << ", result_frame=" << (int)result_frame.status()
-                << ", visibility=" << (int)visibility_history.visibility
-                << ", last_visibility="
-                << (int)visibility_history.last_visibility;
+  LOG(ERROR) << "ValidEncryptionFrame_ValidResultFrame: " << "is_incoming="
+             << params.is_incoming
+             << ", has_valid_cert=" << params.has_valid_certificate
+             << ", encryption_frame_type=" << (int)params.encryption_frame_type
+             << ", result=" << (int)params.result
+             << ", expected_result=" << (int)expected_result
+             << ", result_frame=" << (int)result_frame.status()
+             << ", visibility=" << (int)visibility_history.visibility
+             << ", last_visibility=" << (int)visibility_history.last_visibility;
 
   SetUpPairedKeyEncryptionFrame(params.encryption_frame_type);
   bool encryption_frame_timeout =
@@ -518,5 +517,4 @@ INSTANTIATE_TEST_SUITE_P(
                      testing::ValuesIn(GenerateVisibilityHistory())));
 
 }  // namespace
-}  // namespace sharing
-}  // namespace nearby
+}  // namespace nearby::sharing
diff --git a/sharing/share_session.cc b/sharing/share_session.cc
index 292bd74626..52b7a76b1e 100644
--- a/sharing/share_session.cc
+++ b/sharing/share_session.cc
@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "absl/functional/any_invocable.h"
+#include "absl/functional/bind_front.h"
 #include "absl/strings/str_format.h"
 #include "internal/platform/clock.h"
 #include "internal/platform/task_runner.h"
@@ -34,6 +35,7 @@
 #include "sharing/internal/public/logging.h"
 #include "sharing/nearby_connection.h"
 #include "sharing/nearby_connections_manager.h"
+#include "sharing/nearby_connections_types.h"
 #include "sharing/paired_key_verification_runner.h"
 #include "sharing/payload_tracker.h"
 #include "sharing/proto/wire_format.pb.h"
@@ -158,11 +160,8 @@ void ShareSession::SetConnection(NearbyConnection* connection) {
 }
 
 void ShareSession::Disconnect() {
-  if (connection_ == nullptr) {
-    return;
-  }
   // Do not clear connection_ here.  It will be cleared in OnDisconnect().
-  connection_->Close();
+  connections_manager_.Disconnect(endpoint_id_);
 }
 
 void ShareSession::Abort(TransferMetadata::Status status) {
@@ -172,14 +171,7 @@ void ShareSession::Abort(TransferMetadata::Status status) {
   // First invoke the appropriate transfer callback with the final
   // |status|.
   UpdateTransferMetadata(TransferMetadataBuilder().set_status(status).build());
-
-  // Close connection if necessary.
-  if (connection_ == nullptr) {
-    return;
-  }
-  // Final status already sent above.  No need to send it again.
-  set_disconnect_status(TransferMetadata::Status::kUnknown);
-  connection_->Close();
+  Disconnect();
 }
 
 void ShareSession::RunPairedKeyVerification(
@@ -198,7 +190,8 @@ void ShareSession::RunPairedKeyVerification(
   token_ = TokenToFourDigitString(*token);
 
   key_verification_runner_ = std::make_shared<PairedKeyVerificationRunner>(
-      &clock_, os_type, IsIncoming(), visibility_history, *token, connection_,
+      &clock_, os_type, IsIncoming(), visibility_history, *token,
+      absl::bind_front(&ShareSession::WriteFrame, this),
       certificate_, certificate_manager, frames_reader_.get(),
       kReadFramesTimeout);
   key_verification_runner_->Run(std::move(callback));
@@ -234,7 +227,10 @@ void ShareSession::WriteFrame(const Frame& frame) {
   std::vector<uint8_t> data(frame.ByteSizeLong());
   frame.SerializeToArray(data.data(), frame.ByteSizeLong());
 
-  connection_->Write(std::move(data));
+  connections_manager_.Send(
+      endpoint_id_, std::make_unique<Payload>(std::move(data)),
+      /*listener=*/
+      std::weak_ptr<NearbyConnectionsManager::PayloadStatusListener>());
 }
 
 void ShareSession::WriteResponseFrame(
diff --git a/sharing/share_session_test.cc b/sharing/share_session_test.cc
index 40a1ebb6bc..819b1b2b40 100644
--- a/sharing/share_session_test.cc
+++ b/sharing/share_session_test.cc
@@ -89,7 +89,8 @@ class TestShareSession : public ShareSession {
         endpoint_info, kEndpointId, bluetooth_mac_address,
         nearby::sharing::proto::DataUsage::ONLINE_DATA_USAGE,
         TransportType::kHighQuality,
-        [&](NearbyConnection* connection, Status status) {});
+        [&](absl::string_view endpoint_id, NearbyConnection* connection,
+            Status status) {});
     SetConnection(connection);
   }
 
@@ -152,8 +153,7 @@ TEST(ShareSessionTest, SetDisconnectStatus) {
 TEST(ShareSessionTest, OnConnectedSucceeds) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
 
   session.SetNearbyConnection(&connection);
   EXPECT_EQ(session.connection(), &connection);
@@ -166,8 +166,7 @@ TEST(ShareSessionTest, IncomingRunPairedKeyVerificationSuccess) {
   share_target.is_incoming = true;
   TestShareSession session(std::string(kEndpointId), share_target);
   session.connections_manager().SetRawAuthenticationToken(kEndpointId, token);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   std::queue<std::vector<uint8_t>> frames_data;
   session.connections_manager().set_send_payload_callback(
@@ -260,8 +259,7 @@ TEST(ShareSessionTest, OnDisconnect) {
 TEST(ShareSessionTest, CancelPayloads) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetAttachmentPayloadId(1, 2);
   session.SetAttachmentPayloadId(3, 4);
@@ -275,8 +273,7 @@ TEST(ShareSessionTest, CancelPayloads) {
 TEST(ShareSessionTest, WriteResponseFrame) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   std::queue<std::vector<uint8_t>> frames_data;
   session.connections_manager().set_send_payload_callback(
@@ -300,8 +297,7 @@ TEST(ShareSessionTest, WriteResponseFrame) {
 TEST(ShareSessionTest, WriteCancelFrame) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   std::queue<std::vector<uint8_t>> frames_data;
   session.connections_manager().set_send_payload_callback(
@@ -323,8 +319,7 @@ TEST(ShareSessionTest, WriteCancelFrame) {
 TEST(ShareSessionTest, HandleKeyVerificationResultFail) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -339,8 +334,7 @@ TEST(ShareSessionTest, HandleKeyVerificationResultSelfShareSuccess) {
   ShareTarget share_target;
   share_target.for_self_share = true;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -355,8 +349,7 @@ TEST(ShareSessionTest, HandleKeyVerificationResultSelfShareSuccess) {
 TEST(ShareSessionTest, HandleKeyVerificationResultNotSelfShareSuccess) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -373,8 +366,7 @@ TEST(ShareSessionTest, HandleKeyVerificationResultSelfShareUnable) {
   ShareTarget share_target;
   share_target.for_self_share = true;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -389,8 +381,7 @@ TEST(ShareSessionTest, HandleKeyVerificationResultSelfShareUnable) {
 TEST(ShareSessionTest, HandleKeyVerificationResultNotSelfShareUnable) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -405,8 +396,7 @@ TEST(ShareSessionTest, HandleKeyVerificationResultNotSelfShareUnable) {
 TEST(ShareSessionTest, HandleKeyVerificationResultUnknown) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   session.SetTokenForTests("9876");
 
@@ -430,8 +420,7 @@ TEST(ShareSessionTest, AbortNotConnected) {
 TEST(ShareSessionTest, AbortConnected) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   EXPECT_TRUE(session.connections_manager()
                   .connection_endpoint_info(kEndpointId)
@@ -451,8 +440,7 @@ TEST(ShareSessionTest, AbortConnected) {
 TEST(ShareSessionTest, Disconnect) {
   ShareTarget share_target;
   TestShareSession session(std::string(kEndpointId), share_target);
-  NearbyConnectionImpl connection(session.device_info(),
-                                  &session.connections_manager(), kEndpointId);
+  NearbyConnectionImpl connection(session.device_info(), kEndpointId);
   session.SetNearbyConnection(&connection);
   EXPECT_TRUE(session.connections_manager()
                   .connection_endpoint_info(kEndpointId)