Tons of work
This commit is contained in:
Zef Hemel
+207
-208
@@ -11,220 +11,219 @@ import { PageMeta } from "./server";
|
||||
import { ClientPageState, Page } from "./types";
|
||||
import { safeRun } from "./util";
|
||||
|
||||
export class SocketAPI {
|
||||
openPages = new Map<string, Page>();
|
||||
connectedSockets: Set<Socket> = new Set();
|
||||
pageStore: DiskStorage;
|
||||
export function exposeSocketAPI(rootPath: string, io: Server) {
|
||||
const openPages = new Map<string, Page>();
|
||||
const connectedSockets: Set<Socket> = new Set();
|
||||
const pageStore = new DiskStorage(rootPath);
|
||||
fileWatcher(rootPath);
|
||||
|
||||
constructor(rootPath: string, io: Server) {
|
||||
this.pageStore = new DiskStorage(rootPath);
|
||||
this.fileWatcher(rootPath);
|
||||
io.on("connection", (socket) => {
|
||||
const socketOpenPages = new Set<string>();
|
||||
|
||||
io.on("connection", (socket) => {
|
||||
console.log("Connected", socket.id);
|
||||
this.connectedSockets.add(socket);
|
||||
const socketOpenPages = new Set<string>();
|
||||
console.log("Connected", socket.id);
|
||||
connectedSockets.add(socket);
|
||||
|
||||
socket.on("disconnect", () => {
|
||||
console.log("Disconnected", socket.id);
|
||||
socketOpenPages.forEach(disconnectPageSocket);
|
||||
this.connectedSockets.delete(socket);
|
||||
});
|
||||
|
||||
function onCall(eventName: string, cb: (...args: any[]) => Promise<any>) {
|
||||
socket.on(eventName, (reqId: number, ...args) => {
|
||||
cb(...args).then((result) => {
|
||||
socket.emit(`${eventName}Resp${reqId}`, result);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
const _this = this;
|
||||
function disconnectPageSocket(pageName: string) {
|
||||
let page = _this.openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
if (client.socket === socket) {
|
||||
_this.disconnectClient(client, page);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
onCall("openPage", async (pageName: string) => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (!page) {
|
||||
try {
|
||||
let { text, meta } = await this.pageStore.readPage(pageName);
|
||||
page = new Page(pageName, text, meta);
|
||||
} catch (e) {
|
||||
console.log("Creating new page", pageName);
|
||||
page = new Page(pageName, "", { name: pageName, lastModified: 0 });
|
||||
}
|
||||
this.openPages.set(pageName, page);
|
||||
}
|
||||
page.clientStates.add(new ClientPageState(socket, page.version));
|
||||
socketOpenPages.add(pageName);
|
||||
console.log("Opened page", pageName);
|
||||
this.broadcastCursors(page);
|
||||
return page.toJSON();
|
||||
});
|
||||
|
||||
socket.on("closePage", (pageName: string) => {
|
||||
console.log("Closing page", pageName);
|
||||
socketOpenPages.delete(pageName);
|
||||
disconnectPageSocket(pageName);
|
||||
});
|
||||
|
||||
onCall(
|
||||
"pushUpdates",
|
||||
async (
|
||||
pageName: string,
|
||||
version: number,
|
||||
updates: any[]
|
||||
): Promise<boolean> => {
|
||||
let page = this.openPages.get(pageName);
|
||||
|
||||
if (!page) {
|
||||
console.error(
|
||||
"Received updates for not open page",
|
||||
pageName,
|
||||
this.openPages.keys()
|
||||
);
|
||||
return false;
|
||||
}
|
||||
if (version !== page.version) {
|
||||
console.error("Invalid version", version, page.version);
|
||||
return false;
|
||||
} else {
|
||||
console.log("Applying", updates.length, "updates to", pageName);
|
||||
let transformedUpdates = [];
|
||||
let textChanged = false;
|
||||
for (let update of updates) {
|
||||
let changes = ChangeSet.fromJSON(update.changes);
|
||||
let transformedUpdate = {
|
||||
changes,
|
||||
clientID: update.clientID,
|
||||
effects: update.cursors?.map((c: Cursor) => {
|
||||
page.cursors.set(c.userId, c);
|
||||
return cursorEffect.of(c);
|
||||
}),
|
||||
};
|
||||
page.updates.push(transformedUpdate);
|
||||
transformedUpdates.push(transformedUpdate);
|
||||
let oldText = page.text;
|
||||
page.text = changes.apply(page.text);
|
||||
if (oldText !== page.text) {
|
||||
textChanged = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (textChanged) {
|
||||
if (page.saveTimer) {
|
||||
clearTimeout(page.saveTimer);
|
||||
}
|
||||
|
||||
page.saveTimer = setTimeout(() => {
|
||||
this.flushPageToDisk(pageName, page);
|
||||
}, 1000);
|
||||
}
|
||||
while (page.pending.length) {
|
||||
page.pending.pop()!(transformedUpdates);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall(
|
||||
"pullUpdates",
|
||||
async (pageName: string, version: number): Promise<Update[]> => {
|
||||
let page = this.openPages.get(pageName);
|
||||
// console.log("Pulling updates for", pageName);
|
||||
if (!page) {
|
||||
console.error("Fetching updates for not open page");
|
||||
return [];
|
||||
}
|
||||
// TODO: Optimize this
|
||||
let oldestVersion = Infinity;
|
||||
page.clientStates.forEach((client) => {
|
||||
oldestVersion = Math.min(client.version, oldestVersion);
|
||||
if (client.socket === socket) {
|
||||
client.version = version;
|
||||
}
|
||||
});
|
||||
page.flushUpdates(oldestVersion);
|
||||
if (version < page.version) {
|
||||
return page.updatesSince(version);
|
||||
} else {
|
||||
return new Promise((resolve) => {
|
||||
page.pending.push(resolve);
|
||||
});
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall(
|
||||
"readPage",
|
||||
async (pageName: string): Promise<{ text: string; meta: PageMeta }> => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
console.log("Serving page from memory", pageName);
|
||||
return {
|
||||
text: page.text.sliceString(0),
|
||||
meta: page.meta,
|
||||
};
|
||||
} else {
|
||||
return this.pageStore.readPage(pageName);
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall("writePage", async (pageName: string, text: string) => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
client.socket.emit("reloadPage", pageName);
|
||||
}
|
||||
this.openPages.delete(pageName);
|
||||
}
|
||||
return this.pageStore.writePage(pageName, text);
|
||||
});
|
||||
|
||||
onCall("deletePage", async (pageName: string) => {
|
||||
this.openPages.delete(pageName);
|
||||
socketOpenPages.delete(pageName);
|
||||
// Cascading of this to all connected clients will be handled by file watcher
|
||||
return this.pageStore.deletePage(pageName);
|
||||
});
|
||||
|
||||
onCall("listPages", async (): Promise<PageMeta[]> => {
|
||||
return this.pageStore.listPages();
|
||||
});
|
||||
|
||||
onCall("getPageMeta", async (pageName: string): Promise<PageMeta> => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
return page.meta;
|
||||
}
|
||||
return this.pageStore.getPageMeta(pageName);
|
||||
});
|
||||
socket.on("disconnect", () => {
|
||||
console.log("Disconnected", socket.id);
|
||||
socketOpenPages.forEach(disconnectPageSocket);
|
||||
connectedSockets.delete(socket);
|
||||
});
|
||||
}
|
||||
|
||||
private disconnectClient(client: ClientPageState, page: Page) {
|
||||
socket.on("closePage", (pageName: string) => {
|
||||
console.log("Closing page", pageName);
|
||||
socketOpenPages.delete(pageName);
|
||||
disconnectPageSocket(pageName);
|
||||
});
|
||||
|
||||
const onCall = (
|
||||
eventName: string,
|
||||
cb: (...args: any[]) => Promise<any>
|
||||
) => {
|
||||
socket.on(eventName, (reqId: number, ...args) => {
|
||||
cb(...args).then((result) => {
|
||||
socket.emit(`${eventName}Resp${reqId}`, result);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
const disconnectPageSocket = (pageName: string) => {
|
||||
let page = openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
if (client.socket === socket) {
|
||||
disconnectClient(client, page);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
onCall("openPage", async (pageName: string) => {
|
||||
let page = openPages.get(pageName);
|
||||
if (!page) {
|
||||
try {
|
||||
let { text, meta } = await pageStore.readPage(pageName);
|
||||
page = new Page(pageName, text, meta);
|
||||
} catch (e) {
|
||||
console.log("Creating new page", pageName);
|
||||
page = new Page(pageName, "", { name: pageName, lastModified: 0 });
|
||||
}
|
||||
openPages.set(pageName, page);
|
||||
}
|
||||
page.clientStates.add(new ClientPageState(socket, page.version));
|
||||
socketOpenPages.add(pageName);
|
||||
console.log("Opened page", pageName);
|
||||
broadcastCursors(page);
|
||||
return page.toJSON();
|
||||
});
|
||||
|
||||
onCall(
|
||||
"pushUpdates",
|
||||
async (
|
||||
pageName: string,
|
||||
version: number,
|
||||
updates: any[]
|
||||
): Promise<boolean> => {
|
||||
let page = openPages.get(pageName);
|
||||
|
||||
if (!page) {
|
||||
console.error(
|
||||
"Received updates for not open page",
|
||||
pageName,
|
||||
openPages.keys()
|
||||
);
|
||||
return false;
|
||||
}
|
||||
if (version !== page.version) {
|
||||
console.error("Invalid version", version, page.version);
|
||||
return false;
|
||||
} else {
|
||||
console.log("Applying", updates.length, "updates to", pageName);
|
||||
let transformedUpdates = [];
|
||||
let textChanged = false;
|
||||
for (let update of updates) {
|
||||
let changes = ChangeSet.fromJSON(update.changes);
|
||||
let transformedUpdate = {
|
||||
changes,
|
||||
clientID: update.clientID,
|
||||
effects: update.cursors?.map((c: Cursor) => {
|
||||
page.cursors.set(c.userId, c);
|
||||
return cursorEffect.of(c);
|
||||
}),
|
||||
};
|
||||
page.updates.push(transformedUpdate);
|
||||
transformedUpdates.push(transformedUpdate);
|
||||
let oldText = page.text;
|
||||
page.text = changes.apply(page.text);
|
||||
if (oldText !== page.text) {
|
||||
textChanged = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (textChanged) {
|
||||
if (page.saveTimer) {
|
||||
clearTimeout(page.saveTimer);
|
||||
}
|
||||
|
||||
page.saveTimer = setTimeout(() => {
|
||||
flushPageToDisk(pageName, page);
|
||||
}, 1000);
|
||||
}
|
||||
while (page.pending.length) {
|
||||
page.pending.pop()!(transformedUpdates);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall(
|
||||
"pullUpdates",
|
||||
async (pageName: string, version: number): Promise<Update[]> => {
|
||||
let page = openPages.get(pageName);
|
||||
// console.log("Pulling updates for", pageName);
|
||||
if (!page) {
|
||||
console.error("Fetching updates for not open page");
|
||||
return [];
|
||||
}
|
||||
// TODO: Optimize this
|
||||
let oldestVersion = Infinity;
|
||||
page.clientStates.forEach((client) => {
|
||||
oldestVersion = Math.min(client.version, oldestVersion);
|
||||
if (client.socket === socket) {
|
||||
client.version = version;
|
||||
}
|
||||
});
|
||||
page.flushUpdates(oldestVersion);
|
||||
if (version < page.version) {
|
||||
return page.updatesSince(version);
|
||||
} else {
|
||||
return new Promise((resolve) => {
|
||||
page.pending.push(resolve);
|
||||
});
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall(
|
||||
"readPage",
|
||||
async (pageName: string): Promise<{ text: string; meta: PageMeta }> => {
|
||||
let page = openPages.get(pageName);
|
||||
if (page) {
|
||||
console.log("Serving page from memory", pageName);
|
||||
return {
|
||||
text: page.text.sliceString(0),
|
||||
meta: page.meta,
|
||||
};
|
||||
} else {
|
||||
return pageStore.readPage(pageName);
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
onCall("writePage", async (pageName: string, text: string) => {
|
||||
let page = openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
client.socket.emit("reloadPage", pageName);
|
||||
}
|
||||
openPages.delete(pageName);
|
||||
}
|
||||
return pageStore.writePage(pageName, text);
|
||||
});
|
||||
|
||||
onCall("deletePage", async (pageName: string) => {
|
||||
openPages.delete(pageName);
|
||||
socketOpenPages.delete(pageName);
|
||||
// Cascading of this to all connected clients will be handled by file watcher
|
||||
return pageStore.deletePage(pageName);
|
||||
});
|
||||
|
||||
onCall("listPages", async (): Promise<PageMeta[]> => {
|
||||
return pageStore.listPages();
|
||||
});
|
||||
|
||||
onCall("getPageMeta", async (pageName: string): Promise<PageMeta> => {
|
||||
let page = openPages.get(pageName);
|
||||
if (page) {
|
||||
return page.meta;
|
||||
}
|
||||
return pageStore.getPageMeta(pageName);
|
||||
});
|
||||
});
|
||||
|
||||
function disconnectClient(client: ClientPageState, page: Page) {
|
||||
page.clientStates.delete(client);
|
||||
if (page.clientStates.size === 0) {
|
||||
console.log("No more clients for", page.name, "flushing");
|
||||
this.flushPageToDisk(page.name, page);
|
||||
this.openPages.delete(page.name);
|
||||
flushPageToDisk(page.name, page);
|
||||
openPages.delete(page.name);
|
||||
} else {
|
||||
page.cursors.delete(client.socket.id);
|
||||
this.broadcastCursors(page);
|
||||
broadcastCursors(page);
|
||||
}
|
||||
}
|
||||
|
||||
private broadcastCursors(page: Page) {
|
||||
function broadcastCursors(page: Page) {
|
||||
page.clientStates.forEach((client) => {
|
||||
client.socket.emit(
|
||||
"cursorSnapshot",
|
||||
@@ -234,15 +233,15 @@ export class SocketAPI {
|
||||
});
|
||||
}
|
||||
|
||||
private flushPageToDisk(name: string, page: Page) {
|
||||
function flushPageToDisk(name: string, page: Page) {
|
||||
safeRun(async () => {
|
||||
let meta = await this.pageStore.writePage(name, page.text.sliceString(0));
|
||||
let meta = await pageStore.writePage(name, page.text.sliceString(0));
|
||||
console.log(`Wrote page ${name} to disk`);
|
||||
page.meta = meta;
|
||||
});
|
||||
}
|
||||
|
||||
private fileWatcher(rootPath: string) {
|
||||
function fileWatcher(rootPath: string) {
|
||||
fs.watch(
|
||||
rootPath,
|
||||
{
|
||||
@@ -264,16 +263,16 @@ export class SocketAPI {
|
||||
} catch (e) {
|
||||
// File was deleted
|
||||
console.log("Deleted", pageName);
|
||||
for (let socket of this.connectedSockets) {
|
||||
for (let socket of connectedSockets) {
|
||||
socket.emit("pageDeleted", pageName);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const openPage = this.openPages.get(pageName);
|
||||
const openPage = openPages.get(pageName);
|
||||
if (openPage) {
|
||||
if (openPage.meta.lastModified < modifiedTime) {
|
||||
console.log("Page changed on disk outside of editor, reloading");
|
||||
this.openPages.delete(pageName);
|
||||
openPages.delete(pageName);
|
||||
const meta = {
|
||||
name: pageName,
|
||||
lastModified: modifiedTime,
|
||||
@@ -288,7 +287,7 @@ export class SocketAPI {
|
||||
console.log(
|
||||
"New file created, broadcasting to all connected sockets"
|
||||
);
|
||||
for (let socket of this.connectedSockets) {
|
||||
for (let socket of connectedSockets) {
|
||||
socket.emit("pageCreated", {
|
||||
name: pageName,
|
||||
lastModified: modifiedTime,
|
||||
|
||||
@@ -12,9 +12,7 @@ export class DiskStorage {
|
||||
async listPages(): Promise<PageMeta[]> {
|
||||
let fileNames: PageMeta[] = [];
|
||||
|
||||
let _this = this;
|
||||
|
||||
async function walkPath(dir: string) {
|
||||
const walkPath = async (dir: string) => {
|
||||
let files = await readdir(dir);
|
||||
for (let file of files) {
|
||||
const fullPath = path.join(dir, file);
|
||||
@@ -25,7 +23,7 @@ export class DiskStorage {
|
||||
if (path.extname(file) === ".md") {
|
||||
fileNames.push({
|
||||
name: fullPath.substring(
|
||||
_this.rootPath.length + 1,
|
||||
this.rootPath.length + 1,
|
||||
fullPath.length - 3
|
||||
),
|
||||
lastModified: s.mtime.getTime(),
|
||||
@@ -33,7 +31,7 @@ export class DiskStorage {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
await walkPath(this.rootPath);
|
||||
return fileNames;
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ import express from "express";
|
||||
import { readFile } from "fs/promises";
|
||||
import http from "http";
|
||||
import { Server } from "socket.io";
|
||||
import { SocketAPI } from "./api";
|
||||
import { exposeSocketAPI } from "./api";
|
||||
|
||||
const app = express();
|
||||
const server = http.createServer(app);
|
||||
@@ -24,7 +24,7 @@ export type PageMeta = {
|
||||
};
|
||||
|
||||
app.use("/", express.static(distDir));
|
||||
let filesystem = new SocketAPI(pagesPath, io);
|
||||
exposeSocketAPI(pagesPath, io);
|
||||
|
||||
// Fallback, serve index.html
|
||||
let cachedIndex: string | undefined = undefined;
|
||||
|
||||
Reference in New Issue
Block a user