ff-worker-pipe

在独立 Web Worker 中执行代码，通过链式调用透明访问主线程注册的函数（HTTP 客户端、数据处理、路由等），并支持将 Worker 侧的回调函数跨线程传回主线程执行。

安装

npm install ff-worker-pipe

工作机制

┌─────────────────────────────────────────────────────────────────────┐
│                          主线程 (Main Thread)                        │
│                                                                     │
│  WorkerPipe.fns = { transform, http, route, … }                    │
│                          ↑ 注册主线程函数                            │
│  pipe.exec(`return Util.transform([1,2,3]).map(x => x*2)`)         │
│                          ↓ 发送 eval 消息                           │
└──────────────────────────┬──────────────────────────────────────────┘
                           │  postMessage
        ┌──────────────────▼──────────────────────┐
        │              Web Worker                  │
        │                                          │
        │  eval → 执行代码字符串                    │
        │  Util.transform([1,2,3])                 │
        │    → util 消息 → 主线程执行               │
        │    ← 返回 { __chain_id__ }               │
        │  .map(x => x*2)                          │
        │    → util 消息 + fn_ref → 主线程          │
        │    → invoke_fn → Worker 执行 x=>x*2     │
        │    ← invoke_fn_result → 组装结果          │
        │  return [2, 4, 6]                        │
        │    → eval 结果 → 主线程 resolve           │
        └──────────────────────────────────────────┘

四类消息协议

消息类型	方向	触发时机
eval	主线程 → Worker	pipe.exec(code) 调用时
util	Worker ↔ 主线程	Worker 侧调用 Util.xxx() 时（双向：请求 + 响应）
invoke_fn	主线程 → Worker	主线程需要执行 Worker 侧传来的回调函数时
clear	主线程 → Worker	pipe.clear(session) 清除工作区时

基本调用流程

sequenceDiagram
    participant MT as 主线程
    participant W as Worker

    MT->>W: ① eval { id, method: "return Util('async').double(5)" }
    Note over W: eval 创建 async 函数并执行
    Note over W: Util('async').double(5)<br/>→ send('async/double', [5])
    W->>MT: ② util { id:0, task_id, method:'async/double', args:[5] }
    Note over MT: handleUtil<br/>→ fns.async.double(5)<br/>→ Promise‹10›
    MT->>W: ③ util { id:0, data:10, error:null }
    Note over W: mQueue 回调，Promise 解析为 10<br/>async fn return 10
    W->>MT: ④ eval { id, data:10, error:null }
    Note over MT: pipe.exec() Promise.resolve(10)

WithCallback 回调流程

sequenceDiagram
    participant MT as 主线程
    participant W as Worker

    MT->>W: ① eval { method: "return Util.transform([1,2,3]).map(x=>x*2)" }

    Note over W: Util.transform([1,2,3])<br/>→ send('transform', [[1,2,3]])
    W->>MT: ② util { method:'transform', args:[[1,2,3]] }
    Note over MT: WithCallback(transform)([1,2,3])<br/>→ { map:wrapFn, filter:wrapFn, … }<br/>→ resolveResult → chainStore[cid]
    MT->>W: ③ util { data:{ __chain_id__:cid } }

    Note over W: .map(x=>x*2)<br/>→ storeArg(fn) → { __fn_id__:'fn_0' }<br/>→ send('map', [fn_ref], cid)
    W->>MT: ④ util { method:'map', args:[{__fn_id__:'fn_0'}], chain_id:cid }
    Note over MT: wrapArg → Proxy fn<br/>wrapFn(map)(proxy) → data.map(proxy)<br/>pending = [{fn:proxy,params:[1]},…]<br/>Promise.all → invokeWorkerFn('fn_0',[1])
    MT->>W: ⑤ invoke_fn { fn_id:'fn_0', cb_id:0, args:[1] }
    Note over W: fnStore.get('fn_0') = x=>x*2<br/>fn(1) = 2
    W->>MT: ⑥ invoke_fn_result { cb_id:0, result:2 }
    Note over MT: 重复 ⑤⑥ for 2→4, 3→6<br/>Promise.all([2,4,6])<br/>占位符替换 → [2,4,6]
    MT->>W: ⑦ util { data:[2,4,6] }

    Note over W: .map() 解析为 [2,4,6]<br/>async fn return [2,4,6]
    W->>MT: ⑧ eval { data:[2,4,6] }
    Note over MT: pipe.exec() Promise.resolve([2,4,6])

