add /events route: monitor events in a bounding box

examples/vbb.js

@@ -18,6 +18,7 @@ const config = {
 	docsLink: 'http://example.org/docs',
 	logging: true,
 	aboutPage: true,
+	events: true,
 	healthCheck: async () => {
 		const stop = await hafas.stop('900000100001')
 		return !!stop

index.js

@@ -17,6 +17,7 @@ const defaultConfig = {
 	aboutPage: true,
 	logging: false,
 	healthCheck: null,
+	events: false,
 	addHafasOpts: () => {}
 }
 
@@ -51,7 +52,9 @@ const createApi = (hafas, config, attachMiddleware) => {
 	if ('docsLink' in config) assertNonEmptyString(config, 'docsLink')
 
 	const api = express()
-	api.locals.logger = pino()
+	api.locals.logger = pino({
+		level: process.env.LOGGING_LEVEL || 'info'
+	})
 
 	if (config.cors) {
 		const cors = createCors()

package.json

@@ -34,6 +34,8 @@
 		"date-fns": "^2.12.0",
 		"express": "^4.16.2",
 		"express-pino-logger": "^4.0.0",
+		"hafas-monitor-trips": "^3.0.0",
+		"hafas-monitor-trips-server": "^1.0.0",
 		"hsts": "^2.1.0",
 		"http-link-header": "^1.0.2",
 		"lodash": "^4.17.15",
@@ -43,6 +45,7 @@
 		"parse-human-relative-time": "^2.0.2",
 		"pino": "^5.11.3",
 		"shorthash": "0.0.2",
+		"ssestream": "^1.0.1",
 		"stringify-entities": "^3.0.0"
 	},
 	"devDependencies": {

readme.md

@@ -56,6 +56,7 @@ key | description | mandatory? | default value
 `aboutPage` | Enable the about page on `GET /`? | ✗ | `true`
 `description` | Used for the about page. | ✗ | –
 `docsLink` | Used for the about page. | ✗ | –
+`events` | Enable the `/events` route. | ✗ | `false`
 `addHafasOpts` | Computes additional `hafas-client` opts. `(opt, hafasClientMethod, httpReq) => additionaOpts` | ✗ | –
 
 *Pro Tip:* Use [`hafas-client-health-check`](https://github.com/public-transport/hafas-client-health-check) for `config.healthCheck`.

routes/events.js

@@ -0,0 +1,106 @@
+'use strict'
+
+const createServer = require('hafas-monitor-trips-server')
+const createMonitor = require('hafas-monitor-trips')
+const SSE = require('ssestream').default
+const throttle = require('lodash/throttle')
+const {parseNumber} = require('../lib/parse')
+
+const INTERVAL = parseInt(process.env.MONITOR_INTERVAL || 60) * 1000
+const EVENT_STREAM = 'text/event-stream'
+
+const EVENTS = [
+	'trip', 'new-trip', 'trip-obsolete',
+	'stopover',
+	'position',
+	// todo: stats?
+]
+
+const err400 = (msg) => {
+	const err = new Error(msg)
+	err.statusCode = 400
+	return err
+}
+
+const sseWriter = (req, res, evNames) => {
+	const sse = new SSE(req)
+	sse.pipe(res)
+
+	const createHandler = (evName) => {
+		const sendEvent = (data) => {
+			sse.write({event: evName, data})
+		}
+		return [evName, sendEvent]
+	}
+	return [
+		...evNames.map(createHandler),
+		['error', (err) => {
+			sse.write({event: 'error', data: err.message || (err + '')})
+		}]
+	]
+}
+
+const createEventsRoute = (hafas, config) => {
+	const monitor = createServer(hafas, {
+		createMonitor: (hafas, bbox) => {
+			return createMonitor(hafas, bbox, INTERVAL)
+		}
+	})
+
+	const events = (req, res, next) => {
+		const {logger} = req.app.locals
+
+		const bbox = {}
+		for (const param of ['north', 'west', 'south', 'east']) {
+			if (param in req.query) {
+				bbox[param] = parseNumber(param, req.query[param])
+			} else {
+				return next(err400(`missing ${param} query parameter`))
+			}
+		}
+
+		const events = EVENTS.filter(evName => req.query[evName] === 'true')
+		if (events.length === 0) return next(err400('0 events selected'))
+
+		if (req.accepts(EVENT_STREAM) !== EVENT_STREAM) {
+			return next(err400(`only ${EVENT_STREAM}`))
+		}
+
+		const handlers = sseWriter(req, res, events)
+		const sub = monitor.subscribe(bbox)
+		handlers
+		.filter(([evName]) => evName !== 'error')
+		.forEach(([evName, handler]) => {
+			sub.on(evName, (data) => {
+				logger.debug({event: evName, data, bbox})
+				handler(data)
+			})
+		})
+
+		logger.info({bbox}, 'starting monitor')
+		const logInfos = throttle((stats) => {
+			logger.info({...stats, bbox})
+		}, 10 * 1000)
+		sub.on('stats', logInfos)
+		res.once('close', () => {
+			logger.info({bbox}, 'stopping monitor')
+			sub.removeListener('stats', logInfos)
+			sub.destroy()
+		})
+
+		const onError = handlers.find(([evName]) => evName === 'error')[1]
+		sub.on('error', (err) => {
+			logger.error({bbox}, err)
+			onError(err)
+		})
+	}
+
+	events.cache = false
+	events.queryParameters = [
+		...EVENTS,
+		'north', 'west', 'south', 'east',
+	]
+	return events
+}
+
+module.exports = createEventsRoute

routes/index.js

@@ -24,6 +24,10 @@ const getRoutes = (hafas, config) => {
 		routes['/trips/:id'] = trip(hafas, config)
 	}
 	routes['/locations'] = locations(hafas, config)
+	if (config.events) {
+		const events = require('./events')
+		routes['/events'] = events(hafas, config)
+	}
 	if (hafas.radar) {
 		const radar = require('./radar')
 		routes['/radar'] = radar(hafas, config)