matching: prepare for REF-AUS: refactor

Author: derhuerst

lib/match.js

@@ -39,6 +39,8 @@ const isProgrammerError = (err) => {
 	return PROGRAMMER_ERRORS.some(Err => err instanceof Err)
 }
 
+const KIND_ISTFAHRT = 'istfahrt'
+
 // todo: DRY with OpenDataVBB/nats-consuming-gtfs-rt-server
 const NATS_JETSTREAM_AUS_ISTFAHRT_STREAM_NAME = `AUS_ISTFAHRT_${MAJOR_VERSION}`
 const NATS_JETSTREAM_GTFSRT_STREAM_NAME = `GTFS_RT_${MAJOR_VERSION}`
@@ -180,77 +182,31 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		logger,
 	})
 
-	const matchVdvAusIstFahrtAndPublishAsGtfsRtTripUpdate = async (vdvAusIstFahrt, msg) => {
-		try {
-			const {
-				item: gtfsRtTripUpdate,
-				isMatched,
-				isCached,
-				matchingTime,
-			} = await matchVdvAusIstFahrtWithGtfs(vdvAusIstFahrt)
-
-			if (isMatched || publishUnmatchedTripUpdates) {
-				const topic = getNatsTopicFromGtfsRtTripUpdate(gtfsRtTripUpdate)
-				const tPublished = Date.now()
+	const publishGtfsRtTripUpdateToNats = (gtfsRtTripUpdate, logCtx) => {
+		const topic = getNatsTopicFromGtfsRtTripUpdate(gtfsRtTripUpdate)
+		const tPublished = Date.now()
 
-				logger.trace({
-					topic,
-					isMatched,
-					isCached,
-					matchingTime,
-					gtfsRtTripUpdate,
-					// todo: log just a slice?
-					vdvAusIstFahrt,
-					natsMsgSeq: msg.seq,
-				}, 'publishing GTFS-RT TripUpdate')
-				natsClient.publish(topic, natsJson.encode(gtfsRtTripUpdate))
-
-				// update NATS metrics
-				{
-					// We slice() to keep the cardinality low in case of a bug.
-					const topic_root = (topic.split('.')[0] || '').slice(0, 7)
-					natsNrOfMessagesSentTotal.inc({
-						topic_root,
-					})
-					natsLatestMessageSentTimestampSeconds.set({
-						topic_root,
-					}, tPublished / 1000)
-				}
-			}
+		logger.trace({
+			...logCtx,
+			topic,
+			gtfsRtTripUpdate,
+		}, 'publishing GTFS-RT TripUpdate')
+		natsClient.publish(topic, natsJson.encode(gtfsRtTripUpdate))
 
-			if (isMatched) {
-				successesTotal.inc({
-					cached: isCached ? '1' : '0',
-				})
-			} else {
-				failuresTotal.inc({
-					cached: isCached ? '1' : '0',
-				})
-			}
-			matchingTimeSeconds.observe({
-				matched: isMatched ? '1' : '0',
-				cached: isCached ? '1' : '0',
-			}, matchingTime / 1000)
-
-			if (!isMatched) {
-				// > Indicate to the JetStream server that processing of the message failed and that the message should not be sent to the consumer again.
-				// https://nats-io.github.io/nats.js/jetstream/interfaces/JsMsg.html#term
-				msg.term()
-			}
-		} catch (err) {
-			if (isProgrammerError(err)) {
-				throw err
-			}
-			logger.warn({
-				err,
-				vdvAusIstFahrt,
-				natsMsgSeq: msg.seq,
-			}, `failed to match trip: ${err.message || (err + '')}`)
-			errorsTotal.inc()
+		// update NATS metrics
+		{
+			// We slice() to keep the cardinality low in case of a bug.
+			const topic_root = (topic.split('.')[0] || '').slice(0, 7)
+			natsNrOfMessagesSentTotal.inc({
+				topic_root,
+			})
+			natsLatestMessageSentTimestampSeconds.set({
+				topic_root,
+			}, tPublished / 1000)
 		}
 	}
 
-	const processAusIstFahrtMsg = async (msg) => {
+	const processVdvNatsMsg = async (msg, kind, kindTitle, processVdvItem) => {
 		const tReceived = Date.now()
 		const {
 			subject,
@@ -267,7 +223,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			streamSeq,
 			redelivered,
 			dataSlice: data.slice(0, 100).toString('utf8'),
-		}, 'processing AUS IstFahrt msg')
+		}, `processing ${kindTitle} msg`)
 
 		// > Indicate to the JetStream server that processing of the message is on going, and that the ack wait timer for the message should be reset preventing a redelivery.
 		// https://nats-io.github.io/nats.js/jetstream/interfaces/JsMsg.html#working
@@ -297,35 +253,67 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			natsMsgSeq.set(seq)
 		}
 
