Revert

Author: var-const

Firestore/core/src/firebase/firestore/auth/firebase_credentials_provider_apple.mm

@@ -120,26 +120,19 @@
     }
   };
 
-  bool force_refresh = false;
-  {
-    std::lock_guard<std::mutex> lock{contents_->mutex};
-    force_refresh = contents_->force_refresh;
-    contents_->force_refresh = false;
-  }
-
   // TODO(wilhuff): Need a better abstraction over a missing auth provider.
   if (contents_->auth) {
-    [contents_->auth getTokenForcingRefresh:force_refresh
+    [contents_->auth getTokenForcingRefresh:contents_->force_refresh
                                withCallback:get_token_callback];
   } else {
     // If there's no Auth provider, call back immediately with a nil
     // (unauthenticated) token.
     get_token_callback(nil, nil);
   }
+  contents_->force_refresh = false;
 }
 
 void FirebaseCredentialsProvider::InvalidateToken() {
-  std::lock_guard<std::mutex> lock{contents_->mutex};
   contents_->force_refresh = true;
 }
 

Firestore/core/src/firebase/firestore/remote/grpc_connection.mm

@@ -227,11 +227,6 @@ bool HasSpecialConfig(const std::string& host) {
           call->FinishAndNotify(Status{FirestoreErrorCode::Unavailable,
                                        "Network connectivity changed"});
         }
-        // The old channel may hang for a long time trying to reestablish
-        // connection before eventually failing. Note that gRPC Objective-C
-        // client does the same thing:
-        // https://github.com/grpc/grpc/blob/master/src/objective-c/GRPCClient/private/GRPCHost.m#L309-L314
-        grpc_channel_.reset();
       });
 }
 

Firestore/core/src/firebase/firestore/remote/grpc_stream.cc

@@ -175,20 +175,27 @@ void GrpcStream::FinishAndNotify(const Status& status) {
 void GrpcStream::Shutdown() {
   MaybeUnregister();
   if (completions_.empty()) {
-    // Nothing to cancel -- either the call was already finished, or it has
-    // never been started.
+    // Nothing to cancel.
     return;
   }
 
   // Important: since the stream always has a pending read operation,
   // cancellation has to be called, or else the read would hang forever, and
   // finish operation will never get completed.
+  //
   // (on the other hand, when an operation fails, cancellation should not be
   // called, otherwise the real failure cause will be overwritten by status
   // "canceled".)
   context_->TryCancel();
-  FinishCall({});
-  // Wait until "finish" is off the queue.
+
+  // The observer is not interested in this event -- since it initiated the
+  // finish operation, the observer must know the reason.
+  GrpcCompletion* completion = NewCompletion(Type::Finish, {});
+  // TODO(varconst): is issuing a finish operation necessary in this case? We
+  // don't care about the status, but perhaps it will make the server notice
+  // client disconnecting sooner?
+  call_->Finish(completion->status(), completion);
+
   FastFinishCompletionsBlocking();
 }
 
@@ -199,14 +206,6 @@ void GrpcStream::MaybeUnregister() {
   }
 }
 
-void GrpcStream::FinishCall(const OnSuccess& callback) {
-  // All completions issued by this call must be taken off the queue before
-  // finish operation can be enqueued.
-  FastFinishCompletionsBlocking();
-  GrpcCompletion* completion = NewCompletion(Type::Finish, callback);
-  call_->Finish(completion->status(), completion);
-}
-
 void GrpcStream::FastFinishCompletionsBlocking() {
   // TODO(varconst): reset buffered_writer_? Should not be necessary, because it
   // should never be called again after a call to Finish.
@@ -277,10 +276,29 @@ void GrpcStream::OnWrite() {
 }
 
 void GrpcStream::OnOperationFailed() {
-  FinishCall([this](const GrpcCompletion* completion) {
-        Status status = ConvertStatus(*completion->status());
-        FinishAndNotify(status);
-      });
+  if (is_finishing_) {
+    // `Finish` itself cannot fail. If another failed operation already
+    // triggered `Finish`, there's nothing to do.
+    return;
+  }
+
+  is_finishing_ = true;
+
+  if (observer_) {
+    GrpcCompletion* completion =
+        NewCompletion(Type::Finish, [this](const GrpcCompletion* completion) {
+          OnFinishedByServer(*completion->status());
+        });
+    call_->Finish(completion->status(), completion);
+  } else {
+    // The only reason to finish would be to get the status; if the observer is
+    // no longer interested, there is no need to do that.
+    Shutdown();
+  }
+}
+
+void GrpcStream::OnFinishedByServer(const grpc::Status& status) {
+  FinishAndNotify(ConvertStatus(status));
 }
 
 void GrpcStream::RemoveCompletion(const GrpcCompletion* to_remove) {
@@ -297,9 +315,7 @@ GrpcCompletion* GrpcStream::NewCompletion(Type tag,
         RemoveCompletion(completion);
 
         if (ok) {
-          if (on_success) {
-            on_success(completion);
-          }
+          on_success(completion);
         } else {
           // Use the same error-handling for all operations; all errors are
           // unrecoverable.

Firestore/core/src/firebase/firestore/remote/grpc_stream.h

@@ -191,12 +191,12 @@ class GrpcStream : public GrpcCall {
   void OnRead(const grpc::ByteBuffer& message);
   void OnWrite();
   void OnOperationFailed();
+  void OnFinishedByServer(const grpc::Status& status);
   void RemoveCompletion(const GrpcCompletion* to_remove);
 
   using OnSuccess = std::function<void(const GrpcCompletion*)>;
   GrpcCompletion* NewCompletion(GrpcCompletion::Type type,
                                 const OnSuccess& callback);
-  void FinishCall(const OnSuccess& callback);
 
   // Blocks until all the completions issued by this stream come out from the
   // gRPC completion queue. Once they do, it is safe to delete this `GrpcStream`
@@ -227,6 +227,8 @@ class GrpcStream : public GrpcCall {
   internal::BufferedWriter buffered_writer_;
 
   std::vector<GrpcCompletion*> completions_;
+
+  bool is_finishing_ = false;
 };
 
 }  // namespace remote