module import * as Y from 'yjs' ; import * as bc from 'lib0/broadcastchannel.js' ; import * as time from 'lib0/time.js' ; import * as encoding from 'lib0/encoding.js' ; import * as decoding from 'lib0/decoding.js' ; import * as syncProtocol from 'y-protocols/sync.js' ; import * as authProtocol from 'y-protocols/auth.js' ; import * as awarenessProtocol from 'y-protocols/awareness.js' ; import * as mutex from 'lib0/mutex.js' ; import { Observable } from 'lib0/observable.js' ; import * as math from 'lib0/math.js' ; import * as url from 'lib0/url.js' ; const messageSync = 0 ; const messageQueryAwareness = 3 ; const messageAwareness = 1 ; const messageAuth = 2 ; const reconnectTimeoutBase = 1200 ; const maxReconnectTimeout = 2500 ; const messageReconnectTimeout = 30000 ; param { WebsocketProvider } provider param { string } reason const permissionDeniedHandler = ( provider , reason ) => console . warn ( `Permission denied to access ${ provider . url }.\n${ reason }` ) ; param { WebsocketProvider } provider param { Uint8Array } buf param { boolean } emitSynced return { encoding . Encoder } const readMessage = ( provider , buf , emitSynced ) => { const decoder = decoding . createDecoder ( buf ) ; const encoder = encoding . createEncoder ( ) ; const messageType = decoding . readVarUint ( decoder ) ; switch ( messageType ) { case messageSync : { encoding . writeVarUint ( encoder , messageSync ) ; const syncMessageType = syncProtocol . readSyncMessage ( decoder , encoder , provider . doc , provider ) ; if ( emitSynced && syncMessageType === syncProtocol . messageYjsSyncStep2 && ! provider . synced ) { provider . synced = true ; } break ; } case messageQueryAwareness : encoding . writeVarUint ( encoder , messageAwareness ) ; encoding . writeVarUint8Array ( encoder , awarenessProtocol . encodeAwarenessUpdate ( provider . awareness , Array . from ( provider . awareness . getStates ( ) . keys ( ) ) ) ) ; break ; case messageAwareness : awarenessProtocol . applyAwarenessUpdate ( provider . awareness , decoding . readVarUint8Array ( decoder ) , provider ) ; break ; case messageAuth : authProtocol . readAuthMessage ( decoder , provider . doc , permissionDeniedHandler ) ; break ; default : console . error ( 'Unable to compute message' ) ; return encoder ; } return encoder ; } ; param { WebsocketProvider } provider const setupWS = provider => { if ( provider . shouldConnect && provider . ws === null ) { const websocket = new WebSocket ( provider . url ) ; websocket . binaryType = 'arraybuffer' ; provider . ws = websocket ; provider . wsconnecting = true ; provider . wsconnected = false ; provider . synced = false ; websocket . onmessage = event => { provider . wsLastMessageReceived = time . getUnixTime ( ) ; const encoder = readMessage ( provider , new Uint8Array ( event . data ) , true ) ; if ( encoding . length ( encoder ) > 1 ) { websocket . send ( encoding . toUint8Array ( encoder ) ) ; } } ; websocket . onclose = ( ) => { provider . ws = null ; provider . wsconnecting = false ; if ( provider . wsconnected ) { provider . wsconnected = false ; provider . synced = false ; awarenessProtocol . removeAwarenessStates ( provider . awareness , Array . from ( provider . awareness . getStates ( ) . keys ( ) ) , provider ) ; provider . emit ( 'status' , [ { status : 'disconnected' } ] ) ; } else { provider . wsUnsuccessfulReconnects ++ ; } setTimeout ( setupWS , math . min ( math . log10 ( provider . wsUnsuccessfulReconnects + 1 ) * reconnectTimeoutBase , maxReconnectTimeout ) , provider ) ; } ; websocket . onopen = ( ) => { provider . wsLastMessageReceived = time . getUnixTime ( ) ; provider . wsconnecting = false ; provider . wsconnected = true ; provider . wsUnsuccessfulReconnects = 0 ; provider . emit ( 'status' , [ { status : 'connected' } ] ) ; const encoder = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoder , messageSync ) ; syncProtocol . writeSyncStep1 ( encoder , provider . doc ) ; websocket . send ( encoding . toUint8Array ( encoder ) ) ; if ( provider . awareness . getLocalState ( ) !== null ) { const encoderAwarenessState = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoderAwarenessState , messageAwareness ) ; encoding . writeVarUint8Array ( encoderAwarenessState , awarenessProtocol . encodeAwarenessUpdate ( provider . awareness , [ provider . doc . clientID ] ) ) ; websocket . send ( encoding . toUint8Array ( encoderAwarenessState ) ) ; } } ; } } ; param { WebsocketProvider } provider param { ArrayBuffer } buf const broadcastMessage = ( provider , buf ) => { if ( provider . wsconnected ) { provider . ws . send ( buf ) ; } if ( provider . bcconnected ) { provider . mux ( ( ) => { bc . publish ( provider . url , buf ) ; } ) ; } } ; example extends Observable < string > export class WebsocketProvider extends Observable { param { string } serverUrl param { string } roomname param { Y . Doc } doc param { { connect : boolean , awareness : awarenessProtocol . Awareness , db : any | null , params : Object < string , string > } } conf constructor ( serverUrl , roomname , doc , { connect = true , awareness = new awarenessProtocol . Awareness ( doc ) , db = null , params = { } } = type { any } ( { } ) ) { super ( ) ; while ( serverUrl [ serverUrl . length - 1 ] === '/' ) { serverUrl = serverUrl . slice ( 0 , serverUrl . length - 1 ) ; } const encodedParams = url . encodeQueryParams ( params ) ; this . bcChannel = serverUrl + '/' + roomname ; this . url = serverUrl + '/' + roomname + ( encodedParams . length === 0 ? '' : '?' + encodedParams ) ; this . roomname = roomname ; this . doc = doc ; type { Object < string , Object > } this . _localAwarenessState = { } ; this . db = db ; this . awareness = awareness ; this . wsconnected = false ; this . wsconnecting = false ; this . bcconnected = false ; this . wsUnsuccessfulReconnects = 0 ; this . mux = mutex . createMutex ( ) ; type { boolean } this . _synced = false ; type { WebSocket ? } this . ws = null ; this . wsLastMessageReceived = 0 ; type { boolean } this . shouldConnect = connect ; param { ArrayBuffer } data this . _bcSubscriber = data => { this . mux ( ( ) => { const encoder = readMessage ( this , new Uint8Array ( data ) , false ) ; if ( encoding . length ( encoder ) > 1 ) { bc . publish ( this . bcChannel , encoding . toUint8Array ( encoder ) ) ; } } ) ; } ; param { Uint8Array } update param { any } origin this . _updateHandler = ( update , origin ) => { if ( origin !== this || origin === null ) { const encoder = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoder , messageSync ) ; syncProtocol . writeUpdate ( encoder , update ) ; broadcastMessage ( this , encoding . toUint8Array ( encoder ) ) ; } } ; this . doc . on ( 'update' , this . _updateHandler ) ; param { any } changed param { any } origin this . _awarenessUpdateHandler = ( { added , updated , removed } , origin ) => { const changedClients = added . concat ( updated ) . concat ( removed ) ; const encoder = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoder , messageAwareness ) ; encoding . writeVarUint8Array ( encoder , awarenessProtocol . encodeAwarenessUpdate ( awareness , changedClients ) ) ; broadcastMessage ( this , encoding . toUint8Array ( encoder ) ) ; } ; window . addEventListener ( 'beforeunload' , ( ) => { awarenessProtocol . removeAwarenessStates ( this . awareness , [ doc . clientID ] , 'window unload' ) ; } ) ; awareness . on ( 'change' , this . _awarenessUpdateHandler ) ; this . _checkInterval = setInterval ( ( ) => { if ( this . wsconnected && messageReconnectTimeout < time . getUnixTime ( ) - this . wsLastMessageReceived ) { type { WebSocket } ( this . ws ) . close ( ) ; } } , messageReconnectTimeout / 10 ) ; if ( connect ) { this . connect ( ) ; } } type { boolean } get synced ( ) { return this . _synced ; } set synced ( state ) { if ( this . _synced !== state ) { this . _synced = state ; this . emit ( 'sync' , [ state ] ) ; } } destroy ( ) { clearInterval ( this . _checkInterval ) ; this . disconnect ( ) ; this . awareness . off ( 'change' , this . _awarenessUpdateHandler ) ; this . doc . off ( 'update' , this . _updateHandler ) ; super . destroy ( ) ; } connectBc ( ) { if ( ! this . bcconnected ) { bc . subscribe ( this . bcChannel , this . _bcSubscriber ) ; this . bcconnected = true ; } this . mux ( ( ) => { const encoderSync = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoderSync , messageSync ) ; syncProtocol . writeSyncStep1 ( encoderSync , this . doc ) ; bc . publish ( this . bcChannel , encoding . toUint8Array ( encoderSync ) ) ; const encoderState = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoderState , messageSync ) ; syncProtocol . writeSyncStep2 ( encoderState , this . doc ) ; bc . publish ( this . bcChannel , encoding . toUint8Array ( encoderState ) ) ; const encoderAwarenessQuery = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoderAwarenessQuery , messageQueryAwareness ) ; bc . publish ( this . bcChannel , encoding . toUint8Array ( encoderAwarenessQuery ) ) ; const encoderAwarenessState = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoderAwarenessState , messageAwareness ) ; encoding . writeVarUint8Array ( encoderAwarenessState , awarenessProtocol . encodeAwarenessUpdate ( this . awareness , [ this . doc . clientID ] ) ) ; bc . publish ( this . bcChannel , encoding . toUint8Array ( encoderAwarenessState ) ) ; } ) ; } disconnectBc ( ) { const encoder = encoding . createEncoder ( ) ; encoding . writeVarUint ( encoder , messageAwareness ) ; encoding . writeVarUint8Array ( encoder , awarenessProtocol . encodeAwarenessUpdate ( this . awareness , [ this . doc . clientID ] , new Map ( ) ) ) ; broadcastMessage ( this , encoding . toUint8Array ( encoder ) ) ; if ( this . bcconnected ) { bc . unsubscribe ( this . bcChannel , this . _bcSubscriber ) ; this . bcconnected = false ; } } disconnect ( ) { this . shouldConnect = false ; this . disconnectBc ( ) ; if ( this . ws !== null ) { this . ws . close ( ) ; } } connect ( ) { this . shouldConnect = true ; if ( ! this . wsconnected && this . ws === null ) { setupWS ( this ) ; this . connectBc ( ) ; } } }