Files
@ a4052905ca7d
Branch filter:
Location: light9/web/rdfdbclient.ts - annotation
a4052905ca7d
2.1 KiB
video/MP2T
notes about how rdfdb syncs, or should sync
4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 ac55319a2eac 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 ac55319a2eac 4556eebe5d73 4556eebe5d73 ac55319a2eac 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 ac55319a2eac 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 ac55319a2eac ac55319a2eac 4556eebe5d73 4556eebe5d73 4556eebe5d73 ac55319a2eac ac55319a2eac 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 4556eebe5d73 | import debug from "debug";
import { parseJsonPatch, Patch } from "./patch";
import { RdfDbChannel } from "./RdfDbChannel";
const log = debug("rdfdbclient");
export class RdfDbClient {
private channel: RdfDbChannel;
private patchesToSend: Patch[];
// Send and receive patches from rdfdb. Primarily used in SyncedGraph.
//
// What this should do, and does not yet, is keep the graph
// 'coasting' over a reconnect, applying only the diffs from the old
// contents to the new ones once they're in. Then, remove all the
// clearGraph stuff in graph.coffee that doesn't even work right.
//
constructor(
patchSenderUrl: string,
private clearGraphOnNewConnection: () => void,
private applyPatch: (p: Patch) => void,
setStatus: (status: string) => void
) {
this.patchesToSend = [];
this.channel = new RdfDbChannel(patchSenderUrl);
this.channel.statusDisplay.subscribe((st: string) => {
setStatus(st + `; ${this.patchesToSend.length} pending `);
});
this.channel.newConnection.subscribe(() => {
this.clearGraphOnNewConnection();
});
this.channel.serverMessage.subscribe((m) => {
parseJsonPatch(m.body, (p: Patch) => {
log('patch from server:', p.dump())
if (p.isEmpty()) {
return;
}
this.applyPatch(p);
});
});
}
sendPatch(patch: Patch) {
log("queue patch to server ", patch.summary());
this.patchesToSend.push(patch);
this._continueSending();
}
disconnect(why:string) {
this.channel.disconnect(why);
}
async _continueSending() {
// we could call this less often and coalesce patches together to optimize
// the dragging cases. See rdfdb 'compactPatches' and 'processInbox'.
while (this.patchesToSend.length) {
const patch = this.patchesToSend.splice(0, 1)[0];
const json = await patch.toJsonPatch();
const ret = this.channel.sendMessage(json);
if (!ret) {
log('sendMessage failed- retrying')
this.patchesToSend.unshift(patch);
setTimeout(this._continueSending.bind(this), 500);
// this.disconnect()
return;
}
}
}
}
|