diff --git a/sharing/BUILD b/sharing/BUILD
index 0b6e40db..391203d4 100644
--- a/sharing/BUILD
+++ b/sharing/BUILD
@@ -143,8 +143,9 @@ cc_library(
         "//sharing/internal/public:logging",
         "//sharing/proto:wire_format_cc_proto",
         "@com_google_absl//absl/base:core_headers",
+        "@com_google_absl//absl/memory",
+        "@com_google_absl//absl/synchronization",
         "@com_google_absl//absl/time",
-        "@com_google_absl//absl/types:span",
     ],
 )
 
diff --git a/sharing/incoming_frames_reader.cc b/sharing/incoming_frames_reader.cc
index b713deb7..ccacaa8e 100644
--- a/sharing/incoming_frames_reader.cc
+++ b/sharing/incoming_frames_reader.cc
@@ -16,17 +16,17 @@
 
 #include <stdint.h>
 
+#include <algorithm>
 #include <functional>
-#include <map>
 #include <memory>
 #include <optional>
 #include <queue>
 #include <utility>
 #include <vector>
 
+#include "absl/memory/memory.h"
+#include "absl/synchronization/mutex.h"
 #include "absl/time/time.h"
-#include "absl/types/span.h"
-#include "internal/platform/mutex_lock.h"
 #include "internal/platform/task_runner.h"
 #include "sharing/internal/public/logging.h"
 #include "sharing/nearby_connection.h"
@@ -41,11 +41,12 @@ using FrameType = ::nearby::sharing::service::proto::V1Frame_FrameType;
 using V1Frame = ::nearby::sharing::service::proto::V1Frame;
 using Frame = ::nearby::sharing::service::proto::Frame;
 
-std::unique_ptr<Frame> DecodeFrame(absl::Span<const uint8_t> data) {
+std::unique_ptr<V1Frame> DecodeV1Frame(const std::vector<uint8_t>& data) {
   auto frame = std::make_unique<Frame>();
 
-  if (frame->ParseFromArray(data.data(), data.size())) {
-    return frame;
+  if (frame->ParseFromArray(data.data(), data.size()) &&
+      frame->version() == Frame::V1) {
+    return absl::WrapUnique(frame->release_v1());
   } else {
     return nullptr;
   }
@@ -67,60 +68,55 @@ IncomingFramesReader::~IncomingFramesReader() {
 
 void IncomingFramesReader::ReadFrame(
     std::function<void(std::optional<V1Frame>)> callback) {
-  {
-    MutexLock lock(&mutex_);
-    if (!read_frame_info_queue_.empty()) {
-      ReadFrameInfo read_fame_info{std::nullopt, std::move(callback),
-                                   std::nullopt};
-      read_frame_info_queue_.push(std::move(read_fame_info));
-      return;
-    }
-
-    // Check in the cache for frame.
-    std::optional<V1Frame> cached_frame = GetCachedFrame(std::nullopt);
-    if (cached_frame.has_value()) {
-      callback(std::move(cached_frame));
-      return;
-    }
-
-    ReadFrameInfo read_fame_info{std::nullopt, std::move(callback),
-                                 std::nullopt};
-    read_frame_info_queue_.push(std::move(read_fame_info));
-  }
-  ReadNextFrame();
+  ProcessReadRequest(std::nullopt, std::move(callback), absl::ZeroDuration());
 }
 
 void IncomingFramesReader::ReadFrame(
     FrameType frame_type, std::function<void(std::optional<V1Frame>)> callback,
     absl::Duration timeout) {
+  ProcessReadRequest(frame_type, std::move(callback), timeout);
+}
+
+void IncomingFramesReader::ProcessReadRequest(
+    std::optional<FrameType> frame_type,
+    std::function<void(std::optional<V1Frame>)> callback,
+    absl::Duration timeout) {
+  std::unique_ptr<V1Frame> cached_frame;
   {
-    MutexLock lock(&mutex_);
+    absl::MutexLock lock(&mutex_);
     if (!read_frame_info_queue_.empty()) {
+      // There are already outstanding read requests, just queue this up.
       ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout};
       read_frame_info_queue_.push(std::move(read_fame_info));
       return;
     }
 
     // Check in the cache for frame.
-    std::optional<V1Frame> cached_frame = GetCachedFrame(frame_type);
-    if (cached_frame.has_value()) {
-      callback(std::move(cached_frame));
-      return;
-    }
-
-    ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout};
-    read_frame_info_queue_.push(std::move(read_fame_info));
+    cached_frame = PopCachedFrame(frame_type);
+  }
+  if (cached_frame) {
+    callback(*cached_frame);
+    return;
+  }
+  {
+    // No matching cached frame, queue this request, then read more frames.
+    absl::MutexLock lock(&mutex_);
+    ReadFrameInfo read_frame_info{frame_type, std::move(callback), timeout};
+    read_frame_info_queue_.push(std::move(read_frame_info));
 
+    if (timeout != absl::ZeroDuration()) {
     timeout_timer_ = std::make_unique<ThreadTimer>(
         service_thread_, "frame_reader_timeout", timeout,
         [reader = GetWeakPtr()]() {
           auto frame_reader = reader.lock();
           if (frame_reader == nullptr) {
-            NL_LOG(WARNING) << "IncomingFramesReader is released before.";
+            NL_LOG(WARNING) << "IncomingFramesReader has already been released "
+                               "before read timeout.";
             return;
           }
           frame_reader->OnTimeout();
         });
+    }
   }
   ReadNextFrame();
 }
@@ -133,7 +129,12 @@ void IncomingFramesReader::ReadNextFrame() {
           NL_LOG(WARNING) << "IncomingFramesReader is released before.";
           return;
         }
-        frame_reader->OnDataReadFromConnection(std::move(bytes));
+        if (!bytes.has_value()) {
+          NL_LOG(WARNING) << __func__ << ": Failed to read frame";
+          frame_reader->CloseAllPendingReads();
+          return;
+        }
+        frame_reader->OnDataReadFromConnection(*bytes);
       });
 }
 
