Revert "refactor(runtime-core): reduce the abstraction of base watcher"

johnsoncodehk · johnsoncodehk · commit ceb174b9de2e · 2025-04-27T09:41:01.000+08:00
This reverts commit 9e4c584.
diff --git a/packages/reactivity/__tests__/watch.spec.ts b/packages/reactivity/__tests__/watch.spec.ts
@@ -3,12 +3,40 @@ import {
   type Ref,
   WatchErrorCodes,
   type WatchOptions,
+  type WatchScheduler,
   computed,
   onWatcherCleanup,
   ref,
   watch,
 } from '../src'
 
+const queue: (() => void)[] = []
+
+// a simple scheduler for testing purposes
+let isFlushPending = false
+const resolvedPromise = /*@__PURE__*/ Promise.resolve() as Promise<any>
+const nextTick = (fn?: () => any) =>
+  fn ? resolvedPromise.then(fn) : resolvedPromise
+
+const scheduler: WatchScheduler = (job, isFirstRun) => {
+  if (isFirstRun) {
+    job()
+  } else {
+    queue.push(job)
+    flushJobs()
+  }
+}
+
+const flushJobs = () => {
+  if (isFlushPending) return
+  isFlushPending = true
+  resolvedPromise.then(() => {
+    queue.forEach(job => job())
+    queue.length = 0
+    isFlushPending = false
+  })
+}
+
 describe('watch', () => {
   test('effect', () => {
     let dummy: any
@@ -119,6 +147,54 @@ describe('watch', () => {
     expect(dummy).toBe(30)
   })
 
+  test('nested calls to baseWatch and onWatcherCleanup', async () => {
+    let calls: string[] = []
+    let source: Ref<number>
+    let copyist: Ref<number>
+    const scope = new EffectScope()
+
+    scope.run(() => {
+      source = ref(0)
+      copyist = ref(0)
+      // sync by default
+      watch(
+        () => {
+          const current = (copyist.value = source.value)
+          onWatcherCleanup(() => calls.push(`sync ${current}`))
+        },
+        null,
+        {},
+      )
+      // with scheduler
+      watch(
+        () => {
+          const current = copyist.value
+          onWatcherCleanup(() => calls.push(`post ${current}`))
+        },
+        null,
+        { scheduler },
+      )
+    })
+
+    await nextTick()
+    expect(calls).toEqual([])
+
+    scope.run(() => source.value++)
+    expect(calls).toEqual(['sync 0'])
+    await nextTick()
+    expect(calls).toEqual(['sync 0', 'post 0'])
+    calls.length = 0
+
+    scope.run(() => source.value++)
+    expect(calls).toEqual(['sync 1'])
+    await nextTick()
+    expect(calls).toEqual(['sync 1', 'post 1'])
+    calls.length = 0
+
+    scope.stop()
+    expect(calls).toEqual(['sync 2', 'post 2'])
+  })
+
   test('once option should be ignored by simple watch', async () => {
     let dummy: any
     const source = ref(0)
diff --git a/packages/reactivity/src/index.ts b/packages/reactivity/src/index.ts
@@ -90,11 +90,8 @@ export {
   traverse,
   onWatcherCleanup,
   WatchErrorCodes,
-  /**
-   * @internal
-   */
-  WatcherEffect,
   type WatchOptions,
+  type WatchScheduler,
   type WatchStopHandle,
   type WatchHandle,
   type WatchEffect,
diff --git a/packages/reactivity/src/watch.ts b/packages/reactivity/src/watch.ts
@@ -48,7 +48,12 @@ export interface WatchOptions<Immediate = boolean> extends DebuggerOptions {
   immediate?: Immediate
   deep?: boolean | number
   once?: boolean
+  scheduler?: WatchScheduler
   onWarn?: (msg: string, ...args: any[]) => void
+  /**
+   * @internal
+   */
+  augmentJob?: (job: (...args: any[]) => void) => void
   /**
    * @internal
    */
@@ -70,6 +75,8 @@ export interface WatchHandle extends WatchStopHandle {
 // initial value for watchers to trigger on undefined initial values
 const INITIAL_WATCHER_VALUE = {}
 
+export type WatchScheduler = (job: () => void, isFirstRun: boolean) => void
+
 let activeWatcher: WatcherEffect | undefined = undefined
 
 /**
@@ -110,7 +117,7 @@ export function onWatcherCleanup(
   }
 }
 
-export class WatcherEffect extends ReactiveEffect {
+class WatcherEffect extends ReactiveEffect {
   forceTrigger: boolean
   isMultiSource: boolean
   oldValue: any
@@ -122,7 +129,8 @@ export class WatcherEffect extends ReactiveEffect {
     public cb?: WatchCallback<any, any> | null | undefined,
     public options: WatchOptions = EMPTY_OBJ,
   ) {
-    const { deep, once, call, onWarn } = options
+    const { immediate, deep, once, scheduler, augmentJob, call, onWarn } =
+      options
 
     let getter: () => any
     let forceTrigger = false
@@ -208,53 +216,79 @@ export class WatcherEffect extends ReactiveEffect {
       ? new Array((source as []).length).fill(INITIAL_WATCHER_VALUE)
       : INITIAL_WATCHER_VALUE
 
+    const job = this.scheduler.bind(this)
+
+    if (augmentJob) {
+      augmentJob(job)
+    }
+
+    if (scheduler) {
+      this.scheduler = () => scheduler(job, false)
+    }
+
     if (__DEV__) {
       this.onTrack = options.onTrack
       this.onTrigger = options.onTrigger
     }
+
+    // initial run
+    if (cb) {
+      if (immediate) {
+        job()
+      } else {
+        this.oldValue = this.run()
+      }
+    } else if (scheduler) {
+      scheduler(job, true)
+    } else {
+      this.run()
+    }
   }
 
   scheduler(): void {
     if (!this.dirty) {
       return
     }
-    const newValue = this.run()
-    if (!this.cb) {
-      return
-    }
-    const { deep, call } = this.options
-    if (
-      deep ||
-      this.forceTrigger ||
-      (this.isMultiSource
-        ? (newValue as any[]).some((v, i) => hasChanged(v, this.oldValue[i]))
-        : hasChanged(newValue, this.oldValue))
-    ) {
-      // cleanup before running cb again
-      if (this.cleanups) {
-        cleanup(this, this.cleanups)
-      }
-      const currentWatcher = activeWatcher
-      activeWatcher = this
-      try {
-        const args = [
-          newValue,
-          // pass undefined as the old value when it's changed for the first time
-          this.oldValue === INITIAL_WATCHER_VALUE
-            ? undefined
-            : this.isMultiSource && this.oldValue[0] === INITIAL_WATCHER_VALUE
-              ? []
-              : this.oldValue,
-          this.boundCleanup,
-        ]
-        call
-          ? call(this.cb, WatchErrorCodes.WATCH_CALLBACK, args)
-          : // @ts-expect-error
-            this.cb(...args)
-        this.oldValue = newValue
-      } finally {
-        activeWatcher = currentWatcher
+    if (this.cb) {
+      // watch(source, cb)
+      const newValue = this.run()
+      const { deep, call } = this.options
+      if (
+        deep ||
+        this.forceTrigger ||
+        (this.isMultiSource
+          ? (newValue as any[]).some((v, i) => hasChanged(v, this.oldValue[i]))
+          : hasChanged(newValue, this.oldValue))
+      ) {
+        // cleanup before running cb again
+        if (this.cleanups) {
+          cleanup(this, this.cleanups)
+        }
+        const currentWatcher = activeWatcher
+        activeWatcher = this
+        try {
+          const args = [
+            newValue,
+            // pass undefined as the old value when it's changed for the first time
+            this.oldValue === INITIAL_WATCHER_VALUE
+              ? undefined
+              : this.isMultiSource && this.oldValue[0] === INITIAL_WATCHER_VALUE
+                ? []
+                : this.oldValue,
+            this.boundCleanup,
+          ]
+          call
+            ? call(this.cb, WatchErrorCodes.WATCH_CALLBACK, args)
+            : // @ts-expect-error
+              this.cb(...args)
+          this.oldValue = newValue
+        } finally {
+          activeWatcher = currentWatcher
+        }
       }
+    } else {
+      // watchEffect
+      this.run()
     }
   }
 }
@@ -284,23 +318,10 @@ export function watch(
   options: WatchOptions = EMPTY_OBJ,
 ): WatchHandle {
   const effect = new WatcherEffect(source, cb, options)
-
-  // initial run
-  if (cb) {
-    if (options.immediate) {
-      effect.scheduler()
-    } else {
-      effect.oldValue = effect.run()
-    }
-  } else {
-    effect.run()
-  }
-
   const stop = effect.stop.bind(effect) as WatchHandle
   stop.pause = effect.pause.bind(effect)
   stop.resume = effect.resume.bind(effect)
   stop.stop = stop
-
   return stop
 }
 
diff --git a/packages/runtime-core/__tests__/apiWatch.spec.ts b/packages/runtime-core/__tests__/apiWatch.spec.ts
@@ -505,54 +505,6 @@ describe('api: watch', () => {
     expect(cleanupWatch).toHaveBeenCalledTimes(2)
   })
 
-  it('nested calls to baseWatch and onWatcherCleanup', async () => {
-    let calls: string[] = []
-    let source: Ref<number>
-    let copyist: Ref<number>
-    const scope = effectScope()
-
-    scope.run(() => {
-      source = ref(0)
-      copyist = ref(0)
-      // sync flush
-      watch(
-        () => {
-          const current = (copyist.value = source.value)
-          onWatcherCleanup(() => calls.push(`sync ${current}`))
-        },
-        null,
-        { flush: 'sync' },
-      )
-      // post flush
-      watch(
-        () => {
-          const current = copyist.value
-          onWatcherCleanup(() => calls.push(`post ${current}`))
-        },
-        null,
-        { flush: 'post' },
-      )
-    })
-
-    await nextTick()
-    expect(calls).toEqual([])
-
-    scope.run(() => source.value++)
-    expect(calls).toEqual(['sync 0'])
-    await nextTick()
-    expect(calls).toEqual(['sync 0', 'post 0'])
-    calls.length = 0
-
-    scope.run(() => source.value++)
-    expect(calls).toEqual(['sync 1'])
-    await nextTick()
-    expect(calls).toEqual(['sync 1', 'post 1'])
-    calls.length = 0
-
-    scope.stop()
-    expect(calls).toEqual(['sync 2', 'post 2'])
-  })
-
   it('flush timing: pre (default)', async () => {
     const count = ref(0)
     const count2 = ref(0)
diff --git a/packages/runtime-core/src/apiWatch.ts b/packages/runtime-core/src/apiWatch.ts
