/********************************************************************************
 * Copyright (C) 2020 TypeFox and others.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0.
 *
 * This Source Code may also be made available under the following Secondary
 * Licenses when the conditions for such availability set forth in the Eclipse
 * Public License v. 2.0 are satisfied: GNU General Public License, version 2
 * with the GNU Classpath Exception which is available at
 * https://www.gnu.org/software/classpath/license.html.
 *
 * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 ********************************************************************************/ import { injectable , inject , postConstruct } from '@theia/core/shared/inversify' ; import URI from '@theia/core/lib/common/uri' ; import { Emitter } from '@theia/core/lib/common/event' ; import { Disposable , DisposableCollection } from '@theia/core/lib/common/disposable' ; import { BinaryBuffer } from '@theia/core/lib/common/buffer' ; import { FileWriteOptions , FileOpenOptions , FileChangeType , FileSystemProviderCapabilities , FileChange , Stat , FileOverwriteOptions , WatchOptions , FileType , FileSystemProvider , FileDeleteOptions , hasOpenReadWriteCloseCapability , hasFileFolderCopyCapability , hasReadWriteCapability , hasAccessCapability , FileSystemProviderError , FileSystemProviderErrorCode , FileUpdateOptions , hasUpdateCapability , FileUpdateResult , FileReadStreamOptions , hasFileReadStreamCapability } from './files' ; import { JsonRpcServer , JsonRpcProxy , JsonRpcProxyFactory } from '@theia/core/lib/common/messaging/proxy-factory' ; import { ApplicationError } from '@theia/core/lib/common/application-error' ; import { Deferred } from '@theia/core/lib/common/promise-util' ; import type { TextDocumentContentChangeEvent } from '@theia/core/shared/vscode-languageserver-protocol' ; import { newWriteableStream , ReadableStreamEvents } from '@theia/core/lib/common/stream' ; import { CancellationToken , cancelled } from '@theia/core/lib/common/cancellation' ; export const remoteFileSystemPath = '/services/remote-filesystem' ; export const RemoteFileSystemServer = Symbol ( 'RemoteFileSystemServer' ) ; export interface RemoteFileSystemServer extends JsonRpcServer < RemoteFileSystemClient > { getCapabilities ( ) : Promise < FileSystemProviderCapabilities > ; stat ( resource ) : Promise < Stat > ; access ( resource , mode ? ) : Promise < void > ; fsPath ( resource ) : Promise < string > ; open ( resource , opts ) : Promise < number > ; close ( fd ) : Promise < void > ; read ( fd , pos , length ) : Promise < { bytes : number [ ] ; bytesRead : number ; } > ; readFileStream ( resource , opts , token ) : Promise < number > ; readFile ( resource ) : Promise < number [ ] > ; write ( fd , pos , data , offset , length ) : Promise < number > ; writeFile ( resource , content , opts ) : Promise < void > ; delete ( resource , opts ) : Promise < void > ; mkdir ( resource ) : Promise < void > ; readdir ( resource ) : Promise < [ string , FileType ] [ ] > ; rename ( source , target , opts ) : Promise < void > ; copy ( source , target , opts ) : Promise < void > ; watch ( watcher , resource , opts ) : Promise < void > ; unwatch ( watcher ) : Promise < void > ; updateFile ( resource , changes , opts ) : Promise < FileUpdateResult > ; } export interface RemoteFileChange { readonly type : FileChangeType ; readonly resource : string ; } export interface RemoteFileStreamError extends Error { code ? : FileSystemProviderErrorCode ; } export interface RemoteFileSystemClient { notifyDidChangeFile ( event ) : void ; notifyFileWatchError ( ) : void ; notifyDidChangeCapabilities ( capabilities ) : void ; onFileStreamData ( handle , data ) : void ; onFileStreamEnd ( handle , error ) : void ; } export const RemoteFileSystemProviderError = ApplicationError . declare ( - 33005 , ( message , data , stack ) => ( { message , data , stack } ) ) ; export class RemoteFileSystemProxyFactory < T extends object > extends JsonRpcProxyFactory < T > { protected serializeError ( e ) { if ( e instanceof FileSystemProviderError ) { const { code , name } = e ; return super . serializeError ( RemoteFileSystemProviderError ( e . message , { code , name } , e . stack ) ) ; } return super . serializeError ( e ) ; } protected deserializeError ( capturedError , e ) { const error = super . deserializeError ( capturedError , e ) ; if ( RemoteFileSystemProviderError . is ( error ) ) { const fileOperationError = new FileSystemProviderError ( error . message , error . data . code ) ; fileOperationError . name = error . data . name ; fileOperationError . stack = error . stack ; return fileOperationError ; } return e ; } } /**
 * Frontend component.
 *
 * Wraps the remote filesystem provider living on the backend.
 */ @ injectable ( ) export class RemoteFileSystemProvider implements Required < FileSystemProvider > , Disposable { private readonly onDidChangeFileEmitter = new Emitter < readonly FileChange [ ] > ( ) ; readonly onDidChangeFile = this . onDidChangeFileEmitter . event ; private readonly onFileWatchErrorEmitter = new Emitter < void > ( ) ; readonly onFileWatchError = this . onFileWatchErrorEmitter . event ; private readonly onDidChangeCapabilitiesEmitter = new Emitter < void > ( ) ; readonly onDidChangeCapabilities = this . onDidChangeCapabilitiesEmitter . event ; private readonly onFileStreamDataEmitter = new Emitter < [ number , Uint8Array ] > ( ) ; private readonly onFileStreamData = this . onFileStreamDataEmitter . event ; private readonly onFileStreamEndEmitter = new Emitter < [ number , Error | FileSystemProviderError | undefined ] > ( ) ; private readonly onFileStreamEnd = this . onFileStreamEndEmitter . event ; protected readonly toDispose = new DisposableCollection ( this . onDidChangeFileEmitter , this . onDidChangeCapabilitiesEmitter , this . onFileStreamDataEmitter , this . onFileStreamEndEmitter ) ; protected watcherSequence = 0 ; /**
     * We'll track the currently allocated watchers, in order to re-allocate them
     * with the same options once we reconnect to the backend after a disconnection.
     */ protected readonly watchOptions = new Map < number , { uri : string ; options : WatchOptions ; } > ( ) ; private _capabilities : FileSystemProviderCapabilities = 0 ; get capabilities ( ) : FileSystemProviderCapabilities { return this . _capabilities ; } protected readonly readyDeferred = new Deferred < void > ( ) ; readonly ready = this . readyDeferred . promise ; /**
     * Wrapped remote filesystem.
     */ @ inject ( RemoteFileSystemServer ) protected readonly server : JsonRpcProxy < RemoteFileSystemServer > ; @ postConstruct ( ) protected init ( ) { this . server . getCapabilities ( ) . then ( capabilities => { this . _capabilities = capabilities ; this . readyDeferred . resolve ( ) ; } , this . readyDeferred . reject ) ; this . server . setClient ( { notifyDidChangeFile : ( { changes } ) => { this . onDidChangeFileEmitter . fire ( changes . map ( event => ( { resource : new URI ( event . resource ) , type : event . type } ) ) ) ; } , notifyFileWatchError : ( ) => { this . onFileWatchErrorEmitter . fire ( ) ; } , notifyDidChangeCapabilities : capabilities => this . setCapabilities ( capabilities ) , onFileStreamData : ( handle , data ) => this . onFileStreamDataEmitter . fire ( [ handle , Uint8Array . from ( data ) ] ) , onFileStreamEnd : ( handle , error ) => this . onFileStreamEndEmitter . fire ( [ handle , error ] ) } ) ; const onInitialized = this . server . onDidOpenConnection ( ( ) => { onInitialized . dispose ( ) ; this . toDispose . push ( this . server . onDidOpenConnection ( ( ) => this . reconnect ( ) ) ) ; } ) ; } dispose ( ) { this . toDispose . dispose ( ) ; } protected setCapabilities ( capabilities ) { this . _capabilities = capabilities ; this . onDidChangeCapabilitiesEmitter . fire ( undefined ) ; } stat ( resource ) { return this . server . stat ( resource . toString ( ) ) ; } access ( resource , mode ? ) { return this . server . access ( resource . toString ( ) , mode ) ; } fsPath ( resource ) { return this . server . fsPath ( resource . toString ( ) ) ; } open ( resource , opts ) { return this . server . open ( resource . toString ( ) , opts ) ; } close ( fd ) { return this . server . close ( fd ) ; } async read ( fd , pos , data , offset , length ) { const { bytes , bytesRead } = await this . server . read ( fd , pos , length ) ; data . set ( bytes . slice ( 0 , bytesRead ) , offset ) ; return bytesRead ; } async readFile ( resource ) { const bytes = await this . server . readFile ( resource . toString ( ) ) ; return Uint8Array . from ( bytes ) ; } readFileStream ( resource , opts , token ) { const capturedError = new Error ( ) ; const stream = newWriteableStream < Uint8Array > ( data => BinaryBuffer . concat ( data . map ( data => BinaryBuffer . wrap ( data ) ) ) . buffer ) ; this . server . readFileStream ( resource . toString ( ) , opts , token ) . then ( streamHandle => { if ( token . isCancellationRequested ) { stream . end ( cancelled ( ) ) ; return ; } const toDispose = new DisposableCollection ( token . onCancellationRequested ( ( ) => stream . end ( cancelled ( ) ) ) , this . onFileStreamData ( ( [ handle , data ] ) => { if ( streamHandle === handle ) { stream . write ( data ) ; } } ) , this . onFileStreamEnd ( ( [ handle , error ] ) => { if ( streamHandle === handle ) { if ( error ) { const code = ( 'code' in error && error . code ) || FileSystemProviderErrorCode . Unknown ; const fileOperationError = new FileSystemProviderError ( error . message , code ) ; fileOperationError . name = error . name ; const capturedStack = capturedError . stack || '' ; fileOperationError . stack = `${ capturedStack }\nCaused by: ${ error . stack }` ; stream . end ( fileOperationError ) ; } else { stream . end ( ) ; } } } ) ) ; stream . on ( 'end' , ( ) => toDispose . dispose ( ) ) ; } , error => stream . end ( error ) ) ; return stream ; } write ( fd , pos , data , offset , length ) { return this . server . write ( fd , pos , [ ... data . values ( ) ] , offset , length ) ; } writeFile ( resource , content , opts ) { return this . server . writeFile ( resource . toString ( ) , [ ... content . values ( ) ] , opts ) ; } delete ( resource , opts ) { return this . server . delete ( resource . toString ( ) , opts ) ; } mkdir ( resource ) { return this . server . mkdir ( resource . toString ( ) ) ; } readdir ( resource ) { return this . server . readdir ( resource . toString ( ) ) ; } rename ( resource , target , opts ) { return this . server . rename ( resource . toString ( ) , target . toString ( ) , opts ) ; } copy ( resource , target , opts ) { return this . server . copy ( resource . toString ( ) , target . toString ( ) , opts ) ; } updateFile ( resource , changes , opts ) { return this . server . updateFile ( resource . toString ( ) , changes , opts ) ; } watch ( resource , options ) { const watcherId = this . watcherSequence ++ ; const uri = resource . toString ( ) ; this . watchOptions . set ( watcherId , { uri , options } ) ; this . server . watch ( watcherId , uri , options ) ; const toUnwatch = Disposable . create ( ( ) => { this . watchOptions . delete ( watcherId ) ; this . server . unwatch ( watcherId ) ; } ) ; this . toDispose . push ( toUnwatch ) ; return toUnwatch ; } /**
     * When a frontend disconnects (e.g. bad connection) the backend resources will be cleared.
     *
     * This means that we need to re-allocate the watchers when a frontend reconnects.
     */ protected reconnect ( ) { for ( const [ watcher , { uri , options } ] of this . watchOptions . entries ( ) ) { this . server . watch ( watcher , uri , options ) ; } } } /**
 * Backend component.
 *
 * JSON-RPC server exposing a wrapped file system provider remotely.
 */ @ injectable ( ) export class FileSystemProviderServer implements RemoteFileSystemServer { private readonly BUFFER_SIZE = 64 * 1024 ; /**
     * Mapping of `watcherId` to a disposable watcher handle.
     */ protected watchers = new Map < number , Disposable > ( ) ; protected readonly toDispose = new DisposableCollection ( ) ; dispose ( ) { this . toDispose . dispose ( ) ; } protected client : RemoteFileSystemClient | undefined ; setClient ( client ) { this . client = client ; } /**
     * Wrapped file system provider.
     */ @ inject ( FileSystemProvider ) protected readonly provider : FileSystemProvider & Partial < Disposable > ; @ postConstruct ( ) protected init ( ) { if ( this . provider . dispose ) { this . toDispose . push ( Disposable . create ( ( ) => this . provider . dispose ! ( ) ) ) ; } this . toDispose . push ( this . provider . onDidChangeCapabilities ( ( ) => { if ( this . client ) { this . client . notifyDidChangeCapabilities ( this . provider . capabilities ) ; } } ) ) ; this . toDispose . push ( this . provider . onDidChangeFile ( changes => { if ( this . client ) { this . client . notifyDidChangeFile ( { changes : changes . map ( ( { resource , type } ) => ( { resource : resource . toString ( ) , type } ) ) } ) ; } } ) ) ; this . toDispose . push ( this . provider . onFileWatchError ( ( ) => { if ( this . client ) { this . client . notifyFileWatchError ( ) ; } } ) ) ; } async getCapabilities ( ) { return this . provider . capabilities ; } stat ( resource ) { return this . provider . stat ( new URI ( resource ) ) ; } access ( resource , mode ? ) { if ( hasAccessCapability ( this . provider ) ) { return this . provider . access ( new URI ( resource ) , mode ) ; } throw new Error ( 'not supported' ) ; } async fsPath ( resource ) { if ( hasAccessCapability ( this . provider ) ) { return this . provider . fsPath ( new URI ( resource ) ) ; } throw new Error ( 'not supported' ) ; } open ( resource , opts ) { if ( hasOpenReadWriteCloseCapability ( this . provider ) ) { return this . provider . open ( new URI ( resource ) , opts ) ; } throw new Error ( 'not supported' ) ; } close ( fd ) { if ( hasOpenReadWriteCloseCapability ( this . provider ) ) { return this . provider . close ( fd ) ; } throw new Error ( 'not supported' ) ; } async read ( fd , pos , length ) { if ( hasOpenReadWriteCloseCapability ( this . provider ) ) { const buffer = BinaryBuffer . alloc ( this . BUFFER_SIZE ) ; const bytes = buffer . buffer ; const bytesRead = await this . provider . read ( fd , pos , bytes , 0 , length ) ; return { bytes : [ ... bytes . values ( ) ] , bytesRead } ; } throw new Error ( 'not supported' ) ; } write ( fd , pos , data , offset , length ) { if ( hasOpenReadWriteCloseCapability ( this . provider ) ) { return this . provider . write ( fd , pos , Uint8Array . from ( data ) , offset , length ) ; } throw new Error ( 'not supported' ) ; } async readFile ( resource ) { if ( hasReadWriteCapability ( this . provider ) ) { const buffer = await this . provider . readFile ( new URI ( resource ) ) ; return [ ... buffer . values ( ) ] ; } throw new Error ( 'not supported' ) ; } writeFile ( resource , content , opts ) { if ( hasReadWriteCapability ( this . provider ) ) { return this . provider . writeFile ( new URI ( resource ) , Uint8Array . from ( content ) , opts ) ; } throw new Error ( 'not supported' ) ; } delete ( resource , opts ) { return this . provider . delete ( new URI ( resource ) , opts ) ; } mkdir ( resource ) { return this . provider . mkdir ( new URI ( resource ) ) ; } readdir ( resource ) { return this . provider . readdir ( new URI ( resource ) ) ; } rename ( source , target , opts ) { return this . provider . rename ( new URI ( source ) , new URI ( target ) , opts ) ; } copy ( source , target , opts ) { if ( hasFileFolderCopyCapability ( this . provider ) ) { return this . provider . copy ( new URI ( source ) , new URI ( target ) , opts ) ; } throw new Error ( 'not supported' ) ; } updateFile ( resource , changes , opts ) { if ( hasUpdateCapability ( this . provider ) ) { return this . provider . updateFile ( new URI ( resource ) , changes , opts ) ; } throw new Error ( 'not supported' ) ; } async watch ( requestedWatcherId , resource , opts ) { if ( this . watchers . has ( requestedWatcherId ) ) { throw new Error ( 'watcher id is already allocated!' ) ; } const watcher = this . provider . watch ( new URI ( resource ) , opts ) ; this . watchers . set ( requestedWatcherId , watcher ) ; this . toDispose . push ( Disposable . create ( ( ) => this . unwatch ( requestedWatcherId ) ) ) ; } async unwatch ( watcherId ) { const watcher = this . watchers . get ( watcherId ) ; if ( watcher ) { this . watchers . delete ( watcherId ) ; watcher . dispose ( ) ; } } protected readFileStreamSeq = 0 ; async readFileStream ( resource , opts , token ) { if ( hasFileReadStreamCapability ( this . provider ) ) { const handle = this . readFileStreamSeq ++ ; const stream = this . provider . readFileStream ( new URI ( resource ) , opts , token ) ; stream . on ( 'data' , data => this . client ?. onFileStreamData ( handle , [ ... data . values ( ) ] ) ) ; stream . on ( 'error' , error => { const code = error instanceof FileSystemProviderError ? error . code : undefined ; const { name , message , stack } = error ; this . client ?. onFileStreamEnd ( handle , { code , name , message , stack } ) ; } ) ; stream . on ( 'end' , ( ) => this . client ?. onFileStreamEnd ( handle , undefined ) ) ; return handle ; } throw new Error ( 'not supported' ) ; } }