@@ -143,64 +144,43 @@ void IncomingFramesReader::OnTimeout() {
 }
 
 void IncomingFramesReader::OnDataReadFromConnection(
-    std::optional<std::vector<uint8_t>> bytes) {
-  if (!bytes.has_value()) {
-    NL_LOG(WARNING) << __func__ << ": Failed to read frame";
-    CloseAllPendingReads();
-    return;
-  }
-
-  std::unique_ptr<Frame> frame =
-      DecodeFrame(absl::MakeSpan(bytes->data(), bytes->size()));
+    const std::vector<uint8_t>& bytes) {
+  std::unique_ptr<V1Frame> frame = DecodeV1Frame(bytes);
   if (frame == nullptr) {
     NL_LOG(WARNING)
         << __func__
         << ": Cannot decode frame. Not currently bound to nearby process";
-    CloseAllPendingReads();
+    ReadNextFrame();
     return;
   }
-
-  const V1Frame* v1_frame = OnFrameDecoded(*frame);
-  if (v1_frame != nullptr) {
-    Done(*v1_frame);
-  }
-}
-
-const V1Frame* IncomingFramesReader::OnFrameDecoded(const Frame& frame) {
-  if (frame.version() != Frame::V1) {
-    NL_VLOG(1) << __func__ << ": Frame read does not have V1Frame";
-    ReadNextFrame();
-    return nullptr;
-  }
-  auto v1_frame = frame.v1();
-  FrameType v1_frame_type = v1_frame.type();
+  FrameType frame_type = frame->type();
   bool cached_frame = false;
   {
-    MutexLock lock(&mutex_);
+    absl::MutexLock lock(&mutex_);
     if (read_frame_info_queue_.empty()) {
-      return nullptr;
+      return;
     }
     const ReadFrameInfo& frame_info = read_frame_info_queue_.front();
     if (frame_info.frame_type.has_value() &&
-        *frame_info.frame_type != v1_frame_type) {
+        *frame_info.frame_type != frame_type) {
       NL_LOG(WARNING) << __func__ << ": Failed to read frame of type "
                       << *frame_info.frame_type << ", but got frame of type "
-                      << v1_frame_type << ". Cached for later.";
-      cached_frames_.insert({v1_frame_type, v1_frame});
+                      << frame_type << ". Cached for later.";
+      cached_frames_.push_back(std::move(frame));
       cached_frame = true;
     }
   }
   if (cached_frame) {
     ReadNextFrame();
-    return nullptr;
+    return;
   }
-  return &frame.v1();
+  Done(std::move(frame));
 }
 
 void IncomingFramesReader::CloseAllPendingReads() {
   std::queue<ReadFrameInfo> queue;
   {
-    MutexLock lock(&mutex_);
+    absl::MutexLock lock(&mutex_);
     queue.swap(read_frame_info_queue_);
   }
   while (!queue.empty()) {
@@ -210,18 +190,18 @@ void IncomingFramesReader::CloseAllPendingReads() {
   }
 }
 
-void IncomingFramesReader::Done(const V1Frame& frame) {
+void IncomingFramesReader::Done(std::unique_ptr<V1Frame> frame) {
   ReadFrameInfo read_frame_info;
   {
-    MutexLock lock(&mutex_);
+    absl::MutexLock lock(&mutex_);
     timeout_timer_.reset();
     read_frame_info = std::move(read_frame_info_queue_.front());
     read_frame_info_queue_.pop();
   }
-  read_frame_info.callback(std::move(frame));
+  read_frame_info.callback(*frame);
 
   {
-    MutexLock lock(&mutex_);
+    absl::MutexLock lock(&mutex_);
     if (read_frame_info_queue_.empty()) {
       return;
     }
@@ -229,28 +209,36 @@ void IncomingFramesReader::Done(const V1Frame& frame) {
     read_frame_info_queue_.pop();
   }
 
-  if (read_frame_info.timeout.has_value()) {
+  if (read_frame_info.timeout != absl::ZeroDuration()) {
     ReadFrame(*read_frame_info.frame_type,
-              std::move(read_frame_info.callback), *read_frame_info.timeout);
+              std::move(read_frame_info.callback), read_frame_info.timeout);
   } else {
     ReadFrame(std::move(read_frame_info.callback));
   }
 }
 
-std::optional<V1Frame> IncomingFramesReader::GetCachedFrame(
-    std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
-        frame_type) {
+std::unique_ptr<V1Frame> IncomingFramesReader::PopCachedFrame(
+    std::optional<V1Frame::FrameType> frame_type) {
   NL_VLOG(1) << __func__ << ": Fetching cached frame";
-  if (frame_type.has_value())
-    NL_VLOG(1) << __func__ << ": Requested frame type - " << *frame_type;
+  if (cached_frames_.empty()) {
+    return nullptr;
+  }
+  if (!frame_type.has_value()) {
+    std::unique_ptr<V1Frame> frame = std::move(cached_frames_.front());
+    cached_frames_.pop_front();
+    return frame;
+  }
+  NL_VLOG(1) << __func__ << ": Requested frame type - " << *frame_type;
 
-  auto iter = frame_type.has_value() ? cached_frames_.find(*frame_type)
-                                     : cached_frames_.begin();
-
-  if (iter == cached_frames_.end()) return std::nullopt;
+  auto iter =
+      std::find_if(cached_frames_.begin(), cached_frames_.end(),
+                   [frame_type = *frame_type](std::unique_ptr<V1Frame>& frame) {
+                     return frame->type() == frame_type;
+                   });
 
+  if (iter == cached_frames_.end()) return nullptr;
   NL_VLOG(1) << __func__ << ": Successfully read cached frame";
-  std::optional<V1Frame> frame = std::move(iter->second);
+  std::unique_ptr<V1Frame> frame = std::move(*iter);
   cached_frames_.erase(iter);
   return frame;
 }
diff --git a/sharing/incoming_frames_reader.h b/sharing/incoming_frames_reader.h
index dce4dd8d..41e10640 100644
--- a/sharing/incoming_frames_reader.h
+++ b/sharing/incoming_frames_reader.h
@@ -18,7 +18,7 @@
 #include <stdint.h>
 
 #include <functional>
-#include <map>
+#include <list>
 #include <memory>
 #include <optional>
 #include <queue>
@@ -26,7 +26,7 @@
 
 #include "absl/base/thread_annotations.h"
 #include "absl/time/time.h"
-#include "internal/platform/mutex.h"
+#include "absl/synchronization/mutex.h"
 #include "internal/platform/task_runner.h"
 #include "sharing/nearby_connection.h"
 #include "sharing/proto/wire_format.pb.h"
@@ -63,7 +63,7 @@ class IncomingFramesReader
   // Note: Callers are expected wait for |callback| to be run before scheduling
   // subsequent calls to ReadFrame(..).
   virtual void ReadFrame(
-      nearby::sharing::service::proto::V1Frame_FrameType frame_type,
+      nearby::sharing::service::proto::V1Frame::FrameType frame_type,
       std::function<
           void(std::optional<nearby::sharing::service::proto::V1Frame>)>
           callback,
@@ -75,36 +75,39 @@ class IncomingFramesReader
 
  private:
   struct ReadFrameInfo {
-    std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
+    std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
         frame_type = std::nullopt;
     std::function<void(std::optional<nearby::sharing::service::proto::V1Frame>)>
         callback = nullptr;
-    std::optional<absl::Duration> timeout = std::nullopt;
+    absl::Duration timeout = absl::ZeroDuration();
   };
 
+  void ProcessReadRequest(
+      std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
+          frame_type,
+      std::function<
+          void(std::optional<nearby::sharing::service::proto::V1Frame>)>
+          callback,
+      absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_);
   void CloseAllPendingReads() ABSL_LOCKS_EXCLUDED(mutex_);
   void ReadNextFrame() ABSL_LOCKS_EXCLUDED(mutex_);
-  void OnDataReadFromConnection(std::optional<std::vector<uint8_t>> bytes)
-      ABSL_LOCKS_EXCLUDED(mutex_);
-  const nearby::sharing::service::proto::V1Frame* OnFrameDecoded(
-      const nearby::sharing::service::proto::Frame& frame)
+  void OnDataReadFromConnection(const std::vector<uint8_t>& bytes)
       ABSL_LOCKS_EXCLUDED(mutex_);
   void OnTimeout();
-  void Done(const nearby::sharing::service::proto::V1Frame& frame)
+  void Done(std::unique_ptr<nearby::sharing::service::proto::V1Frame> frame)
       ABSL_LOCKS_EXCLUDED(mutex_);
-  std::optional<nearby::sharing::service::proto::V1Frame> GetCachedFrame(
-      std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
+  std::unique_ptr<nearby::sharing::service::proto::V1Frame> PopCachedFrame(
+      std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
           frame_type) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   TaskRunner& service_thread_;
   NearbyConnection* const connection_;
 
-  RecursiveMutex mutex_;
+  absl::Mutex mutex_;
   std::queue<ReadFrameInfo> read_frame_info_queue_ ABSL_GUARDED_BY(mutex_);
 
   // Caches frames read from NearbyConnection which are not used immediately.
-  std::map<nearby::sharing::service::proto::V1Frame_FrameType,
-           std::optional<nearby::sharing::service::proto::V1Frame>>
+  std::list<std::unique_ptr<nearby::sharing::service::proto::V1Frame>>
       cached_frames_ ABSL_GUARDED_BY(mutex_);
 
   std::unique_ptr<ThreadTimer> timeout_timer_ ABSL_GUARDED_BY(mutex_);
diff --git a/sharing/incoming_frames_reader_test.cc b/sharing/incoming_frames_reader_test.cc
index 950e8705..26c7bae0 100644
--- a/sharing/incoming_frames_reader_test.cc
+++ b/sharing/incoming_frames_reader_test.cc
@@ -35,6 +35,7 @@ namespace nearby {
 namespace sharing {
 namespace {
 
+using ::nearby::sharing::service::proto::ConnectionResponseFrame;
 using ::nearby::sharing::service::proto::V1Frame;
 
 constexpr absl::Duration kTimeout = absl::Seconds(1);
@@ -48,7 +49,7 @@ std::optional<std::vector<uint8_t>> GetIntroductionFrame() {
   v1frame->mutable_introduction();
 
   std::vector<uint8_t> data;
-  data.resize(frame.ByteSize());
+  data.resize(frame.ByteSizeLong());
   if (frame.SerializeToArray(data.data(), data.size())) {
     return data;
   }
@@ -64,7 +65,25 @@ std::optional<std::vector<uint8_t>> GetCancelFrame() {
   v1frame->set_type(service::proto::V1Frame::CANCEL);
 
   std::vector<uint8_t> data;
-  data.resize(frame.ByteSize());
+  data.resize(frame.ByteSizeLong());
+  if (frame.SerializeToArray(data.data(), data.size())) {
+    return data;
+  }
+
+  return std::nullopt;
+}
+
+std::optional<std::vector<uint8_t>> GetResponseFrame() {
+  nearby::sharing::service::proto::Frame frame =
+      nearby::sharing::service::proto::Frame();
+  frame.set_version(nearby::sharing::service::proto::Frame::V1);
+  V1Frame* v1frame = frame.mutable_v1();
+  v1frame->set_type(service::proto::V1Frame::RESPONSE);
+  v1frame->mutable_connection_response()->set_status(
+      ConnectionResponseFrame::ACCEPT);
+
+  std::vector<uint8_t> data;
+  data.resize(frame.ByteSizeLong());
   if (frame.SerializeToArray(data.data(), data.size())) {
     return data;
   }
@@ -137,6 +156,29 @@ TEST_F(IncomingFramesReaderTest, ReadTimedOut) {
   EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
 }
 
+TEST_F(IncomingFramesReaderTest, ReadNonV1FrameSkipped) {
+  nearby::sharing::service::proto::Frame frame =
+      nearby::sharing::service::proto::Frame();
+  V1Frame* v1frame = frame.mutable_v1();
+  v1frame->set_type(service::proto::V1Frame::CANCEL);
+  std::vector<uint8_t> data;
+  data.resize(frame.ByteSizeLong());
+  ASSERT_GT(data.size(), 0);
+  ASSERT_TRUE(frame.SerializeToArray(data.data(), data.size()));
+  connection().WriteMessage(data);
+  std::optional<std::vector<uint8_t>> introduction_frame =
+      GetIntroductionFrame();
+  ASSERT_TRUE(introduction_frame.has_value());
+  connection().WriteMessage(*introduction_frame);
+
+  absl::Notification notification;
+  frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
+    EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION);
+    notification.Notify();
+  });
+  EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
+}
+
 TEST_F(IncomingFramesReaderTest, ReadAnyFrameSuccessful) {
   std::optional<std::vector<uint8_t>> introduction_frame =
       GetIntroductionFrame();
@@ -193,6 +235,9 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
   std::optional<std::vector<uint8_t>> cancel_frame = GetCancelFrame();
   ASSERT_TRUE(cancel_frame.has_value());
   connection().WriteMessage(*cancel_frame);
+  std::optional<std::vector<uint8_t>> response_frame = GetResponseFrame();
+  ASSERT_TRUE(response_frame.has_value());
+  connection().WriteMessage(*response_frame);
 
   std::optional<std::vector<uint8_t>> introduction_frame =
       GetIntroductionFrame();
@@ -209,7 +254,7 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
       kTimeout);
 
   EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
-  // Reading any frame should return CancelFrame.
+  // Reading any frame should return cancel frame, then response frame.
   absl::Notification cancel_notification;
   frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
     ASSERT_NE(frame, std::nullopt);
@@ -217,6 +262,13 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
     cancel_notification.Notify();
   });
   EXPECT_TRUE(cancel_notification.WaitForNotificationWithTimeout(kTimeout));
+  absl::Notification response_notification;
+  frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
+    ASSERT_NE(frame, std::nullopt);
+    EXPECT_EQ(frame->type(), service::proto::V1Frame::RESPONSE);
+    response_notification.Notify();
+  });
+  EXPECT_TRUE(response_notification.WaitForNotificationWithTimeout(kTimeout));
 }
 
 TEST_F(IncomingFramesReaderTest, ReadAfterConnectionClosed) {
@@ -298,19 +350,23 @@ TEST_F(IncomingFramesReaderTest, ReleaseFrameReaderDuringRead) {
   EXPECT_EQ(frames_reader(), nullptr);
 }
 
-TEST_F(IncomingFramesReaderTest, ReadInvalidFrame) {
+TEST_F(IncomingFramesReaderTest, SkipInvalidFrame) {
   absl::Notification notification;
   frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
-    EXPECT_EQ(frame, std::nullopt);
+    EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL);
     notification.Notify();
   });
 
   std::optional<std::vector<uint8_t>> invalid_frame = GetInvalidFrame();
   ASSERT_TRUE(invalid_frame.has_value());
   connection().WriteMessage(*invalid_frame);
+  std::optional<std::vector<uint8_t>> cancel_frame = GetCancelFrame();
+  ASSERT_TRUE(cancel_frame.has_value());
+  connection().WriteMessage(*cancel_frame);
 
   Sync();
   EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
+  ReleaseFrameReader();
 }
 
 }  // namespace