Inside TanStack Query - network mode

•8 min read
tanstack querystate management
Back

在官方文件提到：TanStack Query 提供三種不同的 network modes，區分沒有網路連線的情況下，Queries, Mutations 的運作方式

TanStack Query provides three different network modes to distinguish how Queries and Mutations should behave if you have no network connection. This mode can be set for each Query / Mutation individually, or globally via the query / mutation defaults.

Since TanStack Query is most often used for data fetching in combination with data fetching libraries, the default network mode is online.

這裡紀錄 Query 與 network mode 相關實作，參考版本為 v5.45.1

OnlineManager

先從 query-core 的 OnlineManager 開始看起，OnlineManager 是 Subscribable 的子類別

onlineManager.ts
export class OnlineManager extends Subscribable<Listener> {
  // ...
  constructor() {
    super()
    // ...
  }
  // ...
}

Subscribable 大致實作 observer pattern 中的：

Subscriable.onSubscribe(), Subscriable.onUnsubscribe()，可以視子類別的需求決定是否實作。因為原型鏈繼承，如果子類別有實作會優先選取 (L26-28, L30-32)

subscribable.ts
type Listener = () => void
 
export class Subscribable<TListener extends Function = Listener> {
  protected listeners: Set<TListener>
 
  constructor() {
    this.listeners = new Set()
    this.subscribe = this.subscribe.bind(this)
  }
 
  subscribe(listener: TListener): () => void {
    this.listeners.add(listener)
 
    this.onSubscribe()
 
    return () => {
      this.listeners.delete(listener)
      this.onUnsubscribe()
    }
  }
 
  hasListeners(): boolean {
    return this.listeners.size > 0
  }
 
  protected onSubscribe(): void {
    // Do nothing
  }
 
  protected onUnsubscribe(): void {
    // Do nothing
  }
}

繼續看到 OnlineManager：

onlineManager.ts
export class OnlineManager extends Subscribable<Listener> {
  #online = true
  #cleanup?: () => void
 
  #setup: SetupFn
 
  constructor() {
    super()
    this.#setup = (onOnline) => {
      // addEventListener does not exist in React Native, but window does
      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
      if (!isServer && window.addEventListener) {
        const onlineListener = () => onOnline(true)
        const offlineListener = () => onOnline(false)
        // Listen to online
        window.addEventListener('online', onlineListener, false)
        window.addEventListener('offline', offlineListener, false)
 
        return () => {
          // Be sure to unsubscribe if a new handler is set
          window.removeEventListener('online', onlineListener)
          window.removeEventListener('offline', offlineListener)
        }
      }
 
      return
    }
  }
 
  protected onSubscribe(): void {
    if (!this.#cleanup) {
      this.setEventListener(this.#setup)
    }
  }
 
  protected onUnsubscribe() {
    if (!this.hasListeners()) {
      this.#cleanup?.()
      this.#cleanup = undefined
    }
  }
 
  setEventListener(setup: SetupFn): void {
    this.#setup = setup
    this.#cleanup?.()
    this.#cleanup = setup(this.setOnline.bind(this))
  }
 
  setOnline(online: boolean): void {
    const changed = this.#online !== online
 
    if (changed) {
      this.#online = online
      this.listeners.forEach((listener) => {
        listener(online)
      })
    }
  }
 
  isOnline(): boolean {
    return this.#online
  }
}
 
export const onlineManager = new OnlineManager()

呼叫 QueryClient.mount()，QueryClient 會對 OnlineManager 進行訂閱：

queryClient.ts
export class QueryClient {
  // ...
  mount(): void {
    // ...
    this.#unsubscribeOnline = onlineManager.subscribe(async (online) => {
      if (online) {
        await this.resumePausedMutations()
        this.#queryCache.onOnline()
      }
    })
  }
}

tanstack/react-query 是在 QueryClientProvier 透過 useEffect：

QueryClientProvider.tsx
export type QueryClientProviderProps = {
  client: QueryClient
  children?: React.ReactNode
}
 
export const QueryClientProvider = ({
  client,
  children,
}: QueryClientProviderProps): React.JSX.Element => {
  React.useEffect(() => {
    client.mount()
    return () => {
      client.unmount()
    }
  }, [client])
 
  return (
    <QueryClientContext.Provider value={client}>
      {children}
    </QueryClientContext.Provider>
  )
}
💡

補充：tanstack/react-query 有 re-export query-core，可以 import OnlineManager 實例，如下方範例：

import { onlineManager } from '@tanstack/react-query'
 
const unsubscribe = onlineManager.subscribe((isOnline) => {
  console.log('isOnline', isOnline)
})

QueryCache.onLine()

在 QueryClient.mount() 得知，當收到 OnlineManager 恢復連線的通知時，會呼叫 this.#queryCache.onOnline()

QueryCache.onOnline 會對所有 Query 呼叫 Query.onOnline()

queryCache.ts
export class QueryCache extends Subscribable<QueryCacheListener> {
  //...
  onOnline(): void {
    notifyManager.batch(() => {
      this.getAll().forEach((query) => {
        query.onOnline()
      })
    })
  }
}

Query.onOnline()