---

快速上手

1. 主线程

import { WorkerPipe, WithCallback } from 'ff-worker-pipe/main'

// 注册可供 Worker 侧调用的函数
WorkerPipe.fns = {
  // 普通函数
  getUser: (id) => fetchUser(id),

  // 命名空间
  format: {
    date: (ts) => new Date(ts).toLocaleDateString(),
  },

  // 多层级链式调用（无回调参数，不需要 WithCallback）
  db: {
    query: (table) => ({
      where: (field, value) => ({
        select: (...fields) => fetchFromDB(table, { [field]: value }, fields),
      }),
    }),
  },

  // 含回调参数的函数用 WithCallback 包裹
  transform: WithCallback((data) => ({
    map:    (fn) => data.map(fn),
    filter: (fn) => data.filter(fn),
    sort:   () => ({                      // sort 不接受比较器回调（需同步执行）
      map:    (fn) => [...data].sort((a, b) => a - b).map(fn),
      filter: (fn) => [...data].sort((a, b) => a - b).filter(fn),
    }),
  })),
}

// 初始化（单例，首次调用时传 URL）
const pipe = new WorkerPipe(new URL('./worker.js', import.meta.url), { type: 'module' })

// 执行 Worker 代码
const result = await pipe.exec(`return Util.transform([3, 1, 2]).sort().map(x => x * 10)`)
// → [10, 20, 30]

2. Worker 入口

ff-worker-pipe/worker 打包后即为完整的 Worker 入口文件，直接指向即可：

// worker.js（你自己的 worker 入口）
import 'ff-worker-pipe/worker'
// 文件加载时自动初始化，无需任何额外代码

或在主线程直接指向包内的 worker 文件（开发环境）：

const pipe = new WorkerPipe(
  new URL('../node_modules/ff-worker-pipe/dist/worker.js', import.meta.url)
)

---

Worker 侧调用语法

在 pipe.exec() 的代码字符串中，通过 Util 访问主线程注册的函数：

// 顶层函数
const user = await Util.getUser(1)

// 命名空间
const str = await Util('format').date(Date.now())

// 多层链式（无回调）
const rows = await Util('db').query('users').where('age', 18).select('name', 'email')

// WithCallback 链式（含回调）
const mapped   = await Util.transform([1, 2, 3]).map(x => x * 2)
const filtered = await Util.transform([1, 2, 3]).filter(x => x > 1)
const sorted   = await Util.transform([3, 1, 2]).sort().map(x => x * 10)

// 直接 await（无 .then()）
const { id, name } = await Util('http').get('/api/user')

// .then() 提取字段
const name = await Util('http').get('/api/user').then(d => d.name)

// 顺序 await + 计算
const { value: a } = await Util('http').get('/api/a')
const b = await Util('http').get('/api/b').then(d => d.value)
return a + b

// Promise.all 并发
const [{ n: a }, { n: b }] = await Promise.all([
  Util('http').get('/api/a'),
  Util('http').get('/api/b'),
])
return a * b

// 错误处理
const result = await Util('http').get('/api/data')
  .then(d => d)
  .catch(e => null)

session 工作区

相同 session 的多次 exec() 共享同一个 this（Worker 侧工作区），可在调用间保留状态：

await pipe.exec(`this.count = (this.count || 0) + 1`, {}, {}, 'my-session')
await pipe.exec(`this.count = (this.count || 0) + 1`, {}, {}, 'my-session')
const n = await pipe.exec(`return this.count`, {}, {}, 'my-session')
// n === 2

pipe.clear('my-session')  // 清除工作区

---

WithCallback 详解

对含回调类型参数的函数使用 WithCallback 包裹，不需要修改函数本身的逻辑。

// 原始函数，保持同步写法
const process = (data) => ({
  map:    (fn)       => data.map(fn),
  filter: (fn)       => data.filter(fn),
  reduce: (fn, init) => data.reduce(fn, init),
})

