diff --git a/Firestore/Example/Tests/Integration/API/FIRDatabaseTests.mm b/Firestore/Example/Tests/Integration/API/FIRDatabaseTests.mm
index fb2b9af77ec..53bb196c19e 100644
--- a/Firestore/Example/Tests/Integration/API/FIRDatabaseTests.mm
+++ b/Firestore/Example/Tests/Integration/API/FIRDatabaseTests.mm
@@ -531,6 +531,42 @@ - (void)testAddingToACollectionYieldsTheCorrectDocumentReference {
   [self awaitExpectations];
 }
 
+- (void)testSnapshotsInSyncListenerFiresAfterListenersInSync {
+  FIRCollectionReference *coll = [self.db collectionWithPath:@"collection"];
+  FIRDocumentReference *ref = [coll addDocumentWithData:@{@"foo" : @1}];
+  NSMutableArray<NSString *> *events = [NSMutableArray array];
+
+  XCTestExpectation *gotInitialSnapshot = [self expectationWithDescription:@"gotInitialSnapshot"];
+  __block bool setupComplete = false;
+  [ref addSnapshotListener:^(FIRDocumentSnapshot *snapshot, NSError *error) {
+    XCTAssertNil(error);
+    [events addObject:@"doc"];
+    // Wait for the initial event from the backend so that we know we'll get exactly one snapshot
+    // event for our local write below.
+    if (!setupComplete) {
+      setupComplete = true;
+      [gotInitialSnapshot fulfill];
+    }
+  }];
+
+  [self awaitExpectations];
+  [events removeAllObjects];
+
+  XCTestExpectation *done = [self expectationWithDescription:@"SnapshotsInSyncListenerDone"];
+  [ref.firestore addSnapshotsInSyncListener:^() {
+    [events addObject:@"snapshots-in-sync"];
+    if ([events count] == 3) {
+      // We should have an initial snapshots-in-sync event, then a snapshot event
+      // for set(), then another event to indicate we're in sync again.
+      NSArray<NSString *> *expected = @[ @"snapshots-in-sync", @"doc", @"snapshots-in-sync" ];
+      XCTAssertEqualObjects(events, expected);
+      [done fulfill];
+    }
+  }];
+
+  [self writeDocumentRef:ref data:@{@"foo" : @3}];
+}
+
 - (void)testListenCanBeCalledMultipleTimes {
   FIRCollectionReference *coll = [self.db collectionWithPath:@"collection"];
   FIRDocumentReference *doc = [coll documentWithAutoID];
diff --git a/Firestore/Example/Tests/SpecTests/FSTSpecTests.mm b/Firestore/Example/Tests/SpecTests/FSTSpecTests.mm
index afc59d39e06..0a2a1f5c44b 100644
--- a/Firestore/Example/Tests/SpecTests/FSTSpecTests.mm
+++ b/Firestore/Example/Tests/SpecTests/FSTSpecTests.mm
@@ -278,6 +278,14 @@ - (void)doDelete:(NSString *)key {
   [self.driver writeUserMutation:FSTTestDeleteMutation(key)];
 }
 
+- (void)doAddSnapshotsInSyncListener {
+  [self.driver addSnapshotsInSyncListener];
+}
+
+- (void)doRemoveSnapshotsInSyncListener {
+  [self.driver removeSnapshotsInSyncListener];
+}
+
 - (void)doWatchAck:(NSArray<NSNumber *> *)ackedTargets {
   WatchTargetChange change{WatchTargetChangeState::Added, ConvertTargetsArray(ackedTargets)};
   [self.driver receiveWatchChange:change snapshotVersion:SnapshotVersion::None()];
@@ -479,6 +487,10 @@ - (void)doStep:(NSDictionary *)step {
     [self doPatch:step[@"userPatch"]];
   } else if (step[@"userDelete"]) {
     [self doDelete:step[@"userDelete"]];
+  } else if (step[@"addSnapshotsInSyncListener"]) {
+    [self doAddSnapshotsInSyncListener];
+  } else if (step[@"removeSnapshotsInSyncListener"]) {
+    [self doRemoveSnapshotsInSyncListener];
   } else if (step[@"drainQueue"]) {
     [self doDrainQueue];
   } else if (step[@"watchAck"]) {
@@ -651,6 +663,11 @@ - (void)validateStateExpectations:(nullable NSDictionary *)expected {
   [self validateActiveTargets];
 }
 
+- (void)validateSnapshotsInSyncEvents:(int)expectedSnapshotInSyncEvents {
+  XCTAssertEqual(expectedSnapshotInSyncEvents, [self.driver snapshotsInSyncEvents]);
+  [self.driver resetSnapshotsInSyncEvents];
+}
+
 - (void)validateUserCallbacks:(nullable NSDictionary *)expected {
   NSDictionary *expectedCallbacks = expected[@"userCallbacks"];
   NSArray<NSString *> *actualAcknowledgedDocs =
@@ -728,6 +745,8 @@ - (void)runSpecTestSteps:(NSArray *)steps config:(NSDictionary *)config {
       [self doStep:step];
       [self validateStepExpectations:step[@"expect"]];
       [self validateStateExpectations:step[@"stateExpect"]];
+      int expectedSnapshotsInSyncEvents = [step[@"expectedSnapshotsInSyncEvents"] intValue];
+      [self validateSnapshotsInSyncEvents:expectedSnapshotsInSyncEvents];
     }
     [self.driver validateUsage];
   } @finally {
diff --git a/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h b/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h
index ec3dce79e19..eeb881b6ccb 100644
--- a/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h
+++ b/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h
@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "Firestore/core/src/firebase/firestore/auth/user.h"
+#include "Firestore/core/src/firebase/firestore/core/event_listener.h"
 #include "Firestore/core/src/firebase/firestore/core/query.h"
 #include "Firestore/core/src/firebase/firestore/core/view_snapshot.h"
 #include "Firestore/core/src/firebase/firestore/local/query_data.h"
@@ -32,6 +33,7 @@
 #include "Firestore/core/src/firebase/firestore/model/types.h"
 #include "Firestore/core/src/firebase/firestore/remote/watch_change.h"
 #include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 
 namespace firebase {
 namespace firestore {
@@ -311,6 +313,26 @@ typedef std::unordered_map<firebase::firestore::auth::User,
 /** The current user for the FSTSyncEngine; determines which mutation queue is active. */
 @property(nonatomic, assign, readonly) const firebase::firestore::auth::User &currentUser;
 
+/**
+ * The number of snapshots-in-sync events that have been received.
+ */
+@property(nonatomic, readonly) int snapshotsInSyncEvents;
+
+- (void)incrementSnapshotsInSyncEvents;
+
+- (void)resetSnapshotsInSyncEvents;
+
+/**
+ * Adds a snpahots-in-sync listener to the event manager and keeps track of it so that it
+ * can be easily removed later.
+ */
+- (void)addSnapshotsInSyncListener;
+
+/**
+ * Removes the snapshots-in-sync listener from the event manager.
+ */
+- (void)removeSnapshotsInSyncListener;
+
 /** The set of active targets as observed on the watch stream. */
 - (const std::unordered_map<firebase::firestore::model::TargetId, local::QueryData> &)activeTargets;
 
diff --git a/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm b/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm
index e26beb65194..98ff693f026 100644
--- a/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm
+++ b/Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.mm
@@ -56,6 +56,7 @@
 using firebase::firestore::auth::HashUser;
 using firebase::firestore::auth::User;
 using firebase::firestore::core::DatabaseInfo;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::core::EventManager;
 using firebase::firestore::core::ListenOptions;
 using firebase::firestore::core::Query;
@@ -78,6 +79,7 @@
 using firebase::firestore::remote::WatchChange;
 using firebase::firestore::util::AsyncQueue;
 using firebase::firestore::util::DelayedConstructor;
+using firebase::firestore::util::Empty;
 using firebase::firestore::util::ExecutorLibdispatch;
 using firebase::firestore::util::MakeNSError;
 using firebase::firestore::util::MakeNSString;
@@ -170,7 +172,10 @@ @implementation FSTSyncEngineTestDriver {
   DatabaseInfo _databaseInfo;
   User _currentUser;
 
+  std::vector<std::shared_ptr<EventListener<Empty>>> _snapshotsInSyncListeners;
   std::shared_ptr<MockDatastore> _datastore;
+
+  int _snapshotsInSyncEvents;
 }
 
 - (instancetype)initWithPersistence:(std::unique_ptr<Persistence>)persistence {
@@ -246,6 +251,34 @@ - (void)drainQueue {
   return _currentUser;
 }
 
+- (void)incrementSnapshotsInSyncEvents {
+  _snapshotsInSyncEvents += 1;
+}
+
+- (void)resetSnapshotsInSyncEvents {
+  _snapshotsInSyncEvents = 0;
+}
+
+- (void)addSnapshotsInSyncListener {
+  std::shared_ptr<EventListener<Empty>> eventListener = EventListener<Empty>::Create(
+      [self](const StatusOr<Empty> &) { [self incrementSnapshotsInSyncEvents]; });
+  _snapshotsInSyncListeners.push_back(eventListener);
+  _eventManager->AddSnapshotsInSyncListener(eventListener);
+}
+
+- (void)removeSnapshotsInSyncListener {
+  if (_snapshotsInSyncListeners.empty()) {
+    HARD_FAIL("There must be a listener to unlisten to");
+  } else {
+    _eventManager->RemoveSnapshotsInSyncListener(_snapshotsInSyncListeners.back());
+    _snapshotsInSyncListeners.pop_back();
+  }
+}
+
+- (int)snapshotsInSyncEvents {
+  return _snapshotsInSyncEvents;
+}
+
 - (void)start {
   _workerQueue->EnqueueBlocking([&] {
     _localStore->Start();
diff --git a/Firestore/Example/Tests/SpecTests/json/listen_spec_test.json b/Firestore/Example/Tests/SpecTests/json/listen_spec_test.json
index 680c45bb39f..00df5877114 100644
--- a/Firestore/Example/Tests/SpecTests/json/listen_spec_test.json
+++ b/Firestore/Example/Tests/SpecTests/json/listen_spec_test.json
@@ -10279,5 +10279,814 @@
         "clientIndex": 0
       }
     ]
+  },
+  "onSnapshotsInSync should not fire for doc changes if there are no listeners": {
+    "describeName": "Listens:",
+    "itName": "onSnapshotsInSync should not fire for doc changes if there are no listeners",
+    "tags": [],
+    "config": {
+      "useGarbageCollection": true,
+      "numClients": 1
+    },
+    "steps": [
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 2
+          }
+        ]
+      }
+    ]
+  },
+  "onSnapshotsInSync fires when called even if there are no local listeners": {
+    "describeName": "Listens:",
+    "itName": "onSnapshotsInSync fires when called even if there are no local listeners",
+    "tags": [],
+    "config": {
+      "useGarbageCollection": true,
+      "numClients": 1
+    },
+    "steps": [
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      }
+    ]
+  },
+  "onSnapshotsInSync fires for metadata changes": {
+    "describeName": "Listens:",
+    "itName": "onSnapshotsInSync fires for metadata changes",
+    "tags": [],
+    "config": {
+      "useGarbageCollection": true,
+      "numClients": 1
+    },
+    "steps": [
+      {
+        "userListen": [
+          2,
+          {
+            "path": "collection",
+            "filters": [],
+            "orderBys": []
+          }
+        ],
+        "stateExpect": {
+          "activeTargets": {
+            "2": {
+              "query": {
+                "path": "collection",
+                "filters": [],
+                "orderBys": []
+              },
+              "resumeToken": ""
+            }
+          }
+        }
+      },
+      {
+        "watchAck": [
+          2
+        ]
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 1000,
+              "value": {
+                "v": 1
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            2
+          ]
+        }
+      },
+      {
+        "watchCurrent": [
+          [
+            2
+          ],
+          "resume-token-1000"
+        ]
+      },
+      {
+        "watchSnapshot": {
+          "version": 1000,
+          "targetIds": []
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "added": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 1
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ]
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 2
+          }
+        ],
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 2000,
+              "value": {
+                "v": 2
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            2
+          ]
+        }
+      },
+      {
+        "watchSnapshot": {
+          "version": 2000,
+          "targetIds": []
+        }
+      },
+      {
+        "writeAck": {
+          "version": 2000
+        },
+        "stateExpect": {
+          "userCallbacks": {
+            "acknowledgedDocs": [
+              "collection/a"
+            ],
+            "rejectedDocs": []
+          }
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "metadata": [
+              {
+                "key": "collection/a",
+                "version": 2000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 1
+      }
+    ]
+  },
+  "onSnapshotsInSync fires once for multiple event snapshots": {
+    "describeName": "Listens:",
+    "itName": "onSnapshotsInSync fires once for multiple event snapshots",
+    "tags": [],
+    "config": {
+      "useGarbageCollection": true,
+      "numClients": 1
+    },
+    "steps": [
+      {
+        "userListen": [
+          2,
+          {
+            "path": "collection",
+            "filters": [],
+            "orderBys": []
+          }
+        ],
+        "stateExpect": {
+          "activeTargets": {
+            "2": {
+              "query": {
+                "path": "collection",
+                "filters": [],
+                "orderBys": []
+              },
+              "resumeToken": ""
+            }
+          }
+        }
+      },
+      {
+        "watchAck": [
+          2
+        ]
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 1000,
+              "value": {
+                "v": 1
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            2
+          ]
+        }
+      },
+      {
+        "watchCurrent": [
+          [
+            2
+          ],
+          "resume-token-1000"
+        ]
+      },
+      {
+        "watchSnapshot": {
+          "version": 1000,
+          "targetIds": []
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "added": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 1
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ]
+      },
+      {
+        "userListen": [
+          4,
+          {
+            "path": "collection/a",
+            "filters": [],
+            "orderBys": []
+          }
+        ],
+        "stateExpect": {
+          "activeTargets": {
+            "2": {
+              "query": {
+                "path": "collection",
+                "filters": [],
+                "orderBys": []
+              },
+              "resumeToken": ""
+            },
+            "4": {
+              "query": {
+                "path": "collection/a",
+                "filters": [],
+                "orderBys": []
+              },
+              "resumeToken": ""
+            }
+          }
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection/a",
+              "filters": [],
+              "orderBys": []
+            },
+            "added": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 1
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": true,
+            "hasPendingWrites": false
+          }
+        ]
+      },
+      {
+        "watchAck": [
+          4
+        ]
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 1000,
+              "value": {
+                "v": 1
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            4
+          ]
+        }
+      },
+      {
+        "watchCurrent": [
+          [
+            4
+          ],
+          "resume-token-1000"
+        ]
+      },
+      {
+        "watchSnapshot": {
+          "version": 1000,
+          "targetIds": []
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection/a",
+              "filters": [],
+              "orderBys": []
+            },
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ]
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 2
+          }
+        ],
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          },
+          {
+            "query": {
+              "path": "collection/a",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 2000,
+              "value": {
+                "v": 2
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            2,
+            4
+          ]
+        }
+      },
+      {
+        "watchSnapshot": {
+          "version": 2000,
+          "targetIds": []
+        }
+      },
+      {
+        "writeAck": {
+          "version": 2000
+        },
+        "stateExpect": {
+          "userCallbacks": {
+            "acknowledgedDocs": [
+              "collection/a"
+            ],
+            "rejectedDocs": []
+          }
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "metadata": [
+              {
+                "key": "collection/a",
+                "version": 2000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          },
+          {
+            "query": {
+              "path": "collection/a",
+              "filters": [],
+              "orderBys": []
+            },
+            "metadata": [
+              {
+                "key": "collection/a",
+                "version": 2000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 1
+      }
+    ]
+  },
+  "onSnapshotsInSync fires for multiple listeners": {
+    "describeName": "Listens:",
+    "itName": "onSnapshotsInSync fires for multiple listeners",
+    "tags": [],
+    "config": {
+      "useGarbageCollection": true,
+      "numClients": 1
+    },
+    "steps": [
+      {
+        "userListen": [
+          2,
+          {
+            "path": "collection",
+            "filters": [],
+            "orderBys": []
+          }
+        ],
+        "stateExpect": {
+          "activeTargets": {
+            "2": {
+              "query": {
+                "path": "collection",
+                "filters": [],
+                "orderBys": []
+              },
+              "resumeToken": ""
+            }
+          }
+        }
+      },
+      {
+        "watchAck": [
+          2
+        ]
+      },
+      {
+        "watchEntity": {
+          "docs": [
+            {
+              "key": "collection/a",
+              "version": 1000,
+              "value": {
+                "v": 1
+              },
+              "options": {
+                "hasLocalMutations": false,
+                "hasCommittedMutations": false
+              }
+            }
+          ],
+          "targets": [
+            2
+          ]
+        }
+      },
+      {
+        "watchCurrent": [
+          [
+            2
+          ],
+          "resume-token-1000"
+        ]
+      },
+      {
+        "watchSnapshot": {
+          "version": 1000,
+          "targetIds": []
+        },
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "added": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 1
+                },
+                "options": {
+                  "hasLocalMutations": false,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": false
+          }
+        ]
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 2
+          }
+        ],
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 2
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "addSnapshotsInSyncListener": true,
+        "expectedSnapshotsInSyncEvents": 1
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 3
+          }
+        ],
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 3
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 3
+      },
+      {
+        "removeSnapshotsInSyncListener": true
+      },
+      {
+        "userSet": [
+          "collection/a",
+          {
+            "v": 4
+          }
+        ],
+        "expect": [
+          {
+            "query": {
+              "path": "collection",
+              "filters": [],
+              "orderBys": []
+            },
+            "modified": [
+              {
+                "key": "collection/a",
+                "version": 1000,
+                "value": {
+                  "v": 4
+                },
+                "options": {
+                  "hasLocalMutations": true,
+                  "hasCommittedMutations": false
+                }
+              }
+            ],
+            "errorCode": 0,
+            "fromCache": false,
+            "hasPendingWrites": true
+          }
+        ],
+        "expectedSnapshotsInSyncEvents": 2
+      }
+    ]
   }
 }
