feat: simplify journal connection via WebSocket upgrade
Refactor the journal server to run over WebSockets on the existing HTTP server instead of a separate MQTT TCP port. This removes the need for a separate port configuration and simplifies the connection flow by auto-connecting to the current host. - Remove `--mqtt-port` CLI option - Attach Aedes broker to the HTTP server via WebSocket upgrade - Update UI to remove manual broker URL and player name inputs - Add "Content Source" access from the sidebar - Update JournalPanel to auto-connect to the local server
This commit is contained in:
@@ -210,7 +210,6 @@ export function createContentServer(
|
||||
port: number,
|
||||
distPath: string = distDir,
|
||||
host: string = "0.0.0.0",
|
||||
mqttPort: number = 1883,
|
||||
): ContentServer {
|
||||
let contentIndex: ContentIndex = {};
|
||||
|
||||
@@ -275,28 +274,24 @@ export function createContentServer(
|
||||
// ---- Journal / MQTT broker ----
|
||||
let journal: Awaited<ReturnType<typeof createJournalServer>> | null = null;
|
||||
|
||||
createJournalServer(contentDir, mqttPort).then((j) => {
|
||||
journal = j;
|
||||
});
|
||||
|
||||
// 创建请求处理器
|
||||
// 创建 HTTP 服务器 BEFORE journal so we can pass it in
|
||||
const handleRequest = createRequestHandler(
|
||||
contentDir,
|
||||
distPath,
|
||||
() => contentIndex,
|
||||
);
|
||||
|
||||
// 创建 HTTP 服务器
|
||||
const server = createServer(handleRequest);
|
||||
|
||||
createJournalServer(contentDir, server).then((j) => {
|
||||
journal = j;
|
||||
});
|
||||
|
||||
server.listen(port, host, () => {
|
||||
const displayHost = host === "0.0.0.0" ? "localhost" : host;
|
||||
console.log(`\n开发服务器已启动：http://${displayHost}:${port}`);
|
||||
console.log(`内容目录：${contentDir}`);
|
||||
console.log(`静态资源目录：${distPath}`);
|
||||
console.log(
|
||||
`MQTT 代理端口：${mqttPort} (connect via tcp://${displayHost}:${mqttPort})\n`,
|
||||
);
|
||||
console.log(`Journal 连接：ws://${displayHost}:${port}\n`);
|
||||
});
|
||||
|
||||
return {
|
||||
@@ -319,7 +314,6 @@ export const serveCommand: ServeCommandHandler = async (dir, options) => {
|
||||
const contentDir = resolve(dir);
|
||||
const port = parseInt(options.port, 10);
|
||||
const host = options.host || "0.0.0.0";
|
||||
const mqttPort = parseInt(options.mqttPort || "1883", 10);
|
||||
|
||||
createContentServer(contentDir, port, distDir, host, mqttPort);
|
||||
createContentServer(contentDir, port, distDir, host);
|
||||
};
|
||||
|
||||
@@ -17,7 +17,6 @@ program
|
||||
.description("运行一个 web 服务器预览目录中的内容，并实时监听更新")
|
||||
.argument("[dir]", "要预览的目录", ".")
|
||||
.option("-p, --port <port>", "HTTP 端口号", "3000")
|
||||
.option("--mqtt-port <port>", "MQTT 代理端口号", "1883")
|
||||
.option("-h, --host <host>", "主机地址", "0.0.0.0")
|
||||
.action(serveCommand);
|
||||
|
||||
|
||||
+68
-79
@@ -1,11 +1,11 @@
|
||||
/**
|
||||
* Journal stream persistence — MQTT broker + JSONL append + session manifest
|
||||
*
|
||||
* Starts an aedes MQTT broker and connects to itself to:
|
||||
* 1. Persist all stream messages to .ttrpg/sessions/{id}/stream.jsonl
|
||||
* 2. Manage session lifecycle via retained meta topics
|
||||
* Runs an aedes MQTT broker over WebSocket, attached to the existing
|
||||
* HTTP server via the upgrade event. No separate TCP port.
|
||||
*
|
||||
* No HTTP routes — the JSONL files are served by the existing static server.
|
||||
* Persistence uses aedes's internal subscribe/publish API, so the
|
||||
* server process doesn't need to connect to itself as an MQTT client.
|
||||
*/
|
||||
|
||||
import { join, resolve } from "path";
|
||||
@@ -18,6 +18,8 @@ import {
|
||||
unlinkSync,
|
||||
writeFileSync,
|
||||
} from "fs";
|
||||
import type { Server as HttpServer } from "http";
|
||||
import type { AedesPublishPacket } from "aedes";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
@@ -34,13 +36,8 @@ interface SessionManifest {
|
||||
}
|
||||
|
||||
export interface JournalServer {
|
||||
/** MQTT broker instance */
|
||||
broker: import("aedes").Aedes;
|
||||
/** TCP server the broker is attached to */
|
||||
tcpServer: import("net").Server;
|
||||
/** Persistence client (server subscribes to itself) */
|
||||
persistenceClient: import("mqtt").MqttClient | null;
|
||||
/** Clean shutdown */
|
||||
wsServer: import("ws").WebSocketServer;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -66,10 +63,6 @@ function streamPath(dataRoot: string, id: string): string {
|
||||
return join(sessionDir(dataRoot, id), "stream.jsonl");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Manifest
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function loadManifest(dataRoot: string): SessionManifest {
|
||||
const p = manifestPath(dataRoot);
|
||||
if (!existsSync(p)) return { sessions: {} };
|
||||
@@ -84,10 +77,6 @@ function saveManifest(dataRoot: string, m: SessionManifest): void {
|
||||
writeFileSync(manifestPath(dataRoot), JSON.stringify(m, null, 2), "utf-8");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// JSONL persistence
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function appendStream(dataRoot: string, id: string, line: string): void {
|
||||
const sd = sessionDir(dataRoot, id);
|
||||
mkdirSync(sd, { recursive: true });
|
||||
@@ -98,114 +87,109 @@ function appendStream(dataRoot: string, id: string, line: string): void {
|
||||
// Server factory
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const $SESSIONS = "ttrpg/$SESSIONS";
|
||||
|
||||
export async function createJournalServer(
|
||||
contentDir: string,
|
||||
mqttPort: number,
|
||||
httpServer: HttpServer,
|
||||
): Promise<JournalServer> {
|
||||
const root = dataDir(contentDir);
|
||||
console.log(`[journal] data dir: ${root}`);
|
||||
|
||||
// ---- MQTT Broker ----
|
||||
const { Aedes: AedesFactory } = await import("aedes");
|
||||
const { createServer } = await import("net");
|
||||
|
||||
const broker = new AedesFactory();
|
||||
const tcpServer = createServer(broker.handle.bind(broker));
|
||||
|
||||
tcpServer.listen(mqttPort, "0.0.0.0", () => {
|
||||
console.log(`[journal] MQTT broker on port ${mqttPort}`);
|
||||
});
|
||||
// ---- WebSocket server (attached to HTTP server) ----
|
||||
const { WebSocketServer } = await import("ws");
|
||||
const wsServer = new WebSocketServer({ noServer: true });
|
||||
|
||||
// ---- Persistence + Manifest client ----
|
||||
let client: import("mqtt").MqttClient | null = null;
|
||||
|
||||
const { default: mqtt } = await import("mqtt");
|
||||
client = mqtt.connect(`mqtt://127.0.0.1:${mqttPort}`, {
|
||||
clientId: "journal-server",
|
||||
});
|
||||
|
||||
client.on("connect", () => {
|
||||
// 1. Persistence: append every stream message to JSONL
|
||||
client!.subscribe("ttrpg/+/stream", { qos: 1 }, (err) => {
|
||||
if (err) console.error("[journal] stream sub err:", err);
|
||||
});
|
||||
|
||||
// 2. Session meta: manage manifest from retained meta topics
|
||||
client!.subscribe("ttrpg/+/meta", { qos: 1 }, (err) => {
|
||||
if (err) console.error("[journal] meta sub err:", err);
|
||||
httpServer.on("upgrade", (req, socket, head) => {
|
||||
wsServer.handleUpgrade(req, socket, head, (ws) => {
|
||||
broker.handle(ws as unknown as import("net").Socket);
|
||||
});
|
||||
});
|
||||
|
||||
client.on("message", (topic, payload) => {
|
||||
const parts = topic.split("/");
|
||||
if (parts.length < 3) return;
|
||||
const sessionId = parts[1];
|
||||
const subtopic = parts[2];
|
||||
|
||||
if (subtopic === "stream") {
|
||||
// Persist every stream message
|
||||
try {
|
||||
appendStream(root, sessionId, payload.toString());
|
||||
} catch (e) {
|
||||
console.error(`[journal] stream append err for ${sessionId}:`, e);
|
||||
// ---- Persistence (internal subscribe, no separate MQTT client) ----
|
||||
broker.subscribe(
|
||||
"ttrpg/+/stream",
|
||||
(packet, cb) => {
|
||||
const sessionId = extractSessionId(packet.topic);
|
||||
if (sessionId) {
|
||||
try {
|
||||
appendStream(root, sessionId, packet.payload.toString());
|
||||
} catch (e) {
|
||||
console.error(`[journal] stream append err for ${sessionId}:`, e);
|
||||
}
|
||||
}
|
||||
} else if (subtopic === "meta") {
|
||||
// Session lifecycle
|
||||
handleMetaChange(root, sessionId, client!, payload.toString());
|
||||
}
|
||||
});
|
||||
cb();
|
||||
},
|
||||
() => {},
|
||||
);
|
||||
|
||||
broker.subscribe(
|
||||
"ttrpg/+/meta",
|
||||
(packet, cb) => {
|
||||
const sessionId = extractSessionId(packet.topic);
|
||||
if (sessionId) {
|
||||
handleMetaChange(root, sessionId, broker, packet.payload.toString());
|
||||
}
|
||||
cb();
|
||||
},
|
||||
() => {},
|
||||
);
|
||||
|
||||
console.log("[journal] persistence listener active");
|
||||
console.log("[journal] broker available at ws://<host>:<port>");
|
||||
|
||||
return {
|
||||
broker,
|
||||
tcpServer,
|
||||
persistenceClient: client,
|
||||
wsServer,
|
||||
async close() {
|
||||
console.log("[journal] shutting down...");
|
||||
client?.end(true);
|
||||
await new Promise<void>((r) => wsServer.close(() => r()));
|
||||
await new Promise<void>((r) => broker.close(() => r()));
|
||||
await new Promise<void>((r) => tcpServer.close(() => r()));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function extractSessionId(topic: string): string | null {
|
||||
const parts = topic.split("/");
|
||||
return parts.length >= 3 ? parts[1] : null;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Session lifecycle
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const $SESSIONS = "ttrpg/$SESSIONS";
|
||||
|
||||
function handleMetaChange(
|
||||
root: string,
|
||||
sessionId: string,
|
||||
client: import("mqtt").MqttClient,
|
||||
broker: import("aedes").Aedes,
|
||||
rawPayload: string,
|
||||
): void {
|
||||
const manifest = loadManifest(root);
|
||||
|
||||
try {
|
||||
const meta = JSON.parse(rawPayload) as SessionMeta | null;
|
||||
|
||||
if (!meta || !meta.name) {
|
||||
// Tombstone: empty or invalid payload → delete session
|
||||
if (manifest.sessions[sessionId]) {
|
||||
delete manifest.sessions[sessionId];
|
||||
|
||||
// Wipe files
|
||||
const sp = streamPath(root, sessionId);
|
||||
if (existsSync(sp)) unlinkSync(sp);
|
||||
const sd = sessionDir(root, sessionId);
|
||||
try {
|
||||
rmdirSync(sd);
|
||||
} catch {
|
||||
/* not empty, fine */
|
||||
/* */
|
||||
}
|
||||
|
||||
console.log(`[journal] session deleted: ${sessionId}`);
|
||||
}
|
||||
} else {
|
||||
// Create or update
|
||||
manifest.sessions[sessionId] = {
|
||||
name: meta.name,
|
||||
created: meta.created || Date.now(),
|
||||
@@ -215,12 +199,17 @@ function handleMetaChange(
|
||||
}
|
||||
|
||||
saveManifest(root, manifest);
|
||||
|
||||
// Republish full manifest as a retained message
|
||||
client.publish($SESSIONS, JSON.stringify(manifest), {
|
||||
qos: 1,
|
||||
retain: true,
|
||||
});
|
||||
broker.publish(
|
||||
{
|
||||
topic: $SESSIONS,
|
||||
payload: Buffer.from(JSON.stringify(manifest)),
|
||||
qos: 1,
|
||||
retain: true,
|
||||
cmd: "publish" as const,
|
||||
dup: false,
|
||||
},
|
||||
() => {},
|
||||
);
|
||||
} catch (e) {
|
||||
console.error(`[journal] meta parse err for ${sessionId}:`, e);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
export interface ServeOptions {
|
||||
port: string;
|
||||
host?: string;
|
||||
mqttPort?: string;
|
||||
}
|
||||
|
||||
export interface CompileOptions {
|
||||
|
||||
Reference in New Issue
Block a user