matching: process more NATS messages while some still get processed 🐛

iter-tools' asyncBuffer() retains the input order, causing slow matchings to "block" later messages from getting processed, reducing the throughput significantly! async-iterator-concurrent-map's mapConcurrent() doesn't do this.

Author: derhuerst

lib/match.js

@@ -7,10 +7,10 @@ import {
 } from 'prom-client'
 import {
 	asyncConsume,
-	execPipe,
-	asyncMap,
-	asyncBuffer,
 } from 'iter-tools'
+import {
+	mapConcurrent as mapConcurrently,
+} from 'async-iterator-concurrent-map'
 import {createLogger} from './logger.js'
 import {register} from './metrics.js'
 import {
@@ -368,16 +368,13 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		}
 
 		const istFahrtsSub = await istFahrtsConsumer.consume()
-		execPipe(
-			istFahrtsSub,
-
-			// asyncBuffer workaround
-			// see also https://github.com/iter-tools/iter-tools/issues/425#issuecomment-882875848
-			asyncMap(msg => [processAusIstFahrtMsg(msg)]),
-			asyncBuffer(matchConcurrency),
-			asyncMap(([task]) => task),
-
-			asyncConsume,
+		// We're not interested in the values, processAusIstFahrtMsg() publishes by itself.
+		asyncConsume(
+			mapConcurrently(
+				istFahrtsSub[Symbol.asyncIterator](),
+				matchConcurrency,
+				processAusIstFahrtMsg,
+			),
 		).catch(abortWithError)
 
 		// todo: support IstUmlauf – would require adapting vdv-453-nats-adapter

package.json

@@ -20,6 +20,7 @@
 		"node": ">=22"
 	},
 	"dependencies": {
+		"async-iterator-concurrent-map": "^1.0.2",
 		"ioredis": "^5.4.1",
 		"iter-tools": "^7.5.3",
 		"lodash": "^4.17.21",