feat(effects): skip mark for take filter

packages/effects/README.md

@@ -33,19 +33,28 @@ Two important notes.
 
 ### take
 
-Allow you to wait an atom update.
+This is the simplest and most powerful API that allows you to wait for an atom update, which is useful for describing certain procedures. It is a shortcut for subscribing to the atom and unsubscribing after the first update. `take` respects the main Reatom abort context and will throw `AbortError` when the abort occurs. This allows you to describe redux-saga-like procedural logic in synchronous code style with native async/await.
 
 ```ts
+import { action } from '@reatom/core'
 import { take } from '@reatom/effects'
 
-const currentCount = ctx.get(countAtom)
-const nextCount = await take(ctx, countAtom)
+export const validateBeforeSubmit = action(async (ctx) => {
+  let errors = validate(ctx.get(formDataAtom))
+
+  while (Object.keys(errors).length) {
+    formDataAtom.errorsAtom(ctx, errors)
+    // wait any field change
+    await take(ctx, formDataAtom)
+    // recheck validation
+    errors = validate(ctx.get(formDataAtom))
+  }
+})
 ```
 
-You could await actions too!
+You can also await actions!
 
 ```ts
-// ~/features/someForm.ts
 import { take } from '@reatom/effects'
 import { onConnect } from '@reatom/hooks'
 import { historyAtom } from '@reatom/npm-history'
@@ -54,7 +63,7 @@ import { confirmModalAtom } from '~/features/modal'
 // some model logic, doesn't matter
 export const formAtom = reatomForm(/* ... */)
 
-onConnect(form, (ctx) => {
+onConnect(formAtom, (ctx) => {
   // "history" docs: https://github.com/remix-run/history/blob/main/docs/blocking-transitions.md
   const unblock = historyAtom.block(ctx, async ({ retry }) => {
     if (!ctx.get(formAtom).isSubmitted && !ctx.get(confirmModalAtom).opened) {
@@ -71,6 +80,36 @@ onConnect(form, (ctx) => {
 })
 ```
 
+#### take filter
+
+You can pass the third argument to map the update to the required format.
+
+```ts
+const input = await take(ctx, onChange, (ctx, event) => event.target.value)
+```
+
+More than that, you can filter unneeded updates by returning the `skip` mark from the first argument of your callback.
+
+```ts
+const input = await take(ctx, onChange, (ctx, event, skip) => {
+  const { value } = event.target
+  return value.length < 6 ? skip : value
+})
+```
+
+The cool feature of this skip mark is that it helps TypeScript understand the correct type of the returned value, which is hard to achieve with the extra "filter" function. If you have a union type, you could receive the needed data with the correct type easily. It just works.
+
+```ts
+const someRequest = reatomRequest<{ data: Data } | { error: string }>()
+```
+
+```ts
+// type-safe destructuring
+const { data } = await take(ctx, someRequest, (ctx, payload, skip) =>
+  'error' in payload ? skip : payload,
+)
+```
+
 ### takeNested
 
 Allow you to wait all dependent effects, event if they was called in the nested async effect.

packages/effects/src/index.test.ts

@@ -119,6 +119,26 @@ test('withAbortableSchedule', async () => {
   ;`👍` //?
 })
 
+test('take filter', async () => {
+  const act = action((ctx, v: number) => ctx.schedule(() => Promise.resolve(v)))
+  const track = mockFn()
+  const ctx = createTestCtx()
+
+  take(ctx, act, (ctx, v, skip) => {
+    return v < 4 ? skip : v.toString()
+  }).then(track)
+  act(ctx, 1)
+  await null
+  act(ctx, 2)
+  act(ctx, 3)
+  await null
+  act(ctx, 4)
+  await sleep()
+  assert.is(track.calls.length, 1)
+  assert.is(track.lastInput(), '4')
+  ;`👍` //?
+})
+
 // test('concurrent', async () => {
 //   class CtxMap<T> {
 //     private map = new WeakMap<AtomProto, T>()

packages/effects/src/index.ts

@@ -155,27 +155,34 @@ export const disposable = (
   })
 }
 
+const skip: unique symbol = Symbol()
+type skip = typeof skip
 export const take = <T extends Atom, Res = AtomReturn<T>>(
   ctx: Ctx & { controller?: AbortController },
   anAtom: T,
-  mapper: Fn<[Ctx, Awaited<AtomReturn<T>>], Res> = (ctx, v: any) => v,
-): Promise<Awaited<Res>> =>
-  new Promise<Awaited<Res>>((res: Fn, rej) => {
-    onCtxAbort(ctx, rej)
+  mapper: Fn<[Ctx, Awaited<AtomReturn<T>>, skip], Res | skip> = (ctx, v: any) =>
+    v,
+): Promise<Awaited<Res>> => {
+  const cleanups: Array<Fn> = []
+  return new Promise<Awaited<Res>>((res: Fn, rej) => {
+    cleanups.push(
+      onCtxAbort(ctx, rej) ?? noop,
+      ctx.subscribe(anAtom, async (state) => {
+        // skip the first sync call
+        if (!cleanups.length) return
 
-    let skipFirst = true,
-      un = ctx.subscribe(anAtom, (state) => {
-        if (skipFirst) return (skipFirst = false)
-        un()
-        if (anAtom.__reatom.isAction) state = state[0].payload
-        if (state instanceof Promise) {
-          state.then((v) => res(mapper(ctx, v)), rej)
-        } else {
-          res(mapper(ctx, state))
+        try {
+          if (anAtom.__reatom.isAction) state = state[0].payload
+          const value = await state
+          const result = mapper(ctx, value, skip)
+          if (result !== skip) res(result)
+        } catch (error) {
+          rej(error)
         }
-      })
-    ctx.schedule(un, -1)
-  })
+      }),
+    )
+  }).finally(() => cleanups.forEach((cb) => cb()))
+}
 
 export const takeNested = <I extends any[]>(
   ctx: Ctx & { controller?: AbortController },