Changeset - f2c6b39c155c
[Not reviewed]
default
0 2 0
drewp@bigasterisk.com - 20 months ago 2023-05-30 05:49:40
drewp@bigasterisk.com
fix messagesSend counter
2 files changed with 4 insertions and 3 deletions:
0 comments (0 inline, 0 general)
light9/web/RdfDbChannel.ts
Show inline comments
 
@@ -31,48 +31,50 @@ class ChannelPinger {
 

	
 
export class RdfDbChannel {
 
  // lower level reconnecting websocket -- knows about message types, but not what's inside a patch body
 
  private ws?: WebSocket = undefined;
 
  private pinger?: ChannelPinger;
 
  private connectionId: string = "none"; // server's name for us
 
  private reconnectTimer?: NodeJS.Timeout = undefined;
 
  private messagesReceived = 0; // (non-ping messages)
 
  private messagesSent = 0;
 

	
 
  newConnection: SubEvent<void> = new SubEvent();
 
  serverMessage: SubEvent<{ evType: string; body: SyncgraphPatchMessage }> = new SubEvent();
 
  statusDisplay: SubEvent<string> = new SubEvent();
 

	
 
  constructor(public patchSenderUrl: string) {
 
    this.openConnection();
 
  }
 
  sendMessage(body: string): boolean {
 
    // one try, best effort, true if we think it worked
 
    if (!this.ws || this.ws.readyState !== this.ws.OPEN) {
 
      return false;
 
    }
 
    log("send patch to server, " + body.length + " bytes");
 
    this.ws.send(body);
 
    this.messagesSent++;
 
    this.updateStatus();
 
    return true;
 
  }
 

	
 
  disconnect() {
 
    // will be followed by an autoconnect
 
    log("disconnect requested");
 
    if (this.ws !== undefined) {
 
      const closeHandler = this.ws.onclose?.bind(this.ws);
 
      if (!closeHandler) {
 
        throw new Error();
 
      }
 
      closeHandler(new CloseEvent("forced"));
 
    }
 
  }
 

	
 
  private openConnection() {
 
    const wsOrWss = window.location.protocol.replace("http", "ws");
 
    const fullUrl = wsOrWss + "//" + window.location.host + this.patchSenderUrl;
 
    if (this.ws !== undefined) {
 
      this.ws.close();
 
    }
 
    this.ws = new WebSocket(fullUrl);
 
    this.ws.onopen = this.onWsOpen.bind(this, this.ws);
 
    this.ws.onerror = this.onWsError.bind(this);
light9/web/rdfdbclient.ts
Show inline comments
 
import debug from "debug";
 
import { parseJsonPatch, Patch } from "./patch";
 
import { RdfDbChannel } from "./RdfDbChannel";
 
export const log = debug("rdfdbclient");
 
const log = debug("rdfdbclient");
 

	
 
export class RdfDbClient {
 
  private channel: RdfDbChannel;
 
  _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) => {
 
      log('got server json', m.body)
 
      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() {
 
    this.channel.disconnect();
 
  }
 

	
 
  async _continueSending() {
 
    // we could call this less often and coalesce patches together to optimize
 
    // the dragging cases.
 
    // 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) {
 
        setTimeout(this._continueSending.bind(this), 500);
 

	
 
        // this.disconnect()
 
        return;
 
      }
 
    }
 
  }
 
}
0 comments (0 inline, 0 general)