Events for sync status

Author: paulgb

crates/y-sweet-core/src/doc_connection.rs

@@ -22,6 +22,8 @@ type Callback = Arc<dyn Fn(&[u8]) + 'static>;
 #[cfg(feature = "sync")]
 type Callback = Arc<dyn Fn(&[u8]) + 'static + Send + Sync>;
 
+const SYNC_STATUS_MESSAGE: u8 = 102;
+
 pub struct DocConnection {
     awareness: Arc<RwLock<Awareness>>,
     #[allow(unused)] // acts as RAII guard
@@ -184,6 +186,10 @@ impl DocConnection {
                 let mut awareness = a.write().unwrap();
                 protocol.handle_awareness_update(&mut awareness, update)
             }
+            Message::Custom(SYNC_STATUS_MESSAGE, data) => {
+                // Respond to the client with the same payload it sent.
+                Ok(Some(Message::Custom(SYNC_STATUS_MESSAGE, data)))
+            }
             Message::Custom(tag, data) => {
                 let mut awareness = a.write().unwrap();
                 protocol.missing_handle(&mut awareness, tag, data)

crates/y-sweet/src/server.rs

@@ -17,7 +17,6 @@ use futures::{SinkExt, StreamExt};
 use serde::Deserialize;
 use serde_json::{json, Value};
 use std::{
-    net::SocketAddr,
     sync::{Arc, RwLock},
     time::Duration,
 };

js-pkg/client/src/provider.ts

@@ -8,11 +8,16 @@ import * as Y from 'yjs'
 const messageSync = 0
 const messageQueryAwareness = 3
 const messageAwareness = 1
+const messageSyncStatus = 102
 
 const EVENT_STATUS = 'status'
+
+/** Fired when the _initial_ sync is complete. Only refired after that if there is a reconnection. */
 const EVENT_SYNC = 'sync'
 const EVENT_CONNECTION_CLOSE = 'connection-close'
 const EVENT_CONNECTION_ERROR = 'connection-error'
+/** Fired every time the sync status changes. */
+const EVENT_SYNC_STATUS = 'sync-status'
 
 /**
  * Note: this should always be a superset of y-websocket's valid events.
@@ -23,6 +28,7 @@ type YSweetEvent =
   | typeof EVENT_SYNC
   | typeof EVENT_CONNECTION_CLOSE
   | typeof EVENT_CONNECTION_ERROR
+  | typeof EVENT_SYNC_STATUS
 
 const STATUS_CONNECTED = 'connected'
 const STATUS_DISCONNECTED = 'disconnected'
@@ -101,12 +107,16 @@ async function getClientToken(authEndpoint: AuthEndpoint, roomname: string): Pro
 export class YSweetProvider {
   private websocket: WebSocket | null = null
   public clientToken: ClientToken | null = null
-  public synced: boolean = false
+  private initialSync: boolean = false
+  private synced: boolean = false
   private status: YSweetStatus = { status: STATUS_DISCONNECTED }
   public awareness: awarenessProtocol.Awareness
   private WebSocketPolyfill: WebSocketPolyfillType
   private listeners: Map<YSweetEvent, Set<EventListener>> = new Map()
 
+  private lastSyncSent: number = 0
+  private lastSyncAcked: number = 0
+
   /** Whether we should attempt to connect if we are in a disconnected state. */
   private shouldConnect: boolean
 
@@ -135,6 +145,16 @@ export class YSweetProvider {
     doc.on('update', this.update.bind(this))
   }
 
+  private updateSyncedState() {
+    if (this.lastSyncAcked === this.lastSyncSent) {
+      this.synced = true
+      this.emit(EVENT_SYNC_STATUS, true)
+    } else {
+      this.synced = false
+      this.emit(EVENT_SYNC_STATUS, false)
+    }
+  }
+
   private update(update: Uint8Array, origin: YSweetProvider) {
     if (!this.websocket) {
       console.warn('Websocket not connected')
@@ -146,9 +166,30 @@ export class YSweetProvider {
       encoding.writeVarUint(encoder, messageSync)
       syncProtocol.writeUpdate(encoder, update)
       this.websocket.send(encoding.toUint8Array(encoder))
+
+      this.checkSync()
     }
   }
 
+  private checkSync() {
+    if (!this.websocket) {
+      console.warn('Websocket not connected')
+      return
+    }
+
+    this.lastSyncSent += 1
+    const encoder = encoding.createEncoder()
+    encoding.writeVarUint(encoder, messageSyncStatus)
+
+    const versionEncoder = encoding.createEncoder()
+    encoding.writeVarUint(versionEncoder, this.lastSyncSent)
+
+    encoding.writeVarUint8Array(encoder, encoding.toUint8Array(versionEncoder))
+    this.websocket.send(encoding.toUint8Array(encoder))
+
+    this.updateSyncedState()
+  }
+
   private async ensureClientToken(): Promise<ClientToken> {
     if (this.clientToken) {
       return this.clientToken
@@ -319,6 +360,10 @@ export class YSweetProvider {
       case messageQueryAwareness:
         this.queryAwareness()
         break
+      case messageSyncStatus:
+        this.lastSyncAcked = decoding.readVarUint(decoder)
+        this.updateSyncedState()
+        break
       default:
         break
     }
@@ -356,6 +401,7 @@ export class YSweetProvider {
   }
 
   protected emit(eventName: YSweetEvent, data: any = null): void {
+    console.log('Emitting event', eventName, data)
     const listeners = this.listeners.get(eventName)
     if (listeners) {
       for (const listener of listeners) {
@@ -365,8 +411,8 @@ export class YSweetProvider {
   }
 
   private setSynced(state: boolean) {
-    if (this.synced !== state) {
-      this.synced = state
+    if (this.initialSync !== state) {
+      this.initialSync = state
       this.emit('sync', state)
     }
   }