import { isEqual } from "underscore" ; import * as lean_client from "lean-client-js-node" ; import { callback , delay } from "awaiting" ; import { once } from "@cocalc/util/async-utils" ; import { close } from "@cocalc/util/misc" ; import { reuseInFlight } from "async-await-utils/hof" ; import { EventEmitter } from "events" ; type SyncString = any ; type Client = any ; type LeanServer = any ; const SYNC_INTERVAL = 6000 ; type Message = any ; type Messages = Message [ ] ; interface State { tasks : any ; paths : { [ key ] : Messages ; } ; } export class Lean extends EventEmitter { paths : { [ key ] : SyncString ; } = { } ; client : Client ; _server : LeanServer | undefined ; _state : State = { tasks : [ ] , paths : { } } ; private running : { [ key ] : number ; } = { } ; dbg : Function ; constructor ( client ) { super ( ) ; this . server = reuseInFlight ( this . server ) ; this . client = client ; this . dbg = this . client . dbg ( "LEAN SERVER" ) ; this . running = { } ; } close ( ) { this . kill ( ) ; close ( this ) ; } is_running ( path ) { return ! ! this . running [ path ] && now ( ) - this . running [ path ] < SYNC_INTERVAL ; } private async server ( ) { if ( this . _server != undefined ) { if ( this . _server . alive ( ) ) { return this . _server ; } this . kill ( ) ; } this . _server = new lean_client . Server ( new lean_client . ProcessTransport ( "lean" , process . env . HOME ? process . env . HOME : "." , [ "-M 4096" ] ) ) ; this . _server . error . on ( ( err ) => this . dbg ( "error:" , err ) ) ; this . _server . allMessages . on ( ( allMessages ) => { this . dbg ( "messages: " , allMessages ) ; const new_messages = { } ; for ( const x of allMessages . msgs ) { const path = x . file_name ; delete x . file_name ; if ( new_messages [ path ] === undefined ) { new_messages [ path ] = [ x ] ; } else { new_messages [ path ] . push ( x ) ; } } for ( const path in this . _state . paths ) { this . dbg ( "messages for " , path , new_messages [ path ] ) ; if ( new_messages [ path ] === undefined ) { new_messages [ path ] = [ ] ; } this . dbg ( "messages for " , path , new_messages [ path ] , this . _state . paths [ path ] ) ; if ( new_messages [ path ] . length === 0 || ! isEqual ( this . _state . paths [ path ] , new_messages [ path ] ) ) { this . dbg ( "messages for " , path , "EMIT!" ) ; this . emit ( "messages" , path , new_messages [ path ] ) ; this . _state . paths [ path ] = new_messages [ path ] ; } } } ) ; this . _server . tasks . on ( ( currentTasks ) => { const { tasks } = currentTasks ; this . dbg ( "tasks: " , tasks ) ; const running = { } ; for ( const task of tasks ) { running [ task . file_name ] = true ; } for ( const path in running ) { const v = [ ] ; for ( const task of tasks ) { if ( task . file_name === path ) { delete task . file_name ; v . push ( task ) ; } } this . emit ( "tasks" , path , v ) ; } for ( const path in this . running ) { if ( ! running [ path ] ) { this . dbg ( "server" , path , " done; no longer running" ) ; this . running [ path ] = 0 ; this . emit ( "tasks" , path , [ ] ) ; if ( this . paths [ path ] . changed ) { this . dbg ( "server" , path , " changed while running, so running again" ) ; this . paths [ path ] . on_change ( ) ; } } } } ) ; this . _server . connect ( ) ; return this . _server ; } async register ( path ) { this . dbg ( "register" , path ) ; if ( this . paths [ path ] !== undefined ) { this . dbg ( "register" , path , "already registered" ) ; return ; } let syncstring = undefined ; while ( syncstring == null ) { syncstring = this . client . syncdoc ( { path } ) ; if ( syncstring == null ) { await delay ( 1000 ) ; } else if ( syncstring . get_state ( ) != "ready" ) { await once ( syncstring , "ready" ) ; } } const on_change = async ( ) => { this . dbg ( "sync" , path ) ; if ( syncstring . _closed ) { this . dbg ( "sync" , path , "closed" ) ; return ; } if ( this . is_running ( path ) ) { this . dbg ( "sync" , path , "already running" ) ; this . paths [ path ] . changed = true ; return ; } const value = syncstring . to_str ( ) ; if ( this . paths [ path ] . last_value === value ) { this . dbg ( "sync" , path , "skipping sync since value did not change" ) ; return ; } if ( value . trim ( ) === "" ) { this . dbg ( "sync" , path , "skipping sync document is empty (and LEAN behaves weird in this case)" ) ; this . emit ( "sync" , path , syncstring . hash_of_live_version ( ) ) ; return ; } this . paths [ path ] . last_value = value ; this . _state . paths [ path ] = [ ] ; this . running [ path ] = now ( ) ; this . paths [ path ] . changed = false ; this . dbg ( "sync" , path , "causing server sync now" ) ; await ( await this . server ( ) ) . sync ( path , value ) ; this . emit ( "sync" , path , syncstring . hash_of_live_version ( ) ) ; } ; this . paths [ path ] = { syncstring , on_change , } ; syncstring . on ( "change" , on_change ) ; if ( ! syncstring . _closed ) { on_change ( ) ; } syncstring . on ( "closed" , ( ) => { this . unregister ( path ) ; } ) ; } unregister ( path ) { this . dbg ( "unregister" , path ) ; if ( ! this . paths [ path ] ) { return ; } const x = this . paths [ path ] ; delete this . paths [ path ] ; delete this . running [ path ] ; x . syncstring . removeListener ( "change" , x . on_change ) ; x . syncstring . close ( ) ; } kill ( ) { this . dbg ( "kill" ) ; if ( this . _server != undefined ) { for ( const path in this . paths ) { this . unregister ( path ) ; } this . _server . dispose ( ) ; delete this . _server ; } } async restart ( ) { this . dbg ( "restart" ) ; if ( this . _server != undefined ) { for ( const path in this . paths ) { this . unregister ( path ) ; } await this . _server . restart ( ) ; } } async info ( path , line , column ) { this . dbg ( "info" , path , line , column ) ; if ( ! this . paths [ path ] ) { this . register ( path ) ; await callback ( ( cb ) => this . once ( `sync-#{path}` , cb ) ) ; } return await ( await this . server ( ) ) . info ( path , line , column ) ; } async complete ( path , line , column , skipCompletions ? ) { this . dbg ( "complete" , path , line , column ) ; if ( ! this . paths [ path ] ) { this . register ( path ) ; await callback ( ( cb ) => this . once ( `sync-#{path}` , cb ) ) ; } const resp = await ( await this . server ( ) ) . complete ( path , line , column , skipCompletions ) ; return resp ; } async version ( ) { return ( await this . server ( ) ) . getVersion ( ) ; } state ( ) { return this . _state ; } messages ( path ) { const x = this . _state . paths [ path ] ; if ( x !== undefined ) { return x ; } return [ ] ; } } let singleton ; export function lean_server ( client ) { if ( singleton === undefined ) { singleton = new Lean ( client ) ; } return singleton ; } function now ( ) { return new Date ( ) . valueOf ( ) ; }