From 13dc978a26141c0d1d8465632fa786012505e774 Mon Sep 17 00:00:00 2001 From: ptt <1928627998@qq.com> Date: Sat, 29 Aug 2026 18:59:03 +0800 Subject: [PATCH] Add WebSocket sync example --- examples/sync/index.html | 92 +++++++++++++++++++++++++ examples/sync/package.json | 13 ++++ examples/sync/relay.mjs | 107 ++++++++++++++++++++++++++++ examples/sync/relay.test.mjs | 130 +++++++++++++++++++++++++++++++++++ package-lock.json | 13 +++- package.json | 4 +- 6 files changed, 357 insertions(+), 2 deletions(-) create mode 100644 examples/sync/index.html create mode 100644 examples/sync/package.json create mode 100644 examples/sync/relay.mjs create mode 100644 examples/sync/relay.test.mjs diff --git a/examples/sync/index.html b/examples/sync/index.html new file mode 100644 index 0000000..1a6295f --- /dev/null +++ b/examples/sync/index.html @@ -0,0 +1,92 @@ + + + + + + Quickdraw — real-time sync example + + + + +
+
+

Real-time sync over a WebSocket relay

+

Draw on either board. Diffs are relayed to the other client; conflicts use record-level last-writer-wins.

+
+

npm run dev:sync

