feat: Fetch with limited concurrency

config/config.schema.json

@@ -1,6 +1,6 @@
 {
     "$schema": "https://json-schema.org/draft-07/schema",
-    "$id": "config.schema.json-v0.3.1",
+    "$id": "config.schema.json-v0.4.0",
     "title": "Bulletin IT 设置",
     "type": "object",
     "properties": {
@@ -20,6 +20,23 @@
             "type": "number",
             "default": 90
         },
+        "fetch": {
+            "title": "如何获取通知",
+            "type": "object",
+            "properties": {
+                "concurrency": {
+                    "title": "并发数量上限",
+                    "description": "必须是正整数。",
+                    "type": "number",
+                    "default": 5
+                },
+                "sleep": {
+                    "title": "获取通知后等待的毫秒数",
+                    "type": "number",
+                    "default": 0
+                }
+            }
+        },
         "ding": {
             "title": "钉钉插件",
             "type": "object",

package.json

@@ -28,6 +28,7 @@
     "cli-progress": "^3.9.1",
     "jsdom": "^19.0.0",
     "node-fetch": "^3.2.10",
+    "p-map": "^6.0.0",
     "virtual-bit-network": "github:YDX-2147483647/virtual-BIT-network",
     "winston": "^3.8.1",
     "xml": "^1.0.1",

src/core/config.ts

@@ -11,18 +11,27 @@ interface Config {
     sources_by_selectors: string
     json_path: string
     save_for: number
+    fetch: {
+        concurrency: number
+        sleep: number
+    }
     [propName: string]: any
 }
 
 const defaults: Config = {
     sources_by_selectors: 'config/sources_by_selectors.json',
     json_path: 'output/notices.json',
     save_for: 90,
+    fetch: {
+        concurrency: 5,
+        sleep: 0,
+    },
 }
 
 async function _import_config ({ config_path = 'config/config.yml' } = {}): Promise<Config> {
     const file = await readFile(config_path)
     const given = parse(file.toString())
+    // todo: 目前只支持单层覆盖
     return Object.assign({}, defaults, given) as Config
 }
 

src/core/index.ts

@@ -19,5 +19,7 @@ export function update_notices () {
         write_json_path: config.json_path,
         sources_by_selectors_path: config.sources_by_selectors,
         save_for: config.save_for,
+        fetch_concurrency: config.fetch.concurrency,
+        fetch_sleep: config.fetch.sleep,
     })
 }

src/core/notices/fetcher.ts

@@ -1,20 +1,26 @@
+import { setTimeout } from 'node:timers/promises'
+import pMap from 'p-map'
+
 import type { HookCollectionType } from '../hooks_type.js'
 import type { Notice, Source } from '../models.js'
 
 /**
  * 从一系列来源获取通知
  * @param options 选项
  * @param options.sources
+ * @param options.concurrency
  * @param options._hook (internal usage only) `fetch`, `fetch_each`
  *
  * Here there are 2 hooks: `fetch` and `fetch_each`.
  * - `fetch`: the whole process, fetching all of the sources.
  * - `fetch_each`: several parallel sub-processes, fetching each source.
  */
 export async function fetch_all_sources ({
-    sources, _hook,
+    sources, concurrency, sleep, _hook,
 }: {
     sources: Source[],
+    concurrency: number,
+    sleep: number,
     _hook: HookCollectionType,
 }): Promise<{ notices: Notice[] }> {
     return await _hook(
@@ -24,15 +30,15 @@ export async function fetch_all_sources ({
             // First create a non-hook version.
             async function fetch_each ({ source }: { source: Source }): Promise<{ notices: Notice[] }> {
                 const notices = await source.fetch_notice({ _hook })
+                await setTimeout(sleep)
                 return { notices }
             }
             // Then wrap it with the hook.
             function fetch_each_hooked (s: Source): Promise<{ notices: Notice[] }> {
                 return _hook('fetch_each', fetch_each, { source: s, ...options })
             }
             // Call `fetch_each` in parallel.
-            const notices_grouped = await Promise.all(
-                sources.map(fetch_each_hooked))
+            const notices_grouped = await pMap(sources, fetch_each_hooked, { concurrency })
 
             // Ignore `undefined`.
             // (If `fetch_each` has an error hook, we may get here even though there's nothing.)

src/core/update_notices.ts

@@ -13,15 +13,19 @@ type UpdateNoticesOptions = {
     write_json_path: string
     sources_by_selectors_path: string
     save_for: number
+    fetch_concurrency: number
+    fetch_sleep: number
 }
 
 async function _update_notices ({
-    sources_by_selectors_path, read_json_path, write_json_path, save_for,
+    sources_by_selectors_path, read_json_path, write_json_path, save_for, fetch_concurrency, fetch_sleep,
     _hook, ...options
 }: { _hook: HookCollectionType } & UpdateNoticesOptions) {
     const sources = await import_sources({ sources_by_selectors_path })
     const { notices: latest_notices } = await fetch_all_sources({
         sources,
+        concurrency: fetch_concurrency,
+        sleep: fetch_sleep,
         _hook,
         ...options,
     })