dougal-software/lib/www/server/udp/server.js

#!/usr/bin/node

const dgram = require('dgram');
const cfg = require("../lib/config");
const { navdata } = require('../lib/db');
const sendAlert = require("../lib/alerts");
const headers = require('../lib/headers');
const { ERROR, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);

function maybeSendError(error, payload = {}) {
	const defaults = {
		title: "UDP listener error",
		description: error.message,
		message: error.message,
		error: error
	};

	const packet = Object.assign({}, defaults, payload);
	console.error(packet);
	sendAlert(packet);
}

function parseMessages (buffer, formats = [ "hydronav", "labo", "smartsource" ]) {
	const formatHandlers = formats.map( name => headers[name]);
// 	[ headers.hydronav, headers.labo, headers.smartsource ];

	const navData = {
		tstamp: new Date(),
		sequence: null,
		line: null,
		point: null,
		latitude: null,
		longitude: null,
		lineName: null,
		online: null,
		payload: []
	};

	for (const format of formatHandlers) {
// 		console.log(format.name);
		const pos = format.detect(buffer);
		if (pos !== false) {
			try {
				const header = format.parse(buffer.subarray(pos));
				if (header) {
					if (header.sequence) {
						navData.sequence = header.sequence;
					}
					if (header.line) {
						navData.line = header.line;
					}
					if (header.shotPoint) {
						navData.point = header.shotPoint;
					}
					if (header.latitude) {
						navData.latitude = header.latitude;
					}
					if (header.longitude) {
						navData.longitude = header.longitude;
					}
					if (header.easting) {
						navData.easting = header.easting;
					}
					if (header.northing) {
						navData.northing = header.northing;
					}
					if (header.lineName) {
						navData.lineName = header.lineName;
					}
					if (navData.online === null && header.lineStatus) {
						if (["online", "approach", "runout"].includes(header.lineStatus)) {
							navData.online = true;
						} else if (header.lineStatus == "offline") {
							navData.online = false;
						}
						// else unknown
					}
					if (header.tstamp) {
						navData.tstamp = header.tstamp;
						if (header._received) {
							header._latency = header._received - header.tstamp;
						}
					}
					navData.payload.push(header);
				}
			} catch (error) {
				maybeSendError(error, {title: `Decoding error for format ${format.name}`});
			}
		}
	}
	return navData;
}

INFO("Dougal real-time starting…");

for (const header of (cfg._("global.navigation.headers") || []).filter(h => h.type == "udp")) {
	const server = dgram.createSocket('udp4');

	server.on('error', (err) => {
		ERROR(err);
		// console.error(`server error:\n${err.stack}`);
		maybeSendError(err, {title: "UDP listener error on port "+header.port});
// 		server.close();
	});

	server.on('message', (msg, rinfo) => {
		DEBUG(`${header.type} :${header.port} ← ${msg.length} bytes from ${rinfo.address}:${rinfo.port}`);

		const navData = parseMessages(msg);

		if (navData.payload.length) {
			navData.payload = navData.payload.reduce( (a, b) => Object.assign(a, b), {});
			delete navData.payload._type;
// 			DEBUG("SAVE", JSON.stringify(navData, null, 4));
// 			DEBUG("META", header.meta);
			navdata.save(navData, header.meta);
		}
	});

	server.on('listening', () => {
		const address = server.address();
		INFO(`server listening ${address.address}:${address.port}`);
	});

	server.bind(header.port);
}

INFO("Dougal real-time interface quitting");