diff --git a/sharing/fake_nearby_connection.cc b/sharing/fake_nearby_connection.cc
index 71334df9..09c34b15 100644
--- a/sharing/fake_nearby_connection.cc
+++ b/sharing/fake_nearby_connection.cc
@@ -25,7 +25,6 @@
 #include "absl/synchronization/mutex.h"
 #include "internal/platform/task_runner.h"
 #include "sharing/internal/public/logging.h"
-#include "sharing/nearby_connection.h"
 
 namespace nearby {
 namespace sharing {
@@ -33,7 +32,8 @@ FakeNearbyConnection::FakeNearbyConnection(TaskRunner* task_runner)
     : task_runner_(task_runner) {}
 FakeNearbyConnection::~FakeNearbyConnection() = default;
 
-void FakeNearbyConnection::Read(ReadCallback callback) {
+void FakeNearbyConnection::Read(
+    std::function<void(std::optional<std::vector<uint8_t>> bytes)> callback) {
   NL_DCHECK(!closed_);
   {
     absl::MutexLock lock(&read_mutex_);
@@ -107,7 +107,7 @@ bool FakeNearbyConnection::IsClosed() { return closed_; }
 void FakeNearbyConnection::MaybeRunCallback() {
   NL_DCHECK(!closed_);
   std::vector<uint8_t> item;
-  ReadCallback callback;
+  std::function<void(std::optional<std::vector<uint8_t>> bytes)> callback;
   {
     absl::MutexLock lock(&read_mutex_);
     if (!callback_ || read_data_.empty()) return;
diff --git a/sharing/fake_nearby_connection.h b/sharing/fake_nearby_connection.h
index e6cb391d..1723ad28 100644
--- a/sharing/fake_nearby_connection.h
+++ b/sharing/fake_nearby_connection.h
@@ -18,6 +18,7 @@
 #include <stdint.h>
 
 #include <functional>
+#include <optional>
 #include <queue>
 #include <vector>
 
@@ -35,7 +36,8 @@ class FakeNearbyConnection : public NearbyConnection {
   ~FakeNearbyConnection() override;
 
   // NearbyConnection:
-  void Read(ReadCallback callback) override;
+  void Read(std::function<void(std::optional<std::vector<uint8_t>> bytes)>
+                callback) override;
   void Write(std::vector<uint8_t> bytes) override;
   void Close() override;
   void SetDisconnectionListener(std::function<void()> listener) override;
@@ -58,7 +60,8 @@ class FakeNearbyConnection : public NearbyConnection {
   TaskRunner* const task_runner_;
   absl::Mutex read_mutex_;
   bool has_read_callback_been_run_ ABSL_GUARDED_BY(read_mutex_) = false;
-  ReadCallback callback_ ABSL_GUARDED_BY(read_mutex_);
+  std::function<void(std::optional<std::vector<uint8_t>> bytes)> callback_
+      ABSL_GUARDED_BY(read_mutex_);
   std::queue<std::vector<uint8_t>> read_data_ ABSL_GUARDED_BY(read_mutex_);
   absl::Mutex write_mutex_;
   std::queue<std::vector<uint8_t>> write_data_ ABSL_GUARDED_BY(write_mutex_);
diff --git a/sharing/nearby_connection.h b/sharing/nearby_connection.h
index 518cea27..0988e00b 100644
--- a/sharing/nearby_connection.h
+++ b/sharing/nearby_connection.h
@@ -27,28 +27,28 @@ namespace sharing {
 // reads and writes.
 class NearbyConnection {
  public:
-  using ReadCallback =
-      std::function<void(std::optional<std::vector<uint8_t>> bytes)>;
-
   virtual ~NearbyConnection() = default;
 
-  // Reads a stream of bytes from the remote device. Invoke |callback| when
-  // there is incoming data or when the socket is closed. Previously set
-  // callback will be replaced by |callback|. Must not be used on an already
-  // closed connection.
-  virtual void Read(ReadCallback callback) = 0;
+  // Reads a stream of bytes from the remote device. Invoke `callback` when
+  // there is incoming data or when the socket is closed. Must not be used on
+  // an already closed connection.
+  // There should only be 1 outstanding Read() call at a time.
+  // If packet is already available, `callback` will be invoked immediately.
+  virtual void Read(
+      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 |callback| in SetDisconnectionListener is invoked.
+  // be invalidated after `listener` in SetDisconnectionListener is invoked.
   virtual void Close() = 0;
 
-  // Listens to the socket being closed. Invoke |callback| when the socket is
-  // closed. This object will be invalidated after |listener| is invoked.
-  // Previously set listener will be replaced by |listener|.
+  // 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;
 };
 
diff --git a/sharing/nearby_connection_impl.cc b/sharing/nearby_connection_impl.cc
index d67c3b0b..b498930f 100644
--- a/sharing/nearby_connection_impl.cc
+++ b/sharing/nearby_connection_impl.cc
@@ -24,10 +24,9 @@
 #include <vector>
 
 #include "absl/strings/string_view.h"
+#include "absl/synchronization/mutex.h"
 #include "internal/platform/device_info.h"
-#include "internal/platform/mutex_lock.h"
 #include "sharing/internal/public/logging.h"
-#include "sharing/nearby_connection.h"
 #include "sharing/nearby_connections_manager.h"
 #include "sharing/nearby_connections_types.h"
 
@@ -47,43 +46,49 @@ NearbyConnectionImpl::NearbyConnectionImpl(
 }
 
 NearbyConnectionImpl::~NearbyConnectionImpl() {
-  MutexLock lock(&mutex_);
-  if (!device_info_.AllowSleep()) {
-    NL_LOG(ERROR) << __func__ << ":Failed to allow device sleep.";
+  std::function<void()> disconnect_listener;
+  std::function<void(std::optional<std::vector<uint8_t>> bytes)> read_callback;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (!device_info_.AllowSleep()) {
+      NL_LOG(ERROR) << __func__ << ":Failed to allow device sleep.";
+    }
+    disconnect_listener = std::move(disconnect_listener_);
+    read_callback = std::move(read_callback_);
+  }
+  if (disconnect_listener) {
+    disconnect_listener();
   }
 
-  if (disconnect_listener_) {
-    disconnect_listener_();
-  }
-
-  if (read_callback_) {
-    read_callback_(std::nullopt);
+  if (read_callback) {
+    read_callback(std::nullopt);
   }
 }
 
-void NearbyConnectionImpl::Read(ReadCallback callback) {
-  MutexLock lock(&mutex_);
-  if (reads_.empty()) {
-    read_callback_ = std::move(callback);
-    return;
-  }
+void NearbyConnectionImpl::Read(
+    std::function<void(std::optional<std::vector<uint8_t>> bytes)> callback) {
+  std::vector<uint8_t> bytes;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (reads_.empty()) {
+      read_callback_ = std::move(callback);
+      return;
+    }
 
-  std::vector<uint8_t> bytes = std::move(reads_.front());
-  reads_.pop();
+    bytes = std::move(reads_.front());
+    reads_.pop();
+  }
   std::move(callback)(std::move(bytes));
 }
 
 void NearbyConnectionImpl::Write(std::vector<uint8_t> bytes) {
-  MutexLock lock(&mutex_);
-  Payload payload(bytes);
   nearby_connections_manager_->Send(
-      endpoint_id_, std::make_unique<Payload>(payload),
+      endpoint_id_, std::make_unique<Payload>(bytes),
       /*listener=*/
       std::weak_ptr<NearbyConnectionsManager::PayloadStatusListener>());
 }
 
 void NearbyConnectionImpl::Close() {
-  MutexLock lock(&mutex_);
   // 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_);
@@ -91,20 +96,24 @@ void NearbyConnectionImpl::Close() {
 
 void NearbyConnectionImpl::SetDisconnectionListener(
     std::function<void()> listener) {
-  MutexLock lock(&mutex_);
+  absl::MutexLock lock(&mutex_);
   disconnect_listener_ = std::move(listener);
 }
 
 void NearbyConnectionImpl::WriteMessage(std::vector<uint8_t> bytes) {
-  MutexLock lock(&mutex_);
-  if (read_callback_) {
-    auto callback = std::move(read_callback_);
+  std::function<void(std::optional<std::vector<uint8_t>> bytes)> read_callback;
+  {
+    absl::MutexLock lock(&mutex_);
+    if (!read_callback_) {
+      reads_.push(std::move(bytes));
+      return;
+    }
+    read_callback = std::move(read_callback_);
     read_callback_ = nullptr;
-    callback(std::move(bytes));
-    return;
   }
-
-  reads_.push(std::move(bytes));
+  if (read_callback) {
+    read_callback(std::move(bytes));
+  }
 }
 
 }  // namespace sharing
diff --git a/sharing/nearby_connection_impl.h b/sharing/nearby_connection_impl.h
index a944360c..b5eb0d41 100644
--- a/sharing/nearby_connection_impl.h
+++ b/sharing/nearby_connection_impl.h
@@ -17,14 +17,15 @@
 
 #include <cstdint>
 #include <functional>
+#include <optional>
 #include <queue>
 #include <string>
 #include <vector>
 
 #include "absl/base/thread_annotations.h"
 #include "absl/strings/string_view.h"
+#include "absl/synchronization/mutex.h"
 #include "internal/platform/device_info.h"
-#include "internal/platform/mutex.h"
 #include "sharing/nearby_connection.h"
 
 namespace nearby {
@@ -40,21 +41,25 @@ class NearbyConnectionImpl : public NearbyConnection {
   ~NearbyConnectionImpl() override;
 
   // NearbyConnection:
-  void Read(ReadCallback callback) override;
+  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) override;
+  void SetDisconnectionListener(std::function<void()> listener)
+      ABSL_LOCKS_EXCLUDED(mutex_) override;
 
   // Add bytes to the read queue, notifying ReadCallback.
-  void WriteMessage(std::vector<uint8_t> bytes);
+  void WriteMessage(std::vector<uint8_t> bytes) ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
   nearby::DeviceInfo& device_info_;
   NearbyConnectionsManager* const nearby_connections_manager_;
-  std::string endpoint_id_;
+  const std::string endpoint_id_;
 
-  RecursiveMutex mutex_;
-  ReadCallback read_callback_ ABSL_GUARDED_BY(mutex_) = nullptr;
+  absl::Mutex mutex_;
+  std::function<void(std::optional<std::vector<uint8_t>> bytes)> read_callback_
+      ABSL_GUARDED_BY(mutex_) = nullptr;
   std::function<void()> disconnect_listener_ ABSL_GUARDED_BY(mutex_);
 
   // A read queue. The data that we've read from the remote device ends up here