`.pipe()` not executing `debounceTime`

Question

I'm trying to debounce() an Observable with pipe() and chaining .subscribe() but for some reason the function in the subscribe is still being called over a dozen times in one go.

What I'm trying to do is pipe the withChangesForTables and debounce the sync call because I want it to be called only when a whole batch of changes have been made. So I created a provider for the sync and wrapped it around my RootNavigator

withChangesForTables on WatermelonDB source code

const SyncContext = createContext();
function useSync() {
    return useContext(SyncContext);
}

function SyncProvider({children}) {
    const [isSyncing, setIsSyncing] = useState(false);
    const [hasUnsynced, setHasUnsynced] = useState(false);

    async function checkUnsyncedChanges() {
        const hasChanges = await hasUnsyncedChanges({
            database
        });
        setHasUnsynced(hasChanges);
    }
    async function sync() {
        await checkUnsyncedChanges();
        if (!isSyncing && hasUnsynced) {
            setIsSyncing(true);
            await synchronizeWithServer();
            setIsSyncing(false);
        }
    }

    
    database.withChangesForTables([
        'table_name',
        'table_name2'
    ]).pipe(
        skip(1),
        // ignore records simply becoming `synced`
        filter(changes => !changes.every(change => change.record.syncStatus === 'synced')),
        // debounce to avoid syncing in the middle of related actions - I put 100000 to test only
        debounceTime(100000),
    ).subscribe({
        //calls API endpoint to sync local DB with server
        next: () => sync(), 
        error: e => console.log(e)
    });

    const value = {
        isSyncing,
        hasUnsynced,
        checkUnsyncedChanges,
        sync
    };

    return (
        <SyncContext.Provider value={value}>
            {children}
        </SyncContext.Provider>
    );
}

Answer 1

I had to move withChangesForTables into a useEffect and retrun it in order to unsubcribe which seems to have resolved the issue. The code now looks something like this:

useEffect(() => {
    return database.withChangesForTables([
        'table_name',
        'table_name2'
    ]).pipe(
        skip(1),
        filter(changes => !changes.every(change => change.record.syncStatus === 'synced')),
        debounceTime(500),
    ).subscribe({
        next: () => sync(), 
        error: e => console.log(e)
    });
}, [])