Listening to document changes

To help to work with asynchronous event listener the library supports high level API for listening the events from Firestore on a separate thread:

The listener implementation needs to be provided with a storage for the last received token for specified targets to be able to resume listening the changes from the last handled token and to avoid receiving all previous changes.

The library provides basic implementations for storing the tokens but you can implement your own more sophisticated storage if needed:

• FirestoreTempFilesListenStateStorage - resume tokens stored as temporary files on local FS;
• FirestoreMemListenStateStorage - in memory storage backed by HashMap (with this implementation if you restart your app, you will receive all notifications again);

use firestore::*;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Deserialize, Serialize)]
struct MyTestStructure { some_id: String }
const TEST_COLLECTION_NAME: &str = "test-listen";
const TEST_TARGET_ID_BY_QUERY: FirestoreListenerTarget = FirestoreListenerTarget::new(42_u32);
const TEST_TARGET_ID_BY_DOC_IDS: FirestoreListenerTarget = FirestoreListenerTarget::new(17_u32);
async fn example(db: FirestoreDb, doc_id1: String, doc_id2: String) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut listener = db
    .create_listener(
        FirestoreTempFilesListenStateStorage::new(), // or FirestoreMemListenStateStorage or your own implementation
    )
    .await?;

// Adding query listener
db.fluent()
    .select()
    .from(TEST_COLLECTION_NAME)
    .listen()
    .add_target(TEST_TARGET_ID_BY_QUERY, &mut listener)?;

// Adding docs listener by IDs
db.fluent()
    .select()
    .by_id_in(TEST_COLLECTION_NAME)
    .batch_listen([doc_id1, doc_id2])
    .add_target(TEST_TARGET_ID_BY_DOC_IDS, &mut listener)?;

listener
    .start(|event| async move {
        match event {
            FirestoreListenEvent::DocumentChange(ref doc_change) => {
                println!("Doc changed: {:?}", doc_change);

                if let Some(doc) = &doc_change.document {
                    let obj: MyTestStructure =
                        FirestoreDb::deserialize_doc_to::<MyTestStructure>(doc)
                            .expect("Deserialized object");
                    println!("As object: {:?}", obj);
                }
            }
            _ => {
                println!("Received a listen response event to handle: {:?}", event);
            }
        }

        Ok(())
    })
    .await?;

// Wait some events like Ctrl-C, signals, etc
// <put-your-implementation-for-wait-here>

// and then shutdown
listener.shutdown().await?;
Ok(())
}

shutdown() finishes the current callback and resume-state writes before returning. Use shutdown_with_timeout(Duration::from_secs(5)) to allow five seconds for graceful shutdown, then abort and join the task. Aborting can interrupt callbacks and token writes, so changes may be replayed.

Targets can also be added and removed on a running listener with add_target and remove_target. The listener changes them on its open stream, so the other targets keep listening as they were. add_target returns an error for a target that cannot be turned into a listen request, so a bad target never reaches the running listener.

Target changes with the Remove type reach the callback in two forms:

• with no cause, Firestore acknowledges a remove_target call, or ends a target added with add_target_once. Events for a removed target can still arrive until this acknowledgement;
• with a cause, Firestore removed the target on its own. For PERMISSION_DENIED, NOT_FOUND, FAILED_PRECONDITION, UNIMPLEMENTED, and INVALID_ARGUMENT on a target added without a resume token or read time, the listener stops listening on that target only and unregisters it. For any other cause it adds the target back from scratch after the retry delay. An ended target is unregistered before the callback receives the change, so has_target called from the callback already tells whether the target comes back.

When Firestore removes a target on its own while a remove_target call for it is on its way, Firestore does not acknowledge the removal. It closes the whole stream with NOT_FOUND instead. The listener then reconnects, and the other targets are added back at once.

The listener reconnects by itself after network errors, closed streams and every other error, after the retry delay. It stops only on an error a retry cannot fix:

• Firestore rejects the listen request as invalid again after every resume token was discarded;
• the listen request fails with InvalidParametersError;
• the AddTarget or RemoveTarget request for a target cannot be built;
• the callback or the token storage panics.

A stopped listener rejects add_target with a SystemError, and shutdown() returns the error it stopped with.

A listen stream can also stay open and stop delivering anything, which HTTP/2 keep-alive does not detect while the connection itself still answers. To reset such a stream, set idle_timeout in the listener params:

use firestore::*;
use std::time::Duration;
async fn example(db: FirestoreDb) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let listener = db
    .create_listener_with_params(
        FirestoreMemListenStateStorage::new(),
        FirestoreListenerParams::new().with_idle_timeout(Duration::from_secs(120)),
    )
    .await?;
Ok(())
}

It is off by default. When set:

• any ListenResponse counts as activity, NO_CHANGE target changes included;
• the time counts from when the stream is open, so a connect that never gets that far is not covered, and only while at least one target on the stream waits for an answer from Firestore, not while every target waits to be added back after the retry delay;
• the time starts again once the callback and the token storage have handled each response;
• on expiry the listener waits the retry delay and reconnects, adding every target back with its stored resume token. Changes after the last stored token may be delivered again, as after any reconnect.

120 seconds is the recommended value, and the one Google’s Node.js client uses. Firestore documents no interval for the updates it sends on a quiet stream, so a much shorter value can reset healthy streams.

Full example available here.

A listener takes Core targets only: a query or documents by ID. Pipelines and SQL statements cannot be listened to, and their writes reach a listener as any other change, see Retries and listeners.

On the Enterprise edition

A listener the application opens receives changes on an Enterprise database that has real-time updates enabled; Google ties listener support (onSnapshot) to that setting, which is fixed when the database is created, see Settings fixed at creation. Real-time updates are billed separately on Enterprise, see Cost per edition.