Simple STOMP client for sending and receiving messages.

stomp.js

#!/usr/bin/env node

/*
 * Simple STOMP client for sending and receiving messages.
 * 
 * Pre-requisites: install stompit package (`npm install stompit`).
 * 
 * Usage:
 *  - Send messages: node stomp.js -h <host> -u <username> -p <password> -q <queue> -s
 *  - Receive messages: node stomp.js -h <host> -u <username> -p <password> -q <queue> -l
 */

const stompit = require("stompit");
const {parseArgs} = require("node:util");
const {exit} = require("process");
const readline = require("readline/promises");

const {values: args} = parseArgs({
	options: {
		host: {
			type: "string",
			short: "h",
		},
		username: {
			type: "string",
			short: "u",
		},
		password: {
			type: "string",
			short: "p",
		},
		send: {
			type: "boolean",
			short: "s",
		},
		listen: {
			type: "boolean",
			short: "l",
		},
		queue: {
			type: "string",
			short: "q",
		},
	},
});

if (!args.host || !args.username || !args.password) {
	console.error("Missing parameters (host, username, password)");
	exit(1);
}
if (!args.send && !args.listen) {
	console.error("Missing required option: send or listen");
	exit(1);
}
if (args.send && args.listen) {
	console.error("Invalid options: send and listen are mutually exclusive");
	exit(1);
}
if (!args.queue) {
	console.error("Missing parameter: queue");
	exit(1);
}

process.on("SIGINT", () => {
	process.exit();
});

const main = async () => {
	const connectOptions = {
		host: args.host,
		port: 61614,
		ssl: true,
		connectHeaders: {
			host: "/",
			login: args.username,
			passcode: args.password,
			"heart-beat": "5000,5000",
		},
	};

	await connect(connectOptions);
};

const connect = async (options) => {
	console.log("🦶 Connecting...");

	stompit.connect(options, async (error, client) => {
		if (error) {
			throw new Error("Connection error: " + error.message);
		}

		console.log("✅ Connected");

		client.on("error", error => {
			if (error.message === "connection timed out") {
				console.log("😵 Connection timed out, reconnecting in 1 second...");
				new Promise(resolve => setTimeout(resolve, 1000)).then(() => {
					connect(options);
				});
			}
		});

		if (args.send) {
			const sendHeaders = {
				destination: `/queue/${args.queue}`,
				"content-type": "text/plain",
			};

			const rl = readline.createInterface(process.stdin, process.stdout);

			rl.on("SIGINT", () => {
				process.exit();
			});

			while (true) {
				const message = (await rl.question("Message: "));
				if (message !== "") {
					const frame = client.send(sendHeaders);
					frame.write(message);
					frame.end();
					console.log("✉️  Sent");
				}
			}
		} else {
			const subscribeHeaders = {
				destination: `/queue/${args.queue}`,
				ack: "auto",
			};

			console.log("👂 Listening for messages...");
			client.subscribe(subscribeHeaders, (error, message) => {
				if (error) {
					if (error.message === "connection timed out") {
						// already handled on the connection level
						return;
					} else {
						throw new Error("Subscribe error: " + error.message);
					}
				}

				message.readString("utf-8", (error, body) => {
					if (error) {
						throw new Error("Read message error: " + error.message);
					}

					console.log("✉️  Received: " + body);
				});
			});
		}
	});
};

main().catch(console.error);