[Nearby] Reset URL loader and stream parser before invoking callback

We make a few simple changes to help diagnose the crash occurring in
https://crbug.com/1162157:

1) Reset the URL loader and stream parser before invoking the callback
   to ensure we stop streaming before any callback side effects occur.
2) Add a CHECK to OnDataReceived() to ensure the stream parser is not
   null. Will change to DCHECK when the crash is resolved.
3) In the stream parser, use the native StringPiece methods to get
   the underlying data and string length.
4) Add additional VERBOSE logging to ReceiveMessageExpress.

I'm skeptical that these changes will fix the crash, but they will
help eliminate some possibilities.

Bug: 1162157
Change-Id: Ic1eef5c4bf0d047c66420aad29cbd3d93992ba05
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2627564
Commit-Queue: Alex Gough <ajgo@chromium.org>
Reviewed-by: Alex Gough <ajgo@chromium.org>
Reviewed-by: James Vecore <vecore@google.com>
Auto-Submit: Josh Nohle <nohle@chromium.org>
Cr-Commit-Position: refs/heads/master@{#843119}

chrome/browser/nearby_sharing/common/nearby_share_http_result.cc

@@ -68,6 +68,9 @@ NearbyShareHttpStatus::NearbyShareHttpStatus(
   }
 }
 
+NearbyShareHttpStatus::NearbyShareHttpStatus(
+    const NearbyShareHttpStatus& status) = default;
+
 NearbyShareHttpStatus::~NearbyShareHttpStatus() = default;
 
 bool NearbyShareHttpStatus::IsSuccess() const {

chrome/browser/nearby_sharing/common/nearby_share_http_result.h

@@ -54,6 +54,7 @@ class NearbyShareHttpStatus {
  public:
   NearbyShareHttpStatus(const int net_error,
                         const network::mojom::URLResponseHead* head);
+  NearbyShareHttpStatus(const NearbyShareHttpStatus& status);
   ~NearbyShareHttpStatus();
 
   bool IsSuccess() const;

chrome/browser/nearby_sharing/instantmessaging/receive_messages_express.cc

@@ -54,21 +54,29 @@ const net::NetworkTrafficAnnotationTag kTrafficAnnotation =
             }
           })");
 