-		let ausIstFahrt = null
+		let vdvItem = null
 		try {
-			ausIstFahrt = msg.json(data)
+			vdvItem = msg.json(data)
 		} catch (err) {
 			serviceLogger.warn({
 				err,
 				subject,
 				seq,
-			}, 'failure decoding AUS IstFahrt msg')
+			}, `failure decoding ${kindTitle} msg`)
 			// We don't nak() here because we don't want it to be redelivered, the message is invalid anyways.
 			return;
 		}
 
 		try {
-			// todo: validate against schema, error-log and abort if invalid
-			await matchVdvAusIstFahrtAndPublishAsGtfsRtTripUpdate(ausIstFahrt, msg)
+			const {
+				isMatched,
+				isCached,
+				matchingTime,
+			} = await processVdvItem(vdvItem, msg)
+
+			if (isMatched) {
+				successesTotal.inc({
+					cached: isCached ? '1' : '0',
+				})
+			} else {
+				failuresTotal.inc({
+					cached: isCached ? '1' : '0',
+				})
+			}
+			matchingTimeSeconds.observe({
+				matched: isMatched ? '1' : '0',
+				cached: isCached ? '1' : '0',
+			}, matchingTime / 1000)
 
 			serviceLogger.trace({
 				subject,
 				seq,
-			}, 'successfully processed AUS IstFahrt msg')
-			msg.ack()
+			}, `successfully processed ${kindTitle} msg`)
+			if (isMatched) {
+				msg.ack()
+			} else {
+				// > Indicate to the JetStream server that processing of the message failed and that the message should not be sent to the consumer again.
+				// https://nats-io.github.io/nats.js/jetstream/interfaces/JsMsg.html#term
+				msg.term()
+			}
 		} catch (err) {
+			if (!isProgrammerError(err)) {
+				logger.warn({
+					err,
+					[kind]: vdvItem,
+					natsMsgSeq: msg.seq,
+				}, `failed to match trip: ${err.message || (err + '')}`)
+				errorsTotal.inc()
+			}
+
 			// We catch all non-programmer errors in order not to abort the message processing (see below).
 			serviceLogger.warn({
 				err,
 				subject,
 				seq,
-			}, 'failure processing AUS IstFahrt msg')
+			}, `failure processing ${kindTitle} msg`)
 			// Explicitly signal to NATS JetStream that this message could not be processed.
 			msg.nak()
 			if (isProgrammerError(err)) {
@@ -334,6 +322,36 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		}
 	}
 
+	const matchVdvAusIstFahrtAndPublishAsGtfsRtTripUpdate = async (vdvAusIstFahrt, msg) => {
+		// todo: validate against schema, error-log and abort if invalid
+		const {
+			item: gtfsRtTripUpdate,
+			isMatched,
+			isCached,
+			matchingTime,
+		} = await matchVdvAusIstFahrtWithGtfs(vdvAusIstFahrt)
+
+		if (isMatched || publishUnmatchedTripUpdates) {
+			publishGtfsRtTripUpdateToNats(gtfsRtTripUpdate, {
+				isMatched,
+				isCached,
+				matchingTime,
+				// todo: log just a slice?
+				vdvAusIstFahrt,
+				natsMsgSeq: msg.seq,
+			})
+		}
+	}
+
+	const processAusIstFahrtMsg = async (msg) => {
+		await processVdvNatsMsg(
+			msg,
+			KIND_ISTFAHRT,
+			'AUS IstFahrt',
+			matchVdvAusIstFahrtAndPublishAsGtfsRtTripUpdate,
+		)
+	}
+
 	{
 		{
 			// query details of the NATS JetStream stream for AUS IstFahrts