diff --git a/Firestore/Source/API/FIRFirestore+Internal.h b/Firestore/Source/API/FIRFirestore+Internal.h
index 1895656009e..0a41b22d3a6 100644
--- a/Firestore/Source/API/FIRFirestore+Internal.h
+++ b/Firestore/Source/API/FIRFirestore+Internal.h
@@ -15,6 +15,7 @@
  */
 
 #import "FIRFirestore.h"
+#import "FIRListenerRegistration.h"
 
 #include <memory>
 #include <string>
@@ -68,6 +69,26 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (const std::shared_ptr<util::AsyncQueue> &)workerQueue;
 
+/**
+ * Attaches a listener for a snapshots-in-sync event. Server-generated
+ * updates and local changes can affect multiple snapshot listeners.
+ * The snapshots-in-sync event indicates that all listeners affected by
+ * a given change have fired.
+ *
+ * NOTE: The snapshots-in-sync event only indicates that listeners are
+ * in sync with each other, but does not relate to whether those
+ * snapshots are in sync with the server. Use SnapshotMetadata in the
+ * individual listeners to determine if a snapshot is from the cache or
+ * the server.
+ *
+ * @param listener A callback to be called every time all snapshot
+ * listeners are in sync with each other.
+ * @return A FIRListenerRegistration object that can be used to remove the
+ * listener.
+ */
+- (id<FIRListenerRegistration>)addSnapshotsInSyncListener:(void (^)(void))listener
+    NS_SWIFT_NAME(addSnapshotsInSyncListener(_:));
+
 @property(nonatomic, assign, readonly) std::shared_ptr<api::Firestore> wrapped;
 
 @property(nonatomic, assign, readonly) const model::DatabaseId &databaseID;