先看到前三行 observer 的部分 (observer 為 QueryObserver 的實例，在呼叫 useQuery 後產生，並保存在 Query 中)

  1. 找到任一需要重新請求的 observer，QueryObserver.shouldFetchOnReconnect() 會判斷：
    • queryOptions.enabled !== false
    • queryOptions.refetchOnReconnect === 'always'，不考慮資料是否過期，直接 refetch
    • queryOptions.refetchOnReconnect === true，資料過期才會 refetch
  2. observer?.refetch({ cancelRefetch: false })：
    • { cancelRefetch: false }，表示如果該請求進行中，不做 refetch
    • 最後會呼叫 this.#currentQuery.fetch() 即為 Query.fetch()
query.ts
export class Query<
  TQueryFnData = unknown,
  TError = DefaultError,
  TData = TQueryFnData,
  TQueryKey extends QueryKey = QueryKey,
> extends Removable {
  // ...
  onOnline(): void {
    const observer = this.observers.find((x) => x.shouldFetchOnReconnect())
 
    observer?.refetch({ cancelRefetch: false })
 
    // Continue fetch if currently paused
    this.#retryer?.continue()
  }
  shouldFetchOnReconnect(): boolean {
    return shouldFetchOn(
      this.#currentQuery,
      this.options,
      this.options.refetchOnReconnect,
    )
  }
}
 
function shouldFetchOn(
  query: Query<any, any, any, any>,
  options: QueryObserverOptions<any, any, any, any, any>,
  field: (typeof options)['refetchOnMount'] &
    (typeof options)['refetchOnWindowFocus'] &
    (typeof options)['refetchOnReconnect'],
) {
  if (options.enabled !== false) {
    const value = typeof field === 'function' ? field(query) : field
 
    return value === 'always' || (value !== false && isStale(query, options))
  }
  return false
}

#this.retryer?.continue()：

query.ts
export class Query<
  TQueryFnData = unknown,
  TError = DefaultError,
  TData = TQueryFnData,
  TQueryKey extends QueryKey = QueryKey,
> extends Removable {
  // ...
  onOnline(): void {
    const observer = this.observers.find((x) => x.shouldFetchOnReconnect())
 
    observer?.refetch({ cancelRefetch: false })
 
    // Continue fetch if currently paused
    this.#retryer?.continue()
  }
  // ...
  fetch(
    options?: QueryOptions<TQueryFnData, TError, TData, TQueryKey>,
    fetchOptions?: FetchOptions<TQueryFnData>,
  ): Promise<TData> {
    // ...
 
    // Try to fetch the data
    this.#retryer = createRetryer({
      // ...
      onPause: () => {
        this.#dispatch({ type: 'pause' })
      },
      onContinue: () => {
        this.#dispatch({ type: 'continue' })
      },
      networkMode: context.options.networkMode,
      canRun: () => true,
    })
 
    return this.#retryer.start()
  }
  #dispatch(action: Action<TData, TError>): void {
    const reducer = (
      state: QueryState<TData, TError>,
    ): QueryState<TData, TError> => {
      switch (action.type) {
        // ...
        case 'continue':
          return {
            ...state,
            fetchStatus: 'fetching',
          }
        // ...
      }
    }
 
    this.state = reducer(this.state)
    notifyManager.batch(() => {
      this.observers.forEach((observer) => {
        observer.onQueryUpdate()
      })
 
      this.#cache.notify({ query: this, type: 'updated', action })
    })
  }
}

看到 Retryer 的實作：

retryer.ts
import { onlineManager } from './onlineManager'
 
export function createRetryer<TData = unknown, TError = DefaultError>(
  config: RetryerConfig<TData, TError>,
): Retryer<TData> {
  let isResolved = false
  let continueFn: ((value?: unknown) => void) | undefined
 
  const canContinue = () =>
    focusManager.isFocused() &&
    (config.networkMode === 'always' || onlineManager.isOnline()) &&
    config.canRun()
 
  const canStart = () => canFetch(config.networkMode) && config.canRun()
 
  const pause = () => {
    return new Promise((continueResolve) => {
      continueFn = (value) => {
        if (isResolved || canContinue()) {
          continueResolve(value)
        }
      }
      config.onPause?.()
    }).then(() => {
      continueFn = undefined
      if (!isResolved) {
        config.onContinue?.()
      }
    })
  }
 
  return {
    // ...
    continue: () => {
      continueFn?.()
      return promise
    },
    // ...
    start: () => {
      // Start loop
      if (canStart()) {
        run()
      } else {
        pause().then(run)
      }
      return promise
    },
  }
}
 
export function canFetch(networkMode: NetworkMode | undefined): boolean {
  return (networkMode ?? 'online') === 'online'
    ? onlineManager.isOnline()
    : true
}
 
interface RetryerConfig<TData = unknown, TError = DefaultError> {
  fn: () => TData | Promise<TData>
  initialPromise?: Promise<TData>
  abort?: () => void
  onError?: (error: TError) => void
  onSuccess?: (data: TData) => void
  onFail?: (failureCount: number, error: TError) => void
  onPause?: () => void
  onContinue?: () => void
  retry?: RetryValue<TError>
  retryDelay?: RetryDelayValue<TError>
  networkMode: NetworkMode | undefined
  canRun: () => boolean
}

Summary

Reference