// 注册时包裹
WorkerPipe.fns = {
  process: WithCallback(process),

  // 对象形式，批量包裹
  ...WithCallback({ process, transform }),
}

Worker 侧直接传函数，无需额外处理：

const doubled  = await Util.process([1, 2, 3]).map(x => x * 2)   // [2, 4, 6]
const positive = await Util.process([1, -2, 3]).filter(x => x > 0) // [1, 3]

占位符机制（内部实现）：WithCallback 在主线程侧将 Worker 回调替换为占位符字符串 \x00idx\x00，以占位符调用原始函数得到含占位符的结果，然后并发执行所有 Worker 回调（通过 invoke_fn 往返消息），最后将占位符替换为真实结果。

---

async fn 支持

WorkerPipe.fns 中的函数可以是 async 函数或返回 Promise，Worker 侧无需特殊处理：

WorkerPipe.fns = {
  async: {
    // async 函数
    double: async (x) => x * 2,

    // 显式返回 Promise
    delayed: (x) => Promise.resolve(x * 3),

    // async fn → 返回含方法的对象（同步 + 异步方法均支持）
    build: async (data) => ({
      sum:     () => data.reduce((a, b) => a + b, 0),   // 同步方法
      max:     async () => Math.max(...data),             // async 方法
      doubled: () => data.map(x => x * 2),               // 同步方法
    }),
  },
}

Worker 侧：

// 基础 async
const n = await Util('async').double(5)   // 10

// 链式：async fn → 含方法的对象
const sum = await Util('async').build([1, 2, 3]).sum()   // 6
const max = await Util('async').build([1, 2, 3]).max()   // 3

// Worker 侧 Promise.all
const [a, b] = await Promise.all([
  Util('async').double(3),
  Util('async').double(7),
])
return a + b   // 20

---

自定义可链式类型

对于不可序列化但需要链式调用的对象（如 HTTP 客户端响应），通过 WorkerPipe.chainableTypes 注册：

import { HttpResponse } from 'ff-request'

WorkerPipe.chainableTypes = [HttpResponse]
WorkerPipe.fns = { http: new HttpRequest() }

主线程检测到 instanceof HttpResponse 时，将对象存入 chainStore 而非序列化，Worker 侧可继续调用其方法（.then()、.catch()、.msg() 等）。

---

API

主线程 (ff-worker-pipe/main)

WorkerPipe

成员	类型	说明
fns	static Object	注册可供 Worker 调用的函数树
chainableTypes	static Array	需要存入 chainStore 的类型列表
new WorkerPipe(url, options?)	constructor	创建实例（单例），options 同 new Worker()
WorkerPipe.getInstance(url, options?)	static	获取或创建单例
pipe.exec(source, args?, fns?, session?)	method	在 Worker 中执行代码字符串，返回 Promise
pipe.clear(session)	method	清除指定 session 的 Worker 工作区

WithCallback(fnOrObj)

传入形式	说明
WithCallback(fn)	包裹单个函数
WithCallback({ fn, … })	批量包裹对象中的函数

Worker 侧 (ff-worker-pipe/worker)

打包后直接作为 Worker 入口文件使用，自动初始化。

如需手动控制初始化：

import { createWorkerSide } from 'ff-worker-pipe/worker'

const { Util, onMessage, clearSession } = createWorkerSide(self)
self.addEventListener('message', onMessage)
self.addEventListener('message', ({ data }) => {
  if (data.category === 'clear') clearSession(data.session)
})
self.addEventListener('message', ({ data: { id, method, args, category, session } }) => {
  if (category !== 'eval') return
  const fn = eval(`(async function({${Object.keys(args).join(',')}}, Util){ ${method} })`)
  Promise.resolve(fn.call({}, args, Util(id, session)))
    .then(data => self.postMessage({ id, session, category, data, error: null }))
    .catch(e  => self.postMessage({ id, session, category, data: null, error: String(e) }))
})

---

构建

yarn build   # 输出 dist/main.js（库模块）和 dist/worker.js（Worker 入口）

package.json exports：

{
  "exports": {
    "./main":   { "development": "./src/main.js",   "default": "./dist/main.js" },
    "./worker": { "development": "./src/worker.js", "default": "./dist/worker.js" }
  }
}