-void LogReceiveResult(bool success, const network::SimpleURLLoader* loader) {
+base::Optional<NearbyShareHttpStatus> HttpStatusFromUrlLoader(
+    const network::SimpleURLLoader* loader) {
+  if (!loader)
+    return base::nullopt;
+
+  return NearbyShareHttpStatus(loader->NetError(), loader->ResponseInfo());
+}
+
+void LogReceiveResult(
+    bool success,
+    const base::Optional<NearbyShareHttpStatus>& http_status) {
   std::stringstream ss;
   ss << "Instant messaging receive express "
      << (success ? "succeeded." : "failed.");
   base::UmaHistogramBoolean(
       "Nearby.Connections.InstantMessaging.ReceiveExpress.Result", success);
-  if (loader) {
-    NearbyShareHttpStatus http_status(loader->NetError(),
-                                      loader->ResponseInfo());
-    ss << " HTTP status: " << http_status;
+  if (http_status) {
+    ss << " HTTP status: " << *http_status;
     if (!success) {
       base::UmaHistogramSparse(
           "Nearby.Connections.InstantMessaging.ReceiveExpress.Result."
           "FailureReason",
-          http_status.GetResultCodeForMetrics());
+          http_status->GetResultCodeForMetrics());
     }
   }
 
@@ -101,26 +109,30 @@ void ReceiveMessagesExpress::StartReceivingMessages(
     base::RepeatingCallback<void(const std::string& message)> listener,
     SuccessCallback callback) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  url_loader_.reset();
+  stream_parser_.reset();
+
   if (!success_callback_.is_null()) {
-    // A pending callback was found, this should be marked as false since the
+    // A pending callback was found. This should be marked as false since the
     // previous receiver has not yet started listening for messages and is now
     // going to be replaced by the new listener.
+    NS_LOG(WARNING)
+        << __func__
+        << ": Found pending request to start receiving messages. "
+           "Failing the previous request before handling the new request.";
     std::move(success_callback_).Run(false);
   }
   success_callback_ = std::move(callback);
-  url_loader_.reset();
-  stream_parser_ = std::make_unique<StreamParser>(
-      listener, base::BindOnce(&ReceiveMessagesExpress::OnFastPathReady,
-                               base::Unretained(this)));
 
   token_fetcher_->GetAccessToken(
       base::BindOnce(&ReceiveMessagesExpress::DoStartReceivingMessages,
-                     weak_ptr_factory_.GetWeakPtr(), request));
+                     weak_ptr_factory_.GetWeakPtr(), request, listener));
 }
 
 void ReceiveMessagesExpress::DoStartReceivingMessages(
     const chrome_browser_nearby_sharing_instantmessaging::
         ReceiveMessagesExpressRequest& request,
+    base::RepeatingCallback<void(const std::string& message)> listener,
     const std::string& oauth_token) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   base::UmaHistogramBoolean(
@@ -133,6 +145,13 @@ void ReceiveMessagesExpress::DoStartReceivingMessages(
     return;
   }
 
+  NS_LOG(VERBOSE) << __func__
+                  << ": OAuth token fetched; starting stream download";
+
+  stream_parser_ = std::make_unique<StreamParser>(
+      listener, base::BindOnce(&ReceiveMessagesExpress::OnFastPathReady,
+                               base::Unretained(this)));
+
   auto resource_request = std::make_unique<network::ResourceRequest>();
   resource_request->url = GURL(kInstantMessagingReceiveMessageAPI);
   resource_request->load_flags =
@@ -152,29 +171,42 @@ void ReceiveMessagesExpress::DoStartReceivingMessages(
 
 void ReceiveMessagesExpress::StopReceivingMessages() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  if (!success_callback_.is_null())
-    std::move(success_callback_).Run(false);
-
   url_loader_.reset();
   stream_parser_.reset();
+
+  NS_LOG(VERBOSE) << __func__ << ": callback already invoked? "
+                  << (success_callback_ ? "no" : "yes");
+
+  if (!success_callback_.is_null())
+    std::move(success_callback_).Run(false);
 }
 
 void ReceiveMessagesExpress::OnDataReceived(base::StringPiece data,
                                             base::OnceClosure resume) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+
+  // TODO(https://crbug.com/1162157): Change to DCHECK when bug is resolved.
+  CHECK(stream_parser_) << __func__ << ": Stream parser unexpectedly null";
+
   stream_parser_->Append(data);
   std::move(resume).Run();
 }
 
 void ReceiveMessagesExpress::OnComplete(bool success) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  base::Optional<NearbyShareHttpStatus> http_status =
+      HttpStatusFromUrlLoader(url_loader_.get());
+  url_loader_.reset();
+  stream_parser_.reset();
+
+  NS_LOG(VERBOSE) << __func__ << ": " << (success ? "success" : "failure")
+                  << ", callback already invoked? "
+                  << (success_callback_ ? "no" : "yes");
+
   if (!success_callback_.is_null()) {
-    LogReceiveResult(success, url_loader_.get());
+    LogReceiveResult(success, http_status);
     std::move(success_callback_).Run(success);
   }
-
-  url_loader_.reset();
-  stream_parser_.reset();
 }
 
 void ReceiveMessagesExpress::OnRetry(base::OnceClosure start_retry) {
@@ -183,7 +215,7 @@ void ReceiveMessagesExpress::OnRetry(base::OnceClosure start_retry) {
 
 void ReceiveMessagesExpress::OnFastPathReady() {
   if (!success_callback_.is_null()) {
-    LogReceiveResult(/*success=*/true, /*loader=*/nullptr);
+    LogReceiveResult(/*success=*/true, /*http_status=*/base::nullopt);
     std::move(success_callback_).Run(true);
   }
 }

chrome/browser/nearby_sharing/instantmessaging/receive_messages_express.h

@@ -55,6 +55,7 @@ class ReceiveMessagesExpress : public network::SimpleURLLoaderStreamConsumer {
   void DoStartReceivingMessages(
       const chrome_browser_nearby_sharing_instantmessaging::
           ReceiveMessagesExpressRequest& request,
+      base::RepeatingCallback<void(const std::string& message)> listener,
       const std::string& oauth_token);
 
   // network::SimpleURLLoaderStreamConsumer:

chrome/browser/nearby_sharing/instantmessaging/stream_parser.cc

@@ -15,8 +15,7 @@ StreamParser::StreamParser(
 StreamParser::~StreamParser() = default;
 
 void StreamParser::Append(base::StringPiece data) {
-  size_t size = data.as_string().size();
-  data_.append(data.as_string().data(), size);
+  data_.append(data.data(), data.size());
 
   base::Optional<chrome_browser_nearby_sharing_instantmessaging::StreamBody>
       stream_body = GetNextMessage();