Converting AdsServiceImpl to a Fake

- Allow Resources to be set
- Allow clients to subscribe/unsubscribe to resources
- Managing, storing, and versioning resources correctly based on updates
- Correct ACK/NACK according to XDS protocol
- Started a separate blocking thread to be used for stream read
- Handling both requests from clients and updates from tests and respond
with the correct version of resource
- Consolidated all 4 resources (LDS, CDS, EDS, RDS) to share common code
- Updated all tests to use the fake properly
- Fixed potential crash cases in the subchannel thread for clean TSAN,
UBSAN runs

src/core/ext/filters/client_channel/xds/xds_client.cc

@@ -1524,8 +1524,11 @@ void XdsClient::ChannelState::LrsCallState::MaybeStartReportingLocked() {
   // Don't start if the ADS call hasn't received any valid response. Note that
   // this must be the first channel because it is the current channel but its
   // ADS call hasn't seen any response.
-  AdsCallState* ads_calld = chand()->ads_calld_->calld();
-  if (ads_calld == nullptr || !ads_calld->seen_response()) return;
+  if (chand()->ads_calld_ == nullptr ||
+      chand()->ads_calld_->calld() == nullptr ||
+      !chand()->ads_calld_->calld()->seen_response()) {
+    return;
+  }
   // Start reporting.
   reporter_ = MakeOrphanable<Reporter>(
       Ref(DEBUG_LOCATION, "LRS+load_report+start"), load_reporting_interval_);

src/core/lib/gprpp/sync.h

@@ -117,6 +117,15 @@ class CondVar {
     }
   }
 
+  // Returns true iff we timed-out
+  template <typename Predicate>
+  bool WaitUntil(Mutex* mu, Predicate pred, const gpr_timespec& deadline) {
+    while (!pred()) {
+      if (Wait(mu, deadline)) return true;
+    }
+    return false;
+  }
+
  private:
   gpr_cv cv_;
 };