Merge branch 'cancellable-flows' into mobx4

CHANGELOG.md

@@ -11,7 +11,9 @@ This is the extensive list of all changes.
 The changes mentioned here are discussed in detail in the [release highlights](https://medium.com/p/c1fbc08008da/), or were simply updated in the docs.
 
 * MobX 4 introduces separation between the production and non production build. The production build strips most typechecks, resulting in a faster and smaller build. Make sure to substitute process.env.NODE_ENV = "production" in your build process! If you are using MobX in a react project, you most probably already have set this up. Otherwise, they idea is explained [here](https://reactjs.org/docs/add-react-to-an-existing-app.html).
-* Introduced `flow` to create a chain of async actions. This is the same function as [`asyncActions`](https://github.com/mobxjs/mobx-utils#asyncaction) of the mobx-utils package. (Note, unlike `asyncAction` and unlike in the beta, it can _not_ be used as decorator)
+* Introduced `flow` to create a chain of async actions. This is the same function as [`asyncActions`](https://github.com/mobxjs/mobx-utils#asyncaction) of the mobx-utils package
+* These `flow`'s are now cancellable, by calling `.cancel()` on the returned promise, which will throw a cancellation exception into the generator function.
+* `flow` also has experimental support for async iterators (`async * function`)
 * Introduced `decorate(thing, decorators)` to decorate classes or object without needing decorator syntax.
 * Introduced `onBecomeObserved` and `onBecomeUnobserved`. These API's enable hooking into the observability system and get notified about when an observable starts / stops becoming used. This is great to automaticaly fetch data from external resources, or stop doing so.
 * `computed` / `@computed` now accepts a `requiresReaction` option. If it set, the computed value will throw an exception if it is being read while not being tracked by some reaction.

src/api/flow.ts

@@ -2,9 +2,12 @@ import { action } from "./action"
 
 let generatorId = 0
 
-// non-decorator forms
-export function flow<R>(generator: () => IterableIterator<any>): () => Promise<R>
-export function flow<A1>(generator: (a1: A1) => IterableIterator<any>): (a1: A1) => Promise<any> // Ideally we want to have R instead of Any, but cannot specify R without specifying A1 etc... 'any' as result is better then not specifying request args
+export type CancellablePromise<T> = Promise<T> & { cancel(): void }
+
+export function flow<R>(generator: () => IterableIterator<any>): () => CancellablePromise<R>
+export function flow<A1>(
+    generator: (a1: A1) => IterableIterator<any>
+): (a1: A1) => CancellablePromise<any> // Ideally we want to have R instead of Any, but cannot specify R without specifying A1 etc... 'any' as result is better then not specifying request args
 export function flow<A1, A2, A3, A4, A5, A6, A7, A8>(
     generator: (
         a1: A1,
@@ -16,22 +19,22 @@ export function flow<A1, A2, A3, A4, A5, A6, A7, A8>(
         a7: A7,
         a8: A8
     ) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6, a7: A7, a8: A8) => Promise<any>
+): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6, a7: A7, a8: A8) => CancellablePromise<any>
 export function flow<A1, A2, A3, A4, A5, A6, A7>(
     generator: (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6, a7: A7) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6, a7: A7) => Promise<any>
+): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6, a7: A7) => CancellablePromise<any>
 export function flow<A1, A2, A3, A4, A5, A6>(
     generator: (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6) => Promise<any>
+): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5, a6: A6) => CancellablePromise<any>
 export function flow<A1, A2, A3, A4, A5>(
     generator: (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5) => Promise<any>
+): (a1: A1, a2: A2, a3: A3, a4: A4, a5: A5) => CancellablePromise<any>
 export function flow<A1, A2, A3, A4>(
     generator: (a1: A1, a2: A2, a3: A3, a4: A4) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3, a4: A4) => Promise<any>
+): (a1: A1, a2: A2, a3: A3, a4: A4) => CancellablePromise<any>
 export function flow<A1, A2, A3>(
     generator: (a1: A1, a2: A2, a3: A3) => IterableIterator<any>
-): (a1: A1, a2: A2, a3: A3) => Promise<any>
+): (a1: A1, a2: A2, a3: A3) => CancellablePromise<any>
 export function flow<A1, A2>(
     generator: (a1: A1, a2: A2) => IterableIterator<any>
 ): (a1: A1, a2: A2) => Promise<any>