diff --git a/Firestore/Source/API/FIRFirestore.mm b/Firestore/Source/API/FIRFirestore.mm
index 4d17729dac5..e85c423f1e9 100644
--- a/Firestore/Source/API/FIRFirestore.mm
+++ b/Firestore/Source/API/FIRFirestore.mm
@@ -30,6 +30,7 @@
 #import "Firestore/Source/API/FIRCollectionReference+Internal.h"
 #import "Firestore/Source/API/FIRDocumentReference+Internal.h"
 #import "Firestore/Source/API/FIRFirestore+Internal.h"
+#import "Firestore/Source/API/FIRListenerRegistration+Internal.h"
 #import "Firestore/Source/API/FIRQuery+Internal.h"
 #import "Firestore/Source/API/FIRTransaction+Internal.h"
 #import "Firestore/Source/API/FIRWriteBatch+Internal.h"
@@ -45,24 +46,30 @@
 #include "Firestore/core/src/firebase/firestore/core/transaction.h"
 #include "Firestore/core/src/firebase/firestore/model/database_id.h"
 #include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 #include "Firestore/core/src/firebase/firestore/util/error_apple.h"
 #include "Firestore/core/src/firebase/firestore/util/executor_libdispatch.h"
 #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
 #include "Firestore/core/src/firebase/firestore/util/hard_assert_apple.h"
 #include "Firestore/core/src/firebase/firestore/util/log.h"
 #include "Firestore/core/src/firebase/firestore/util/status.h"