+
+
+
+
Client Aconnecting
+
+
+
+
Client Bconnecting
+
+
+
+ + + + diff --git a/examples/sync/package.json b/examples/sync/package.json new file mode 100644 index 0000000..5d12dee --- /dev/null +++ b/examples/sync/package.json @@ -0,0 +1,13 @@ +{ + "name": "quickdraw-sync-example", + "private": true, + "version": "0.1.0", + "type": "module", + "scripts": { + "dev": "node relay.mjs", + "test": "node --test relay.test.mjs" + }, + "dependencies": { + "ws": "^8.18.0" + } +} diff --git a/examples/sync/relay.mjs b/examples/sync/relay.mjs new file mode 100644 index 0000000..dfcb946 --- /dev/null +++ b/examples/sync/relay.mjs @@ -0,0 +1,107 @@ +import { createReadStream } from 'node:fs' +import { stat } from 'node:fs/promises' +import { createServer } from 'node:http' +import { dirname, extname, resolve, sep } from 'node:path' +import { fileURLToPath, pathToFileURL } from 'node:url' +import { WebSocket, WebSocketServer } from 'ws' + +const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), '../..') +const mimeTypes = { + '.css': 'text/css; charset=utf-8', + '.html': 'text/html; charset=utf-8', + '.js': 'text/javascript; charset=utf-8', + '.json': 'application/json; charset=utf-8', + '.mjs': 'text/javascript; charset=utf-8', + '.png': 'image/png', + '.svg': 'image/svg+xml', +} + +function serve(root, req, res) { + if (req.method !== 'GET' && req.method !== 'HEAD') { + res.writeHead(405, { Allow: 'GET, HEAD' }).end() + return + } + + const url = new URL(req.url, 'http://localhost') + if (url.pathname === '/') { + res.writeHead(302, { Location: '/examples/sync/' }).end() + return + } + + let pathname + try { + pathname = decodeURIComponent(url.pathname) + } catch { + res.writeHead(400).end('Bad request') + return + } + + const relative = pathname.endsWith('/') ? `${pathname}index.html` : pathname + const file = resolve(root, `.${relative}`) + if (file !== root && !file.startsWith(`${root}${sep}`)) { + res.writeHead(403).end('Forbidden') + return + } + + stat(file).then((info) => { + if (!info.isFile()) throw new Error('not a file') + res.writeHead(200, { + 'Content-Length': info.size, + 'Content-Type': mimeTypes[extname(file)] || 'application/octet-stream', + }) + if (req.method === 'HEAD') res.end() + else createReadStream(file).pipe(res) + }).catch(() => res.writeHead(404).end('Not found')) +} + +export function createSyncServer({ root = repoRoot } = {}) { + const server = createServer((req, res) => serve(root, req, res)) + const relay = new WebSocketServer({ server }) + + relay.on('connection', (socket) => { + socket.on('message', (data, isBinary) => { + for (const peer of relay.clients) { + if (peer !== socket && peer.readyState === WebSocket.OPEN) { + peer.send(data, { binary: isBinary }) + } + } + }) + }) + + return { + server, + relay, + listen(port = 8080, host = '127.0.0.1') { + return new Promise((resolveListen, reject) => { + const onError = (error) => reject(error) + server.once('error', onError) + server.listen(port, host, () => { + server.off('error', onError) + resolveListen(server.address()) + }) + }) + }, + close() { + for (const socket of relay.clients) socket.terminate() + return new Promise((resolveClose, reject) => { + relay.close(() => server.close((error) => error ? reject(error) : resolveClose())) + }) + }, + } +} + +const isMain = process.argv[1] && pathToFileURL(resolve(process.argv[1])).href === import.meta.url + +if (isMain) { + const app = createSyncServer() + const port = Number(process.env.PORT || 8080) + const address = await app.listen(port) + console.log(`Quickdraw sync example: http://localhost:${address.port}/examples/sync/`) + + const stop = async () => { + await app.close() + process.exit(0) + } + process.once('SIGINT', stop) + process.once('SIGTERM', stop) +} diff --git a/examples/sync/relay.test.mjs b/examples/sync/relay.test.mjs new file mode 100644 index 0000000..14ddcbe --- /dev/null +++ b/examples/sync/relay.test.mjs @@ -0,0 +1,130 @@ +import assert from 'node:assert/strict' +import { afterEach, describe, it } from 'node:test' +import WebSocket from 'ws' +import { Store } from '../../packages/core/src/store.js' +import { createSyncServer } from './relay.mjs' + +const apps = new Set() +const sockets = new Set() + +afterEach(async () => { + for (const socket of sockets) socket.terminate() + sockets.clear() + await Promise.all([...apps].map((app) => app.close())) + apps.clear() +}) + +async function setup() { + const app = createSyncServer() + apps.add(app) + const { port } = await app.listen(0) + return { app, url: `ws://127.0.0.1:${port}` } +} + +function connect(url) { + return new Promise((resolve, reject) => { + const socket = new WebSocket(url) + sockets.add(socket) + const timer = setTimeout(() => { + socket.terminate() + reject(new Error('timed out opening WebSocket')) + }, 1000) + socket.once('open', () => { + clearTimeout(timer) + resolve(socket) + }) + socket.once('error', (error) => { + clearTimeout(timer) + reject(error) + }) + }) +} + +function nextMessage(socket) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + socket.off('message', onMessage) + reject(new Error('timed out waiting for WebSocket message')) + }, 1000) + const onMessage = (data, isBinary) => { + clearTimeout(timer) + resolve({ data, isBinary }) + } + socket.once('message', onMessage) + socket.once('error', (error) => { + clearTimeout(timer) + socket.off('message', onMessage) + reject(error) + }) + }) +} + +const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms)) + +async function waitFor(check, timeout = 1000) { + const started = Date.now() + while (!check()) { + if (Date.now() - started > timeout) throw new Error('timed out waiting for condition') + await wait(10) + } +} + +describe('sync relay', () => { + it('forwards bytes to peers without echoing to the sender', async () => { + const { url } = await setup() + const [a, b] = await Promise.all([connect(url), connect(url)]) + let echoed = false + a.on('message', () => { echoed = true }) + + const received = nextMessage(b) + a.send(Buffer.from([0, 255, 1, 2])) + + const message = await received + assert.deepEqual([...message.data], [0, 255, 1, 2]) + assert.equal(message.isBinary, true) + await wait(50) + assert.equal(echoed, false) + }) + + it('forwards in both directions and keeps serving after a peer disconnects', async () => { + const { url } = await setup() + const [a, b, c] = await Promise.all([connect(url), connect(url), connect(url)]) + + const fromB = nextMessage(a) + b.send('from b') + assert.equal((await fromB).data.toString(), 'from b') + + await new Promise((resolve) => { + b.once('close', resolve) + b.close() + }) + sockets.delete(b) + + const fromA = nextMessage(c) + a.send('from a') + assert.equal((await fromA).data.toString(), 'from a') + }) + + it('keeps remote records out of the receiving store undo history', async () => { + const { url } = await setup() + const [socketA, socketB] = await Promise.all([connect(url), connect(url)]) + const storeA = new Store() + const storeB = new Store() + const localB = { id: 'shape:local-b', typeName: 'shape', type: 'rect', x: 0, y: 0, z: 1, props: {} } + const remoteA = { id: 'shape:remote-a', typeName: 'shape', type: 'rect', x: 10, y: 10, z: 2, props: {} } + + storeB.put(localB) + storeA.listen((diff) => socketA.send(JSON.stringify(diff)), { source: 'user' }) + storeB.listen((diff) => socketB.send(JSON.stringify(diff)), { source: 'user' }) + socketA.on('message', (data) => storeA.applyDiff(JSON.parse(data.toString()), 'remote')) + socketB.on('message', (data) => storeB.applyDiff(JSON.parse(data.toString()), 'remote')) + + storeA.put(remoteA) + await waitFor(() => storeB.has(remoteA.id)) + + assert.equal(storeB.undos.length, 1) + storeB.undo() + assert.equal(storeB.has(localB.id), false) + assert.deepEqual(storeB.get(remoteA.id), remoteA) + }) +}) diff --git a/package-lock.json b/package-lock.json index dd78346..03d24e4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,6 +11,7 @@ "workspaces": [ "packages/*", "examples/react-demo", + "examples/sync", "apps/*" ], "devDependencies": { @@ -229,6 +230,13 @@ } } }, + "examples/sync": { + "name": "quickdraw-sync-example", + "version": "0.1.0", + "dependencies": { + "ws": "^8.18.0" + } + }, "node_modules/@antfu/install-pkg": { "version": "1.1.0", "resolved": "https://registry.npmjs.org/@antfu/install-pkg/-/install-pkg-1.1.0.tgz", @@ -11275,6 +11283,10 @@ "resolved": "examples/react-demo", "link": true }, + "node_modules/quickdraw-sync-example": { + "resolved": "examples/sync", + "link": true + }, "node_modules/quickdraw-website": { "resolved": "apps/website", "link": true @@ -14431,7 +14443,6 @@ "version": "8.21.1", "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", - "dev": true, "license": "MIT", "engines": { "node": ">=10.0.0" diff --git a/package.json b/package.json index 5983fb9..63d51e5 100644 --- a/package.json +++ b/package.json @@ -7,13 +7,15 @@ "workspaces": [ "packages/*", "examples/react-demo", + "examples/sync", "apps/*" ], "scripts": { - "test": "vitest run", + "test": "vitest run && npm run test --workspace=quickdraw-sync-example", "test:watch": "vitest", "build": "npm run build --workspace=@quickdrawjs/react-native", "dev": "npm run dev --workspace=quickdraw-react-demo", + "dev:sync": "npm run dev --workspace=quickdraw-sync-example", "typecheck": "tsc -p tsconfig.json", "star-history": "node scripts/star-history.mjs", "publish:all": "npm publish --workspace=@quickdrawjs/core --access public && npm publish --workspace=@quickdrawjs/react --access public && npm publish --workspace=@quickdrawjs/react-native --access public"