mattuy/ot-demo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commit

Browse Source
This commit is contained in:
呈祥
2026-01-06 23:05:37 +08:00
parent 7052675535
commit c605b89cf5
3 changed files with 140 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ot/client.ts
+29 -46
View File
@@ -9,8 +9,6 @@ export class Client {
public name: string = "";
// 当前文档
public document: string = "";
// 当前服务端文档状态
public syncedDoc: string = "";
// 下一个期望的修订号
public revision: number;
// 正在处理的操作，已经发送给服务端等待确认
@@ -35,64 +33,48 @@ export class Client {
this.sendOperationIfNeed();
}
// 接收到来自服务器的新操作，可能是另一个客户端的操作，也可能是服务端确认本地操作的响应
applyServer(revision: number, operation: TextOperation): void {
if (revision < 0 || this.revision > revision) {
throw new Error("operation revision not in history");
}
if (revision < this.revision) {
// 接收到其它客户端的操作
applyServer(operation: TextOperation): void {
if (!this.enableSync) {
// 模拟离线，忽略其它客户端的操作
return;
}
const isSelfOp = operation.id === this.outstanding?.id;
if (this.outstanding) {
// 将outstanding操作转化为执行operation后的操作
const newOp = TextOperation.transform(this.outstanding, operation);
newOp.id = this.outstanding.id;
this.outstanding = newOp;
// 将operation转换为基于outstanding操作后的操作
operation = TextOperation.transform(this.outstanding, operation)[1];
}
// 现在，operation是基于outstanding操作后的操作，buffer[0]也是
// （如果outstanding为空，则都是基于server文档状态操作）
for (let i = 0; i < this.buffer.length; i++) {
// 将buffer中的操作转换为基于服务端文档状态的操作
// FIXME: 对buffer中的操作应该合并后再transform
const newOp = TextOperation.transform(
this.buffer[i],
// 如果有outstanding，buffer[0]应该基于outstanding
this.buffer[i - 1] || this.outstanding || operation
);
newOp.id = this.buffer[i].id;
this.buffer[i] = newOp;
// 将operation转换基于buffer[i]操作后的操作
operation = TextOperation.transform(this.buffer[i], operation)[1];
}
// 本地重新应用buffer中的操作，合并其它客户端的操作到本地
let newDoc = this.syncedDoc;
newDoc = operation.apply(newDoc);
for (let i = 0; i < this.buffer.length; i++) {
newDoc = this.buffer[i].apply(newDoc);
}
this.document = newDoc;
console.log("client", this.name, "received operation", operation);
// 更新本地同步状态
this.syncedDoc = operation.apply(this.syncedDoc);
this.revision = revision;
// 操作来自服务端确认本地操作的响应
if (isSelfOp) {
this.serverAck();
}
console.log("mattuy", "client", this.name, this);
// 现在operation是基于本地最新document的操作了，直接在本地应用它！
this.applyOperation(operation);
this.revision++;
}
// 服务端已确认收到发送的操作，从buffer中移除该操作，尝试发送下一个
// 本地应用操作
applyOperation(operation: TextOperation): void {
this.document = operation.apply(this.document);
}
// 服务端已确认收到发送的操作，尝试发送下一个操作
// 简单实现。即使是模拟离线状态我们也接收Ack，避免实现操作去重
serverAck(): void {
console.log("client", this.name, "operation ack", this.outstanding);
this.outstanding = null;
this.revision++;
this.sendOperationIfNeed();
}
// 逐个向服务端发送本地待提交的操作。发送后等待服务端确认，触发serverAck方法后继续发送下一个操作
async sendOperationIfNeed() {
// 如果禁用了同步，则不向服务端发送操作
// 模拟离线，不向服务端发送操作
if (!this.enableSync) {
return;
}
@@ -102,16 +84,17 @@ export class Client {
}
// 向服务端发送本地待提交的操作，这里没有实现操作压缩，我们一个一个发送
this.outstanding = this.buffer.shift()!;
// 模拟向服务端发送操作，实际实现应该通过网络
// 模拟向服务端发送操作，实际实现应该通过网络发送
server.receiveOperation(this, this.revision, this.outstanding);
// 应该有超时重发机制，这里没有实现
}
// 设置是否启用同步到服务端
// 设置模拟离线状态
setEnableSync(enabled: boolean): void {
console.log("client", this.name, enabled ? "online" : "offline");
this.enableSync = enabled;
// 如果重新启用同步，尝试发送待发送的操作
// 重新启用同步，继续发送本地操作，发送一个空操作，确保服务端向我们推送最新操作
if (enabled) {
// this.buffer.push(new Noop());
this.sendOperationIfNeed();
}
}
src/ot/op.ts
+97 -47
View File
@@ -1,93 +1,143 @@
// 简单文本操作的实现
import { v4 as uuidv4 } from "uuid";
export abstract class SimpleTextOperation {
public id: string = uuidv4();
abstract toString(): string;
abstract equals(other: SimpleTextOperation): boolean;
abstract apply(doc: string): string;
// 把操作a转换为基于应用操作b后的文档状态的操作，返回操作a'
static transform(a: SimpleTextOperation, b: SimpleTextOperation): SimpleTextOperation {
/**
* OT算法转换函数
* @param a 操作a
* @param b 操作b
* @returns
* [0] = a' 操作a基于操作b应用后的操作
* [1] = b' 操作b基于操作a应用后的操作
* apply(apply(S, a), b') = apply(apply(S, b), a')
*/
static transform(
a: SimpleTextOperation,
b: SimpleTextOperation
): [SimpleTextOperation, SimpleTextOperation] {
if (a instanceof Noop || b instanceof Noop) {
return a;
return [a, b];
}
// 情况2: a 和 b 都是 Insert（插入操作）
if (a instanceof Insert && b instanceof Insert) {
if (a.position < b.position) {
// b在a插入的地方之后插入，不转换
return a;
// 情况2.1: a 的插入位置在 b 之前，或者位置相同但 a 的字符串字典序更小
// 处理: a 保持不变，b 的插入位置需要向后偏移 a 插入的字符串长度
if (a.position < b.position || (a.position === b.position && a.str < b.str)) {
return [a, new Insert(b.str, b.position + a.str.length)];
}
// b在a插入的地方之前插入，位置偏移b插入的str长度后插入
return new Insert(a.str, a.position + b.str.length);
// 情况2.2: a 的插入位置在 b 之后，或者位置相同但 a 的字符串字典序更大
// 处理: a 的插入位置需要向后偏移 b 插入的字符串长度，b 保持不变
if (a.position > b.position || (a.position === b.position && a.str > b.str)) {
return [new Insert(a.str, a.position + b.str.length), b];
}
// 情况2.3: a 和 b 插入位置相同且字符串完全相同
// 处理: 两个操作都变为 noop，因为插入的内容相同
return [noop, noop];
}
// 情况3: a 是 Insert（插入），b 是 Delete（删除）
if (a instanceof Insert && b instanceof Delete) {
// 情况3.1: a 的插入位置在 b 的删除位置之前或相同
// 处理: a 保持不变，b 的删除位置需要向后偏移 a 插入的字符串长度
if (a.position <= b.position) {
// b在a插入的地方之后删除，不转换
return a;
return [a, new Delete(b.count, b.position + a.str.length)];
}
// 情况3.2: a 的插入位置在 b 的删除范围之后
// 处理: a 的插入位置需要向前偏移 b 删除的字符数，b 保持不变
if (a.position >= b.position + b.count) {
// b在a插入的位置之前删除，并且a插入位置的字符没有被删除
return new Insert(a.str, a.position - b.count);
return [new Insert(a.str, a.position - b.count), b];
}
// b在a插入的位置之前删除，并且a插入位置的字符被删除，在b删除的位置插入a的str
return new Insert(a.str, b.position);
// 情况3.3: a 的插入位置在 b 的删除范围内部（有冲突）
// 处理: a 变为 noop（插入的内容被删除），b 的删除范围需要包含 a 插入的字符串长度
// 注意: 这会丢失 a 的插入意图，但这是保持转换函数有效性的唯一方法
return [noop, new Delete(b.count + a.str.length, b.position)];
}
// 情况4: a 是 Delete（删除），b 是 Insert（插入）
if (a instanceof Delete && b instanceof Insert) {
// 情况4.1: a 的删除位置在 b 的插入位置之后或相同
// 处理: a 的删除位置需要向后偏移 b 插入的字符串长度，b 保持不变
if (a.position >= b.position) {
// a在插入位置之后删除，增加删除操作的偏移
return new Delete(a.count, a.position + b.str.length);
return [new Delete(a.count, a.position + b.str.length), b];
}
// 情况4.2: a 的删除范围在 b 的插入位置之前（完全不相交）
// 处理: a 保持不变，b 的插入位置需要向前偏移 a 删除的字符数
if (a.position + a.count <= b.position) {
// a在插入位置之前删除，并且b插入之前的字符没有被删除，不转换
return a;
return [a, new Insert(b.str, b.position - a.count)];
}
// b插入的内容在a删除的范围内，直接把b插入的内容一起删除
// NOTICE: 这里b插入的内容丢失了，在生产环境应该裂变为前后两个删除操作，这里简化了实现
return new Delete(a.count + b.str.length, a.position);
// 情况4.3: b 的插入位置在 a 的删除范围内部（有冲突）
// 处理: a 的删除范围需要包含 b 插入的字符串长度，b 变为 noop（插入的内容被删除）
// 注意: 这会丢失 b 的插入意图，但这是保持转换函数有效性的唯一方法
return [new Delete(a.count + b.str.length, a.position), noop];
}
// 情况5: a 和 b 都是 Delete（删除操作）
if (a instanceof Delete && b instanceof Delete) {
// 删除位置相同
// 情况5.1: a 和 b 的删除位置相同
if (a.position === b.position) {
if (a.count <= b.count) {
// b删除得更多，已经把a要删除的删掉了，no-op
return noop;
// 情况5.1.1: 删除的字符数相同
// 处理: 两个操作都变为 noop，因为删除的内容完全相同
if (a.count === b.count) {
return [noop, noop];
}
// b删除得少，a删除得多，删除b没删掉的部分
return new Delete(a.count - b.count, a.position);
// 情况5.1.2: a 删除的字符数少于 b
// 处理: a 变为 noop（已经被 b 删除），b 只删除剩余的部分
else if (a.count < b.count) {
return [noop, new Delete(b.count - a.count, b.position)];
}
// 情况5.1.3: a 删除的字符数多于 b
// 处理: a 只删除剩余的部分，b 变为 noop（已经被 a 删除）
return [new Delete(a.count - b.count, a.position), noop];
}
// 情况5.2: a 的删除位置在 b 之前
if (a.position < b.position) {
// 情况5.2.1: a 的删除范围完全在 b 的删除位置之前（不相交）
// 处理: a 保持不变，b 的删除位置需要向前偏移 a 删除的字符数
if (a.position + a.count <= b.position) {
// a在b删除的位置之前删除，并且a删除的结束位置在b删除的开始位置之前，不转换
return a;
return [a, new Delete(b.count, b.position - a.count)];
}
// 情况5.2.2: a 的删除范围完全包含 b 的删除范围
// 处理: a 只删除 b 未删除的部分，b 变为 noop（已经被 a 删除）
if (a.position + a.count >= b.position + b.count) {
// a删除的范围包含b删除范围，减去b已经删除的长度
return new Delete(a.count - b.count, a.position);
return [new Delete(a.count - b.count, a.position), noop];
}
// a删除的范围在b删除的范围之前且有交集，减去交集部分长度
// b: *****------****
// a: ***----********
// a': ***--**********
return new Delete(b.position - a.position, a.position);
// 情况5.2.3: a 和 b 的删除范围有交集，a 在前，b 在后
// 处理: a 只删除到 b 开始位置之前的部分，b 删除从 a 结束位置之后到 b 结束位置的部分
// 示例: 文档 "0123456789"
// a: 删除位置0-5 (删除 "01234")
// b: 删除位置3-7 (删除 "3456")
// 结果: a' 删除位置0-3 (删除 "012")，b' 删除位置3-5 (删除 "67")
return [
new Delete(b.position - a.position, a.position),
new Delete(b.position + b.count - (a.position + a.count), a.position),
];
}
// 情况5.3: a 的删除位置在 b 之后
if (a.position > b.position) {
// 情况5.3.1: a 的删除范围完全在 b 的删除范围之后（不相交）
// 处理: a 的删除位置需要向前偏移 b 删除的字符数，b 保持不变
if (a.position >= b.position + b.count) {
// a删除的范围在b之后且没有交集，删除位置减去b删除的长度
return new Delete(a.count, a.position - b.count);
return [new Delete(a.count, a.position - b.count), b];
}
// 情况5.3.2: b 的删除范围完全包含 a 的删除范围
// 处理: a 变为 noop（已经被 b 删除），b 只删除剩余的部分
if (a.position + a.count <= b.position + b.count) {
// b删除范围包含a删除范围，no-op
return noop;
return [noop, new Delete(b.count - a.count, b.position)];
}
// a删除的范围在b删除的范围之后且有交集，减去交集部分长度，并把开始位置放到交集结束位置
// b: ***----*********
// a: *****------*****
// a': *******----*****
return new Delete(a.count - b.count, b.position + b.count);
// 情况5.3.3: a 和 b 的删除范围有交集，b 在前，a 在后
// 处理: a 删除从 b 结束位置之后到 a 结束位置的部分，b 删除从 b 开始位置到 a 开始位置的部分
// 示例: 文档 "0123456789"
// b: 删除位置3-7 (删除 "3456")
// a: 删除位置5-9 (删除 "5678")
// 结果: a' 删除位置7-9 (删除 "89")，b' 删除位置3-5 (删除 "34")
return [
new Delete(a.position + a.count - (b.position + b.count), b.position),
new Delete(a.position - b.position, b.position),
];
}
}
src/ot/server.ts
+14 -8
View File
@@ -16,25 +16,24 @@ class Server {
}
// 每当从客户端接收到操作时调用此方法
async receiveOperation(client: Client, revision: number, operation: TextOperation) {
receiveOperation(client: Client, revision: number, operation: TextOperation) {
console.log("server", "received operation", revision, operation);
if (revision < 0 || this.operations.length < revision) {
throw new Error("operation revision not in history");
}
console.log("mattuy", "received operation", revision, operation);
// 查找客户端发送操作时不知道的所有操作...
const concurrentOperations = this.operations.slice(revision);
// ...并将操作与所有这些操作进行转换...
const opId = operation.id;
const transform = TextOperation.transform;
for (let i = 0; i < concurrentOperations.length; i++) {
// op' = transform(op, op1)
// op基于 revision+i 版本的文档操作，op'基于 revision+i+1 版本的文档操作
operation = transform(operation, concurrentOperations[i]);
operation = transform(operation, concurrentOperations[i])[0];
// 向客户端发送其它客户端的操作
client.applyServer(revision + i + 1, concurrentOperations[i]);
client.applyServer(concurrentOperations[i]);
}
operation.id = opId;
// ...并将其应用到文档上
this.document = operation.apply(this.document);
@@ -42,9 +41,16 @@ class Server {
this.operations.push(operation);
// 将操作发送给所有连接的客户端并向创建者发送确认
this.clients.forEach((client) => {
client.applyServer(this.operations.length, operation);
this.clients.forEach((otherClient) => {
if (otherClient === client) {
// 自己不需要接收自己的操作，会响应Ack
return;
}
otherClient.applyServer(operation);
});
// 客户端确认收到操作
client.serverAck();
}
// 模拟客户端连接