1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
// OT state machine for main process (mirror of renderer otClient)
import type { OtOp } from './otTypes'
import { transformOps } from './otTransform'
export type SendFn = (ops: OtOp[], version: number) => void
export type ApplyFn = (ops: OtOp[]) => void
interface OtState {
name: 'synchronized' | 'awaitingConfirm' | 'awaitingWithBuffer'
inflight: OtOp[] | null
buffer: OtOp[] | null
version: number
}
export class OtClient {
private state: OtState
private sendFn: SendFn
private applyFn: ApplyFn
constructor(version: number, sendFn: SendFn, applyFn: ApplyFn) {
this.state = { name: 'synchronized', inflight: null, buffer: null, version }
this.sendFn = sendFn
this.applyFn = applyFn
}
get version(): number {
return this.state.version
}
get stateName(): string {
return this.state.name
}
onLocalOps(ops: OtOp[]) {
if (ops.length === 0) return
switch (this.state.name) {
case 'synchronized':
this.state = {
name: 'awaitingConfirm',
inflight: ops,
buffer: null,
version: this.state.version
}
this.sendFn(ops, this.state.version)
break
case 'awaitingConfirm':
this.state = {
name: 'awaitingWithBuffer',
inflight: this.state.inflight,
buffer: ops,
version: this.state.version
}
break
case 'awaitingWithBuffer':
this.state = {
...this.state,
buffer: [...(this.state.buffer || []), ...ops]
}
break
}
}
onAck() {
switch (this.state.name) {
case 'awaitingConfirm':
this.state = {
name: 'synchronized',
inflight: null,
buffer: null,
version: this.state.version + 1
}
break
case 'awaitingWithBuffer': {
const bufferOps = this.state.buffer || []
this.state = {
name: 'awaitingConfirm',
inflight: bufferOps,
buffer: null,
version: this.state.version + 1
}
this.sendFn(bufferOps, this.state.version)
break
}
case 'synchronized':
console.warn('[OtClient:main] unexpected ack in synchronized state')
break
}
}
onRemoteOps(ops: OtOp[], newVersion: number) {
switch (this.state.name) {
case 'synchronized':
this.state = { ...this.state, version: newVersion }
this.applyFn(ops)
break
case 'awaitingConfirm': {
const { left: transformedRemote, right: transformedInflight } = transformOps(ops, this.state.inflight || [])
this.state = {
...this.state,
inflight: transformedInflight,
version: newVersion
}
this.applyFn(transformedRemote)
break
}
case 'awaitingWithBuffer': {
const { left: remoteAfterInflight, right: inflightAfterRemote } = transformOps(ops, this.state.inflight || [])
const { left: remoteAfterBuffer, right: bufferAfterRemote } = transformOps(remoteAfterInflight, this.state.buffer || [])
this.state = {
...this.state,
inflight: inflightAfterRemote,
buffer: bufferAfterRemote,
version: newVersion
}
this.applyFn(remoteAfterBuffer)
break
}
}
}
reset(version: number) {
this.state = { name: 'synchronized', inflight: null, buffer: null, version }
}
}
|
