moleculerjs · icebob · Sep 9, 2023 · Aug 21, 2023 · Sep 9, 2023
diff --git a/examples/sse/assets/index.html b/examples/sse/assets/index.html
@@ -0,0 +1,37 @@
+<!DOCTYPE html>
+<html lang="en">
+	<head>
+		<meta charset="UTF-8" />
+		<meta http-equiv="X-UA-Compatible" content="IE=edge" />
+		<meta name="viewport" content="width=device-width, initial-scale=1.0" />
+		<link rel="stylesheet" href="style.css" />
+		<title>moleculer-web sse</title>
+	</head>
+	<body>
+		<body>
+			<div class="chat-box">
+				<div class="chat-title">Simple Chat</div>
+				<input type="text" class="user-name" placeholder="Your Name" />
+				<div class="messages" id="messages1"></div>
+				<input
+					type="text"
+					class="message"
+					placeholder="Type your message"
+				/>
+				<button class="send-button">Send</button>
+			</div>
+			<div class="chat-box">
+				<div class="chat-title">Simple Chat</div>
+				<input type="text" class="user-name" placeholder="Your Name" />
+				<div class="messages" id="messages2"></div>
+				<input
+					type="text"
+					class="message"
+					placeholder="Type your message"
+				/>
+				<button class="send-button">Send</button>
+			</div>
+			<script src="index.js"></script>
+		</body>
+	</body>
+</html>
diff --git a/examples/sse/assets/index.js b/examples/sse/assets/index.js
@@ -0,0 +1,56 @@
+/* eslint-disable no-undef */
+document.addEventListener("DOMContentLoaded", () => {
+	const messages1 = document.getElementById("messages1");
+	const messages2 = document.getElementById("messages2");
+
+	const sendMessage = async (userName, message) => {
+		return fetch("/api/chat/message", {
+			method: "POST",
+			headers: {
+				"Content-Type": "application/json",
+			},
+			body: JSON.stringify({
+				user: userName,
+				message: message,
+			}),
+		});
+	};
+
+	const addMessage = (element, message) => {
+		const messageElement = document.createElement("div");
+		messageElement.textContent = message;
+		element.appendChild(messageElement);
+		element.scrollTop = element.scrollHeight;
+	};
+
+	const setupSSE = (element) => {
+		const eventSource = new EventSource("/api/chat/message");
+
+		eventSource.addEventListener("chat.message", (event) => {
+			const data = JSON.parse(event.data);
+			addMessage(element, `${data.user}: ${data.message}`);
+		});
+
+		eventSource.addEventListener("error", (error) => {
+			console.error(error);
+		});
+	};
+
+	document.querySelectorAll(".send-button").forEach((button) => {
+		button.addEventListener("click", () => {
+			const chatBox = button.closest(".chat-box");
+			const userNameInput = chatBox.querySelector(".user-name");
+			const messageInput = chatBox.querySelector(".message");
+			const userName = userNameInput.value.trim();
+			const message = messageInput.value.trim();
+
+			if (userName && message) {
+				sendMessage(userName, message);
+				messageInput.value = "";
+			}
+		});
+	});
+
+	setupSSE(messages1);
+	setupSSE(messages2);
+});
diff --git a/examples/sse/assets/style.css b/examples/sse/assets/style.css
@@ -0,0 +1,65 @@
+body {
+	font-family: "Roboto", sans-serif;
+	margin: 0;
+	display: flex;
+	justify-content: center;
+	align-items: center;
+	min-height: 100vh;
+	background-color: #f0f0f0;
+}
+
+.chat-box {
+	display: flex;
+	flex-direction: column;
+	background-color: #ffffff;
+	box-shadow: 0px 3px 6px rgba(0, 0, 0, 0.1);
+	border-radius: 4px;
+	width: 300px;
+	padding: 16px;
+	margin: 16px;
+	gap: 20px;
+}
+
+.messages {
+	background-color: #f0f0f0;
+	border-radius: 4px;
+	height: 300px;
+	overflow-y: scroll;
+	padding: 8px;
+	margin-top: 8px;
+	box-shadow: inset 0px -3px 6px rgba(0, 0, 0, 0.1);
+}
+
+.message {
+	padding: 4px;
+	margin: 4px 0;
+	border-radius: 4px;
+}
+
+.user-input {
+	display: flex;
+	margin-top: 8px;
+}
+
+.user-name,
+.message {
+	flex-grow: 1;
+	border: none;
+	padding: 8px;
+	border-radius: 4px;
+	box-shadow: 0px 2px 3px rgba(0, 0, 0, 0.1);
+}
+
+.send-button {
+	background-color: #1976d2;
+	color: #ffffff;
+	border: none;
+	border-radius: 4px;
+	padding: 8px 16px;
+	cursor: pointer;
+	transition: background-color 0.3s;
+}
+
+.send-button:hover {
+	background-color: #1565c0;
+}
diff --git a/examples/sse/chat.service.js b/examples/sse/chat.service.js
@@ -0,0 +1,15 @@
+module.exports = {
+	name: "chat",
+	actions: {
+		postMessage: {
+			params: {
+				message: "string",
+				user: "string",
+			},
+			handler(context) {
+				const { params } = context;
+				context.emit("chat.sse.message", params);
+			},
+		},
+	},
+};
diff --git a/examples/sse/index.js b/examples/sse/index.js
@@ -0,0 +1,115 @@
+"use strict";
+
+const path = require("path");
+const { ServiceBroker, Errors } = require("moleculer");
+const ApiGatewayService = require("../../index");
+const ChatService = require("./chat.service");
+
+const { MoleculerError } = Errors;
+
+const SSE_RETRY_TIMEOUT = 15000; // 15 seconds
+const PORT = 3000;
+const HOST = "0.0.0.0";
+const SSE_HEADERS = {
+	Connection: "keep-alive",
+	"Content-Type": "text/event-stream",
+	"Cache-Control": "no-cache",
+};
+
+// Create broker
+const broker = new ServiceBroker({
+	logger: console,
+	metrics: true,
+	validation: true,
+});
+
+broker.createService(ChatService);
+
+// Load API Gateway
+broker.createService({
+	name: "sse.gateway",
+	mixins: [ApiGatewayService],
+	settings: {
+		port: PORT,
+
+		ip: HOST,
+
+		assets: {
+			folder: path.join(__dirname, "assets"),
+		},
+
+		routes: [
+			{
+				path: "/api/chat",
+				aliases: {
+					"POST message": "chat.postMessage",
+					"GET message"(request, response) {
+						response.writeHead(200, SSE_HEADERS);
+						response.$service.addSSEListener(
+							response,
+							"chat.message"
+						);
+					},
+				},
+			},
+		],
+	},
+
+	events: {
+		"chat.sse*"(context) {
+			this.handleSSE(context);
+		},
+	},
+
+	methods: {
+		handleSSE(context) {
+			const { eventName, params } = context;
+			const event = eventName.replace("sse.", "");
+			if (!this.sseListeners.has(event)) return;
+			const listeners = this.sseListeners.get(event);
+			for (const listener of listeners.values()) {
+				const id = this.sseIds.get(listener) || 0;
+				const message = this.createSSEMessage(params, event, id);
+				listener.write(message);
+				this.sseIds.set(listener, id + 1);
+			}
+		},
+
+		addSSEListener(stream, event) {
+			if (!stream.write)
+				throw new MoleculerError("Only writable can listen to SSE.");
+			const listeners = this.sseListeners.get(event) || new Set();
+			listeners.add(stream);
+			this.sseListeners.set(event, listeners);
+			this.sseIds.set(stream, 0);
+			stream.on("close", () => {
+				this.sseIds.delete(stream);
+				listeners.delete(stream);
+			});
+		},
+
+		createSSEMessage(data, event, id) {
+			return `event: ${event}\ndata: ${JSON.stringify(
+				data
+			)}\nid: ${id}\nretry: ${this.sseRetry}\n\n`;
+		},
+	},
+
+	started() {
+		this.sseListeners = new Map();
+		this.sseIds = new WeakMap();
+		this.sseRetry = SSE_RETRY_TIMEOUT;
+	},
+
+	stopped() {
+		for (const listeners of this.sseListeners.values()) {
+			for (const listener of listeners.values()) {
+				if (listener.end) listener.end();
+				listeners.delete(listener);
+			}
+		}
+	},
+});
+
+// Start server
+broker.start();