@@ -45,13 +48,17 @@ export function flow(generator: Function) {
     return function() {
         const ctx = this
         const args = arguments
-        return new Promise(function(resolve, reject) {
-            const runId = ++generatorId
+        const runId = ++generatorId
+        const gen = action(`${name} - runid: ${runId} - init`, generator).apply(ctx, args)
+        let rejector: (error: any) => void
+        let pendingPromise: CancellablePromise<any> | undefined = undefined
+
+        const res = new Promise(function(resolve, reject) {
             let stepId = 0
-            const gen = action(`${name} - runid: ${runId} - init`, generator).apply(ctx, args)
-            onFulfilled(undefined) // kick off the process
+            rejector = reject
 
             function onFulfilled(res: any) {
+                pendingPromise = undefined
                 let ret
                 try {
                     ret = action(`${name} - runid: ${runId} - yield ${stepId++}`, gen.next).call(
@@ -61,11 +68,12 @@ export function flow(generator: Function) {
                 } catch (e) {
                     return reject(e)
                 }
+
                 next(ret)
-                return null
             }
 
             function onRejected(err: any) {
+                pendingPromise = undefined
                 let ret
                 try {
                     ret = action(`${name} - runid: ${runId} - yield ${stepId++}`, gen.throw).call(
@@ -79,12 +87,29 @@ export function flow(generator: Function) {
             }
 
             function next(ret: any) {
+                if (ret && typeof ret.then === "function") {
+                    // an async iterator
+                    ret.then(next, reject)
+                    return
+                }
                 if (ret.done) return resolve(ret.value)
-                // TODO: support more type of values? See https://github.com/tj/co/blob/249bbdc72da24ae44076afd716349d2089b31c4c/index.js#L100
-                if (!ret.value || typeof ret.value.then !== "function")
-                    fail("Only promises can be yielded to asyncAction, got: " + ret)
-                return ret.value.then(onFulfilled, onRejected)
+                pendingPromise = Promise.resolve(ret.value) as any
+                return pendingPromise!.then(onFulfilled, onRejected)
+            }
+
+            onFulfilled(undefined) // kick off the process
+        }) as any
+
+        res.cancel = action(`${name} - runid: ${runId} - cancel`, function() {
+            try {
+                if (pendingPromise && typeof pendingPromise.cancel === "function")
+                    pendingPromise.cancel()
+                gen.return()
+                rejector(new Error("FLOW_CANCELLED"))
+            } catch (e) {
+                rejector(e) // there could be a throwing finally block
             }
         })
+        return res
     }
 }

test/base/flow.js

@@ -1,4 +1,5 @@
-var mobx = require("../../src/mobx.ts")
+import * as mobx from "../../src/mobx.ts"
+const { flow } = mobx
 
 function delay(time, value, shouldThrow = false) {
     return new Promise((resolve, reject) => {
@@ -157,3 +158,172 @@ function stripEvents(events) {
         return e
     })
 }
+
+test("flows can be cancelled - 1 - uncatched cancellation", done => {
+    let steps = 0
+    const start = flow(function*() {
+        steps = 1
+        yield Promise.resolve()
+        steps = 2
+    })
+
+    const promise = start()
+    promise.then(
+        () => {
+            debugger
+            fail()
+        },
+        err => {
+            debugger
+            expect(steps).toBe(1)
+            expect("" + err).toBe("Error: FLOW_CANCELLED")
+            done()
+        }
+    )
+    debugger
+    promise.cancel()
+})
+
+test("flows can be cancelled - 2 - finally clauses are run", done => {
+    let steps = 0
+    let finallyHandled = false
+    const start = flow(function*() {
+        steps = 1
+        try {
+            yield Promise.resolve()
+            steps = 2
+        } finally {
+            expect(steps).toBe(1)
+            finallyHandled = true
+        }
+    })
+    const promise = start()
+    promise.then(
+        res => {
+            fail()
+        },
+        err => {
+            expect("" + err).toBe("Error: FLOW_CANCELLED")
+            expect(finallyHandled).toBeTruthy()
+            done()
+        }
+    )
+    promise.cancel()
+})
+
+test("flows can be cancelled - 3 - throw in finally should be catched", done => {
+    const counter = mobx.observable({ counter: 0 })
+    const d = mobx.reaction(() => counter.counter, () => {})
+    mobx.configure({ enforceActions: true })
+
+    const start = flow(function*() {
+        counter.counter = 1
+        try {
+            yield Promise.resolve()
+            counter.counter = 15
+        } finally {
+            counter.counter = 4
+            throw "OOPS"
+        }
+    })
+
+    const promise = start()
+    promise.then(
+        () => fail("flow should not have failed"),
+        err => {
+            expect("" + err).toBe("OOPS")
+            expect(counter.counter).toBe(4)
+            mobx.configure({ enforceActions: false })
+            d()
+            done()
+        }
+    )
+    promise.cancel()
+})
+
+test("flows can be cancelled - 4 - pending Promise will be ignored", done => {
+    let steps = 0
+    const start = flow(function*() {
+        steps = 1
+        yield Promise.reject("This won't be catched anywhere!") // cancel will resolve this flow before this one is throw, so this promise goes uncatched
+        steps = 2
+    })
+
+    const promise = start()
+    promise.then(
+        () => fail(),
+        err => {
+            expect(steps).toBe(1)
+            expect("" + err).toBe("Error: FLOW_CANCELLED")
+            done()
+        }
+    )
+    promise.cancel()
+})
+
+test("flows can be cancelled - 5 - return before cancel", done => {
+    let steps = 0
+    const start = flow(function*() {
+        steps = 1
+        return Promise.resolve(2) // cancel will be to late..
+    })
+
+    const promise = start()
+    promise.then(
+        value => {
+            expect(value).toBe(2), done()
+        },
+        err => {
+            fail()
+        }
+    )
+    promise.cancel() // no-op
+})
+
+test("flows can be cancelled - 5 - flows cancel recursively", done => {
+    let flow1cancelled = false
+    let flow2cancelled = false
+    let stepsReached = 0
+
+    const flow1 = flow(function*() {
+        try {
+            yield Promise.resolve()
+            stepsReached++
+        } finally {
+            flow1cancelled = true
+        }
+    })
+
+    const flow2 = flow(function*() {
+        try {
+            yield flow1()
+            stepsReached++
+        } finally {
+            flow2cancelled = true
+        }
+    })
+
+    const p = flow2()
+    p.then(
+        () => fail(),
+        err => {
+            expect("" + err).toBe("Error: FLOW_CANCELLED")
+            expect(stepsReached).toBe(0)
+            expect(flow2cancelled).toBeTruthy()
+            expect(flow1cancelled).toBeTruthy()
+            done()
+        }
+    )
+    p.cancel()
+})
+
+test("flows yield anything", async () => {
+    let steps = 0
+    const start = flow(function*() {
+        const x = yield 2
+        return x
+    })
+
+    const res = await start()
+    expect(res).toBe(2)
+})

test/base/typescript-tests.ts

@@ -1578,3 +1578,59 @@ test("it should support asyncAction as decorator (ts)", async () => {
 
     expect(await x.f(3)).toBe(16)
 })
+
+test("flow support async generators", async () => {
+    ;(Symbol as any).asyncIterator =
+        (Symbol as any).asyncIterator || Symbol.for("Symbol.asyncIterator")
+
+    async function* someNumbers() {
+        await Promise.resolve()
+        yield 1
+        await Promise.resolve()
+        yield 2
+        await Promise.resolve()
+        yield 3
+    }
+
+    let steps = 0
+    const start = mobx.flow(async function*() {
+        let total = 0
+        for await (const number of someNumbers()) {
+            total += number
+        }
+        return total
+    })
+
+    const p = start()
+    const res = await p
+    expect(res).toBe(6)
+})
+
+test.only("flow support throwing async generators", async () => {
+    ;(Symbol as any).asyncIterator =
+        (Symbol as any).asyncIterator || Symbol.for("Symbol.asyncIterator")
+
+    async function* someNumbers() {
+        await Promise.resolve()
+        yield 1
+        await Promise.resolve()
+        throw "OOPS"
+    }
+
+    let steps = 0
+    const start = mobx.flow(async function*() {
+        let total = 0
+        for await (const number of someNumbers()) {
+            total += number
+        }
+        return total
+    })
+
+    const p = start()
+    try {
+        await p
+        fail()
+    } catch (e) {
+        expect("" + e).toBe("OOPS")
+    }
+})

test/tsconfig.json

@@ -2,9 +2,11 @@
     "compilerOptions": {
         "module": "commonjs",
         "target": "es5",
+        "lib": ["esnext"],
         "noImplicitAny": true,
         "sourceMap": false,
         "experimentalDecorators": true,
-        "strict":true
+        "strict":true,
+        "downlevelIteration": true
     }
 }

tsconfig.json

@@ -13,6 +13,7 @@
         "noUnusedLocals": true,
         "noImplicitAny": false,
         "moduleResolution": "node",
+        "downlevelIteration": true,
         "lib": [
             "es6"
         ]