+#include "Firestore/core/src/firebase/firestore/util/statusor.h"
 #include "Firestore/core/src/firebase/firestore/util/string_apple.h"
 
 namespace util = firebase::firestore::util;
 using firebase::firestore::api::DocumentReference;
 using firebase::firestore::api::Firestore;
+using firebase::firestore::api::ListenerRegistration;
 using firebase::firestore::api::ThrowIllegalState;
 using firebase::firestore::api::ThrowInvalidArgument;
 using firebase::firestore::auth::CredentialsProvider;
+using firebase::firestore::core::EventListener;
 using firebase::firestore::model::DatabaseId;
 using firebase::firestore::util::ObjcFailureHandler;
 using firebase::firestore::util::AsyncQueue;
 using firebase::firestore::util::SetFailureHandler;
+using firebase::firestore::util::StatusOr;
+using firebase::firestore::util::Empty;
 
 NS_ASSUME_NONNULL_BEGIN
 
@@ -330,6 +337,14 @@ - (void)terminateInternalWithCompletion:(nullable void (^)(NSError *_Nullable er
   _firestore->Terminate(util::MakeCallback(completion));
 }
 
+- (id<FIRListenerRegistration>)addSnapshotsInSyncListener:(void (^)(void))listener {
+  std::unique_ptr<core::EventListener<Empty>> eventListener =
+      core::EventListener<Empty>::Create([listener](const StatusOr<Empty> &v) { listener(); });
+  std::unique_ptr<ListenerRegistration> result =
+      _firestore->AddSnapshotsInSyncListener(std::move(eventListener));
+  return [[FSTListenerRegistration alloc] initWithRegistration:std::move(result)];
+}
+
 @end
 
 NS_ASSUME_NONNULL_END
diff --git a/Firestore/core/src/firebase/firestore/api/firestore.h b/Firestore/core/src/firebase/firestore/api/firestore.h
index ecac59698b7..1f35cada7e5 100644
--- a/Firestore/core/src/firebase/firestore/api/firestore.h
+++ b/Firestore/core/src/firebase/firestore/api/firestore.h
@@ -22,13 +22,16 @@
 #include <string>
 #include <utility>
 
+#include "Firestore/core/src/firebase/firestore/api/listener_registration.h"
 #include "Firestore/core/src/firebase/firestore/api/settings.h"
 #include "Firestore/core/src/firebase/firestore/auth/credentials_provider.h"
 #include "Firestore/core/src/firebase/firestore/core/database_info.h"
+#include "Firestore/core/src/firebase/firestore/core/event_listener.h"
 #include "Firestore/core/src/firebase/firestore/core/transaction.h"
 #include "Firestore/core/src/firebase/firestore/model/database_id.h"
 #include "Firestore/core/src/firebase/firestore/objc/objc_class.h"
 #include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 #include "Firestore/core/src/firebase/firestore/util/nullability.h"
 #include "Firestore/core/src/firebase/firestore/util/status_fwd.h"
 #include "absl/types/any.h"
@@ -90,6 +93,8 @@ class Firestore : public std::enable_shared_from_this<Firestore> {
   void Terminate(util::StatusCallback callback);
   void ClearPersistence(util::StatusCallback callback);
   void WaitForPendingWrites(util::StatusCallback callback);
+  std::unique_ptr<ListenerRegistration> AddSnapshotsInSyncListener(
+      std::unique_ptr<core::EventListener<util::Empty>> listener);
 
   void EnableNetwork(util::StatusCallback callback);
   void DisableNetwork(util::StatusCallback callback);
diff --git a/Firestore/core/src/firebase/firestore/api/firestore.mm b/Firestore/core/src/firebase/firestore/api/firestore.mm
index 25e1bfa3f79..e66c73b8099 100644
--- a/Firestore/core/src/firebase/firestore/api/firestore.mm
+++ b/Firestore/core/src/firebase/firestore/api/firestore.mm
@@ -19,6 +19,7 @@
 #include "Firestore/core/src/firebase/firestore/api/collection_reference.h"
 #include "Firestore/core/src/firebase/firestore/api/document_reference.h"
 #include "Firestore/core/src/firebase/firestore/api/settings.h"
+#include "Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.h"
 #include "Firestore/core/src/firebase/firestore/api/write_batch.h"
 #include "Firestore/core/src/firebase/firestore/auth/firebase_credentials_provider_apple.h"
 #include "Firestore/core/src/firebase/firestore/core/firestore_client.h"
@@ -38,6 +39,7 @@
 namespace api {
 
 using auth::CredentialsProvider;
+using core::AsyncEventListener;
 using core::DatabaseInfo;
 using core::FirestoreClient;
 using core::Transaction;
@@ -45,6 +47,7 @@
 using model::DocumentKey;
 using model::ResourcePath;
 using util::AsyncQueue;
+using util::Empty;
 using util::Executor;
 using util::ExecutorLibdispatch;
 using util::Status;
@@ -173,6 +176,16 @@
   client_->DisableNetwork(std::move(callback));
 }
 
+std::unique_ptr<ListenerRegistration> Firestore::AddSnapshotsInSyncListener(
+    std::unique_ptr<core::EventListener<Empty>> listener) {
+  EnsureClientConfigured();
+  auto async_listener = AsyncEventListener<Empty>::Create(
+      client_->user_executor(), std::move(listener));
+  client_->AddSnapshotsInSyncListener(std::move(async_listener));
+  return absl::make_unique<SnapshotsInSyncListenerRegistration>(
+      client_, std::move(async_listener));
+}
+
 void Firestore::EnsureClientConfigured() {
   std::lock_guard<std::mutex> lock{mutex_};
 
diff --git a/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.h b/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.h
new file mode 100644
index 00000000000..1d94fbf0421
--- /dev/null
+++ b/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.h
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2019 Google
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_API_SNAPSHOTS_IN_SYNC_LISTENER_REGISTRATION_H_
+#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_API_SNAPSHOTS_IN_SYNC_LISTENER_REGISTRATION_H_
+
+#include <memory>
+
+#include "Firestore/core/src/firebase/firestore/api/listener_registration.h"
+#include "Firestore/core/src/firebase/firestore/core/event_listener.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
+
+namespace firebase {
+namespace firestore {
+namespace core {
+
+class FirestoreClient;
+
+}  // namespace core
+
+namespace api {
+
+/**
+ * An internal handle that encapsulates a user's ability to request that we
+ * stop listening to the snapshots-in-sync listener. When a user calls Remove(),
+ * SnapshotsInSyncListenerRegistration will synchronously mute the listener and
+ * then send a request to actually unlisten.
+ */
+class SnapshotsInSyncListenerRegistration : public ListenerRegistration {
+ public:
+  SnapshotsInSyncListenerRegistration(
+      std::shared_ptr<core::FirestoreClient> client,
+      std::shared_ptr<core::AsyncEventListener<util::Empty>> async_listener);
+
+  /**
+   * Removes the listener being tracked by this FIRListenerRegistration. After
+   * the initial call, subsequent calls have no effect.
+   */
+  void Remove() override;
+
+ private:
+  /** The client that was used to register this listen. */
+  std::shared_ptr<core::FirestoreClient> client_;
+
+  /** The async listener that is used to mute events synchronously. */
+  std::weak_ptr<core::AsyncEventListener<util::Empty>> async_listener_;
+};
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_API_SNAPSHOTS_IN_SYNC_LISTENER_REGISTRATION_H_
diff --git a/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.mm b/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.mm
new file mode 100644
index 00000000000..6a688c3b4b8
--- /dev/null
+++ b/Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.mm
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 Google
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <utility>
+
+#include "Firestore/core/src/firebase/firestore/api/snapshots_in_sync_listener_registration.h"
+
+#include "Firestore/core/src/firebase/firestore/core/firestore_client.h"
+
+namespace firebase {
+namespace firestore {
+namespace api {
+
+SnapshotsInSyncListenerRegistration::SnapshotsInSyncListenerRegistration(
+    std::shared_ptr<core::FirestoreClient> client,
+    std::shared_ptr<core::AsyncEventListener<util::Empty>> async_listener)
+    : client_(std::move(client)), async_listener_(std::move(async_listener)) {
+}
+
+void SnapshotsInSyncListenerRegistration::Remove() {
+  auto async_listener = async_listener_.lock();
+  if (async_listener) {
+    async_listener->Mute();
+    async_listener_.reset();
+  }
+
+  client_->RemoveSnapshotsInSyncListener(async_listener);
+  client_.reset();
+}
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/src/firebase/firestore/core/event_manager.h b/Firestore/core/src/firebase/firestore/core/event_manager.h
index c11c028471d..bdc943529d7 100644
--- a/Firestore/core/src/firebase/firestore/core/event_manager.h
+++ b/Firestore/core/src/firebase/firestore/core/event_manager.h
@@ -19,6 +19,7 @@
 
 #include <memory>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 #include "Firestore/core/src/firebase/firestore/core/query.h"
@@ -27,6 +28,7 @@
 #include "Firestore/core/src/firebase/firestore/core/view_snapshot.h"
 #include "Firestore/core/src/firebase/firestore/model/types.h"
 #include "Firestore/core/src/firebase/firestore/objc/objc_class.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 #include "Firestore/core/src/firebase/firestore/util/nullability.h"
 #include "Firestore/core/src/firebase/firestore/util/status_fwd.h"
 #include "absl/algorithm/container.h"
@@ -56,16 +58,28 @@ class EventManager : public SyncEngineCallback {
   model::TargetId AddQueryListener(
       std::shared_ptr<core::QueryListener> listener);
 
-  /** Removes a previously added listener. It's a no-op if the listener is not
-   * found. */
+  /**
+   * Removes a previously added listener. It's a no-op if the listener is not
+   * found.
+   */
   void RemoveQueryListener(std::shared_ptr<core::QueryListener> listener);
 
+  void AddSnapshotsInSyncListener(
+      const std::shared_ptr<EventListener<util::Empty>>& listener);
+  void RemoveSnapshotsInSyncListener(
+      const std::shared_ptr<EventListener<util::Empty>>& listener);
+
   // Implements `QueryEventCallback`.
   void HandleOnlineStateChange(model::OnlineState online_state) override;
   void OnViewSnapshots(std::vector<core::ViewSnapshot>&& snapshots) override;
   void OnError(const core::Query& query, const util::Status& error) override;
 
  private:
+  /**
+   * Call all global snapshot listeners that have been set.
+   */
+  void RaiseSnapshotsInSyncEvent();
+
   /**
    * Holds the listeners and the last received ViewSnapshot for a query being
    * tracked by EventManager.
@@ -100,6 +114,8 @@ class EventManager : public SyncEngineCallback {
   QueryEventSource* query_event_source_ = nullptr;
   model::OnlineState online_state_ = model::OnlineState::Unknown;
   std::unordered_map<core::Query, QueryListenersInfo> queries_;
+  std::unordered_set<std::shared_ptr<EventListener<util::Empty>>>
+      snapshots_in_sync_listeners_;
 };
 
 }  // namespace core
diff --git a/Firestore/core/src/firebase/firestore/core/event_manager.mm b/Firestore/core/src/firebase/firestore/core/event_manager.mm
index ae6f094476c..ee8a374cd31 100644
--- a/Firestore/core/src/firebase/firestore/core/event_manager.mm
+++ b/Firestore/core/src/firebase/firestore/core/event_manager.mm
@@ -14,14 +14,17 @@
  * limitations under the License.
  */
 
-#include "Firestore/core/src/firebase/firestore/core/event_manager.h"
-
 #include <utility>
 
+#include "Firestore/core/src/firebase/firestore/core/event_manager.h"
+#include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
+
 namespace firebase {
 namespace firestore {
 namespace core {
 
+using util::Empty;
+
 EventManager::EventManager(QueryEventSource* query_event_source)
     : query_event_source_(query_event_source) {
   query_event_source->SetCallback(this);
@@ -37,10 +40,15 @@
 
   query_info.listeners.push_back(listener);
 
-  listener->OnOnlineStateChanged(online_state_);
+  bool raised_event = listener->OnOnlineStateChanged(online_state_);
+  HARD_ASSERT(!raised_event, "onOnlineStateChanged() shouldn't raise an event "
+                             "for brand-new listeners.");
 
   if (query_info.view_snapshot().has_value()) {
-    listener->OnViewSnapshot(query_info.view_snapshot().value());
+    raised_event = listener->OnViewSnapshot(query_info.view_snapshot().value());
+    if (raised_event) {
+      RaiseSnapshotsInSyncEvent();
+    }
   }
 
   if (first_listen) {
@@ -67,30 +75,60 @@
   }
 }
 
+void EventManager::AddSnapshotsInSyncListener(
+    const std::shared_ptr<EventListener<Empty>>& listener) {
+  snapshots_in_sync_listeners_.insert(listener);
+  listener->OnEvent(Empty());
+}
+
+void EventManager::RemoveSnapshotsInSyncListener(
+    const std::shared_ptr<EventListener<Empty>>& listener) {
+  snapshots_in_sync_listeners_.erase(listener);
+}
+
 void EventManager::HandleOnlineStateChange(model::OnlineState online_state) {
+  bool raised_event = false;
   online_state_ = online_state;
 
   for (auto&& kv : queries_) {
     QueryListenersInfo& info = kv.second;
     for (auto&& listener : info.listeners) {
-      listener->OnOnlineStateChanged(online_state_);
+      if (listener->OnOnlineStateChanged(online_state_)) {
+        raised_event = true;
+      }
     }
   }
+  if (raised_event) {
+    RaiseSnapshotsInSyncEvent();
+  }
+}
+
+void EventManager::RaiseSnapshotsInSyncEvent() {
+  Empty empty{};
+  for (const auto& listener : snapshots_in_sync_listeners_) {
+    listener->OnEvent(empty);
+  }
 }
 
 void EventManager::OnViewSnapshots(
     std::vector<core::ViewSnapshot>&& snapshots) {
+  bool raised_event = false;
   for (ViewSnapshot& snapshot : snapshots) {
     const Query& query = snapshot.query();
     auto found_iter = queries_.find(query);
     if (found_iter != queries_.end()) {
       QueryListenersInfo& query_info = found_iter->second;
       for (const auto& listener : query_info.listeners) {
-        listener->OnViewSnapshot(snapshot);
+        if (listener->OnViewSnapshot(snapshot)) {
+          raised_event = true;
+        }
       }
       query_info.set_view_snapshot(std::move(snapshot));
     }
   }
+  if (raised_event) {
+    RaiseSnapshotsInSyncEvent();
+  }
 }
 
 void EventManager::OnError(const core::Query& query,
diff --git a/Firestore/core/src/firebase/firestore/core/firestore_client.h b/Firestore/core/src/firebase/firestore/core/firestore_client.h
index 478ec09be54..5f06dff2e56 100644
--- a/Firestore/core/src/firebase/firestore/core/firestore_client.h
+++ b/Firestore/core/src/firebase/firestore/core/firestore_client.h
@@ -24,6 +24,7 @@
 
 #include "Firestore/core/src/firebase/firestore/api/document_reference.h"
 #include "Firestore/core/src/firebase/firestore/api/document_snapshot.h"
+#include "Firestore/core/src/firebase/firestore/api/listener_registration.h"
 #include "Firestore/core/src/firebase/firestore/api/query_core.h"
 #include "Firestore/core/src/firebase/firestore/api/settings.h"
 #include "Firestore/core/src/firebase/firestore/auth/credentials_provider.h"
@@ -38,6 +39,7 @@
 #include "Firestore/core/src/firebase/firestore/model/mutation.h"
 #include "Firestore/core/src/firebase/firestore/util/async_queue.h"
 #include "Firestore/core/src/firebase/firestore/util/delayed_constructor.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 #include "Firestore/core/src/firebase/firestore/util/executor.h"
 #include "Firestore/core/src/firebase/firestore/util/status_fwd.h"
 
@@ -139,6 +141,18 @@ class FirestoreClient : public std::enable_shared_from_this<FirestoreClient> {
                    TransactionUpdateCallback update_callback,
                    TransactionResultCallback result_callback);
 
+  /**
+   * Adds a listener to be called when a snapshots-in-sync event fires.
+   */
+  void AddSnapshotsInSyncListener(
+      const std::shared_ptr<EventListener<util::Empty>>& listener);
+
+  /**
+   * Removes a specific listener for snapshots-in-sync events.
+   */
+  void RemoveSnapshotsInSyncListener(
+      const std::shared_ptr<EventListener<util::Empty>>& listener);
+
   /** The database ID of the DatabaseInfo this client was initialized with. */
   const model::DatabaseId& database_id() const {
     return database_info_.database_id();
diff --git a/Firestore/core/src/firebase/firestore/core/firestore_client.mm b/Firestore/core/src/firebase/firestore/core/firestore_client.mm
index 5fc36557492..854ae4b632e 100644
--- a/Firestore/core/src/firebase/firestore/core/firestore_client.mm
+++ b/Firestore/core/src/firebase/firestore/core/firestore_client.mm
@@ -51,6 +51,7 @@
 using firestore::Error;
 using api::DocumentReference;
 using api::DocumentSnapshot;
+using api::ListenerRegistration;
 using api::QuerySnapshot;
 using api::Settings;
 using api::SnapshotMetadata;
@@ -70,11 +71,12 @@
 using model::OnlineState;
 using remote::Datastore;
 using remote::RemoteStore;
-using util::Path;
 using util::AsyncQueue;
 using util::DelayedConstructor;
 using util::DelayedOperation;
+using util::Empty;
 using util::Executor;
+using util::Path;
 using util::Status;
 using util::StatusCallback;
 using util::StatusOr;
@@ -451,6 +453,20 @@ QuerySnapshot result(query.firestore(), query.query(), std::move(snapshot),
   });
 }
 
+void FirestoreClient::AddSnapshotsInSyncListener(
+    const std::shared_ptr<EventListener<Empty>>& user_listener) {
+  auto shared_this = shared_from_this();
+  worker_queue()->Enqueue([shared_this, user_listener] {
+    shared_this->event_manager_->AddSnapshotsInSyncListener(
+        std::move(user_listener));
+  });
+}
+
+void FirestoreClient::RemoveSnapshotsInSyncListener(
+    const std::shared_ptr<EventListener<Empty>>& user_listener) {
+  event_manager_->RemoveSnapshotsInSyncListener(user_listener);
+}
+
 }  // namespace core
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/firebase/firestore/core/query_listener.cc b/Firestore/core/src/firebase/firestore/core/query_listener.cc
index ddefeb7fb37..fe4e9a44ab6 100644
--- a/Firestore/core/src/firebase/firestore/core/query_listener.cc
+++ b/Firestore/core/src/firebase/firestore/core/query_listener.cc
@@ -40,11 +40,11 @@ QueryListener::QueryListener(Query query,
       listener_(std::move(listener)) {
 }
 
-void QueryListener::OnViewSnapshot(ViewSnapshot snapshot) {
+bool QueryListener::OnViewSnapshot(ViewSnapshot snapshot) {
   HARD_ASSERT(
       !snapshot.document_changes().empty() || snapshot.sync_state_changed(),
       "We got a new snapshot with no changes?");
-
+  bool raised_event = false;
   if (!options_.include_document_metadata_changes()) {
     // Remove the metadata-only changes.
     std::vector<DocumentViewChange> changes;
@@ -67,24 +67,33 @@ void QueryListener::OnViewSnapshot(ViewSnapshot snapshot) {
   if (!raised_initial_event_) {
     if (ShouldRaiseInitialEvent(snapshot, online_state_)) {
       RaiseInitialEvent(snapshot);
+      raised_event = true;
     }
   } else if (ShouldRaiseEvent(snapshot)) {
     listener_->OnEvent(snapshot);
+    raised_event = true;
   }
 
   snapshot_ = std::move(snapshot);
+  return raised_event;
 }
 
 void QueryListener::OnError(Status error) {
   listener_->OnEvent(std::move(error));
 }
 
-void QueryListener::OnOnlineStateChanged(OnlineState online_state) {
+/**
+ * Returns whether a snaphsot was raised.
+ */
+bool QueryListener::OnOnlineStateChanged(OnlineState online_state) {
   online_state_ = online_state;
+  bool raised_event = false;
   if (snapshot_.has_value() && !raised_initial_event_ &&
       ShouldRaiseInitialEvent(snapshot_.value(), online_state)) {
     RaiseInitialEvent(snapshot_.value());
+    raised_event = true;
   }
+  return raised_event;
 }
 
 bool QueryListener::ShouldRaiseInitialEvent(const ViewSnapshot& snapshot,
diff --git a/Firestore/core/src/firebase/firestore/core/query_listener.h b/Firestore/core/src/firebase/firestore/core/query_listener.h
index 6da4510ba14..4fd3a638873 100644
--- a/Firestore/core/src/firebase/firestore/core/query_listener.h
+++ b/Firestore/core/src/firebase/firestore/core/query_listener.h
@@ -83,9 +83,18 @@ class QueryListener {
     return snapshot_;
   }
 
-  virtual void OnViewSnapshot(ViewSnapshot snapshot);
+  /**
+   * Applies the new ViewSnapshot to this listener, raising a user-facing event
+   * if applicable (depending on what changed, whether the user has opted into
+   * metadata-only changes, etc.). Returns true if a user-facing event was
+   * indeed raised.
+   */
+  virtual bool OnViewSnapshot(ViewSnapshot snapshot);
+
   virtual void OnError(util::Status error);
-  virtual void OnOnlineStateChanged(model::OnlineState online_state);
+
+  /** Returns whether a snapshot was raised. */
+  virtual bool OnOnlineStateChanged(model::OnlineState online_state);
 
  private:
   bool ShouldRaiseInitialEvent(const ViewSnapshot& snapshot,
diff --git a/Firestore/core/src/firebase/firestore/immutable/sorted_set.h b/Firestore/core/src/firebase/firestore/immutable/sorted_set.h
index 8609e27e855..7907a3bff88 100644
--- a/Firestore/core/src/firebase/firestore/immutable/sorted_set.h
+++ b/Firestore/core/src/firebase/firestore/immutable/sorted_set.h
@@ -23,6 +23,7 @@
 #include "Firestore/core/src/firebase/firestore/immutable/sorted_container.h"
 #include "Firestore/core/src/firebase/firestore/immutable/sorted_map.h"
 #include "Firestore/core/src/firebase/firestore/util/comparison.h"
+#include "Firestore/core/src/firebase/firestore/util/empty.h"
 #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
 #include "Firestore/core/src/firebase/firestore/util/hashing.h"
 #include "absl/base/attributes.h"
@@ -31,20 +32,9 @@ namespace firebase {
 namespace firestore {
 namespace immutable {
 
-namespace impl {
-
-// An empty value to associate with keys in the underlying map.
-struct Empty {
-  friend bool operator==(Empty /* left */, Empty /* right */) {
-    return true;
-  }
-};
-
-}  // namespace impl
-
 template <typename K,
           typename C = util::Comparator<K>,
-          typename V = impl::Empty,
+          typename V = util::Empty,
           typename M = SortedMap<K, V, C>>
 class SortedSet : public SortedContainer {
  public:
diff --git a/Firestore/core/src/firebase/firestore/util/empty.h b/Firestore/core/src/firebase/firestore/util/empty.h
new file mode 100644
index 00000000000..8e1abfdb28e
--- /dev/null
+++ b/Firestore/core/src/firebase/firestore/util/empty.h
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2019 Google
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_UTIL_EMPTY_H_
+#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_UTIL_EMPTY_H_
+
+namespace firebase {
+namespace firestore {
+namespace util {
+
+/**
+ * Used to represent an empty value.
+ */
+struct Empty {
+  friend bool operator==(Empty /* left */, Empty /* right */) {
+    return true;
+  }
+};
+
+}  // namespace util
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_UTIL_EMPTY_H_
diff --git a/Firestore/core/test/firebase/firestore/core/event_manager_test.mm b/Firestore/core/test/firebase/firestore/core/event_manager_test.mm
index 657714b2435..f5e32f71cd3 100644
--- a/Firestore/core/test/firebase/firestore/core/event_manager_test.mm
+++ b/Firestore/core/test/firebase/firestore/core/event_manager_test.mm
@@ -151,8 +151,9 @@ explicit FakeQueryListener(core::Query query)
                         NoopViewSnapshotHandler()) {
     }
 
-    void OnOnlineStateChanged(OnlineState online_state) override {
+    bool OnOnlineStateChanged(OnlineState online_state) override {
       events.push_back(online_state);
+      return false;
     }
 
     std::vector<OnlineState> events;