packageDocumentation module import * as fs from "fs" ; import * as https from "https" ; import { Base64 } from "js-base64" ; import * as os from "os" ; import * as path from "path" ; import { Transform , TransformCallback } from "stream" ; import { CancelRequest , DownloadFailed , FileHandler , SasUrlExpired , UserCancelledError , } from "./FileHandler" ; import { ProgressCallback , ProgressInfo , request , RequestOptions } from "./Request" ; import { AccessToken , Logger } from "@itwin/core-bentley" ; import { ArgumentCheck } from "../imodelhub/Errors" ; import { IModelHubClientLoggerCategory } from "../IModelHubClientLoggerCategories" ; import { AzCopy , InitEventArgs , ProgressEventArgs , StringEventArgs } from "./AzCopy" ; import { BlobDownloader , ConfigData , ProgressData } from "./BlobDownloader" ; const loggerCategory = IModelHubClientLoggerCategory . FileHandlers ; internal export class BufferedStream extends Transform { private _buffer ? : Buffer ; private _bufferPointer : number = 0 ; private _bufferSize : number ; public constructor ( bufferSize ) { super ( ) ; this . _bufferSize = bufferSize ; if ( ! Number . isInteger ( this . _bufferSize ) ) throw new TypeError ( `BufferSize must be integer.` ) ; } private allocNewBuffer ( ) { this . _buffer = Buffer . allocUnsafe ( this . _bufferSize ) ; this . _bufferPointer = 0 ; } param chunk param encoding param callback public override _transform ( chunk , encoding , callback ) { if ( encoding !== "buffer" && encoding !== "binary" ) throw new TypeError ( `Encoding '${ encoding }' is not supported.` ) ; if ( chunk . length >= this . _bufferSize ) { if ( ! this . _buffer ) { callback ( undefined , chunk ) ; return ; } this . _buffer = this . _buffer . slice ( 0 , this . _bufferPointer ) ; callback ( undefined , Buffer . concat ( [ this . _buffer , chunk ] ) ) ; this . _bufferPointer = 0 ; this . _buffer = undefined ; return ; } if ( ! this . _buffer ) { this . allocNewBuffer ( ) ; } if ( this . _bufferPointer + chunk . length <= this . _bufferSize ) { chunk . copy ( this . _buffer , this . _bufferPointer , 0 , chunk . length ) ; this . _bufferPointer += chunk . length ; if ( this . _bufferPointer === this . _bufferSize ) { callback ( undefined , this . _buffer ) ; this . allocNewBuffer ( ) ; } else { callback ( ) ; } } else { const chunkEndPosition = ( this . _bufferSize - this . _bufferPointer ) ; chunk . copy ( this . _buffer , this . _bufferPointer , 0 , chunkEndPosition ) ; this . _bufferPointer += chunkEndPosition ; callback ( undefined , this . _buffer ) ; this . allocNewBuffer ( ) ; chunk . copy ( this . _buffer , this . _bufferPointer , chunkEndPosition , chunk . length ) ; this . _bufferPointer += chunk . length - chunkEndPosition ; } } param callback public override _flush ( callback ) { if ( ! this . _buffer ) { callback ( ) ; return ; } callback ( undefined , this . _buffer . slice ( 0 , this . _bufferPointer ) ) ; this . _buffer = undefined ; this . _bufferPointer = 0 ; } } internal export class AzureFileHandler implements FileHandler { internal public agent ? : https . Agent ; private _threshold : number ; private _useDownloadBuffer : boolean | undefined ; private _config : ConfigData | undefined ; param useDownloadBuffer param threshold constructor ( useDownloadBuffer ? , threshold = 1024 * 1024 * 20 , config ? ) { this . _threshold = threshold ; this . _useDownloadBuffer = useDownloadBuffer ; this . _config = config ; } /** Check if using Azure File Share with UNC path. This is a temporary optimization for Design Review, until they move to using SSD disks. */ private useBufferedDownload ( downloadPath ) { if ( this . _useDownloadBuffer === undefined ) { return downloadPath . includes ( "file.core.windows.net" ) ; } else { return this . _useDownloadBuffer ; } } /** Create a directory, recursively setting up the path as necessary. */ private static makeDirectoryRecursive ( dirPath ) { if ( fs . existsSync ( dirPath ) ) return ; AzureFileHandler . makeDirectoryRecursive ( path . dirname ( dirPath ) ) ; fs . mkdirSync ( dirPath ) ; } private async transferFileUsingAzCopy ( source , target , progressCallback ? ) { Logger . logTrace ( loggerCategory , `Using AzCopy with version ${ AzCopy . getVersion ( ) } located at ${ AzCopy . execPath }` ) ; const azLogDir = path . join ( os . tmpdir ( ) , "bentley" , "log" , "azcopy" ) ; if ( ! fs . existsSync ( azLogDir ) ) { AzureFileHandler . makeDirectoryRecursive ( azLogDir ) ; } const azcopy = new AzCopy ( { logLocation : azLogDir } ) ; if ( progressCallback ) { const cb = ( args ) => { progressCallback ( { total : args . TotalBytesEnumerated , loaded : args . BytesOverWire , percent : args . BytesOverWire ? ( args . BytesOverWire / args . TotalBytesEnumerated ) * 100.0 : 0 } ) ; } ; azcopy . on ( "azprogress" , cb ) ; azcopy . on ( "azexit" , cb ) ; } azcopy . on ( "azerror" , ( args ) => { Logger . logError ( loggerCategory , `AzCopy reported error: '${ args . MessageContent }'` ) ; } ) ; azcopy . on ( "azinit" , ( args ) => { Logger . logInfo ( loggerCategory , `AzCopy started JobId: ${ args . JobID } and log file located at ${ args . LogFileLocation }` ) ; } ) ; azcopy . on ( "azruntimeerror" , ( args ) => { Logger . logInfo ( loggerCategory , `AzCopy runtime error: ${ args }` ) ; } ) ; const rc = await azcopy . copy ( source , target ) ; if ( rc !== 0 ) { throw new Error ( `AzCopy failed with return code: ${ rc }` ) ; } } private async downloadFileUsingHttps ( downloadUrl , downloadToPathname , _fileSize ? , progressCallback ? , cancelRequest ? ) { let lastProgressStat ; const onProgress = ( data ) => { if ( progressCallback ) progressCallback ( { percent : data . percentage , loaded : data . bytesDone , total : data . bytesTotal } ) ; lastProgressStat = data ; } ; await BlobDownloader . downloadFile ( downloadUrl , downloadToPathname , this . _config , onProgress , cancelRequest ) ; if ( ! fs . existsSync ( downloadToPathname ) ) { throw new Error ( "file not found" ) ; } Logger . logInfo ( loggerCategory , `BlobDownloader finished` , ( ) => { return { file : downloadToPathname , overallSpeed : BlobDownloader . formatRate ( lastProgressStat . downloadRateBytesPerSec ) , lastTwoSecSpeed : BlobDownloader . formatRate ( lastProgressStat . windowRateBytesPerSec ) , blobSize : BlobDownloader . formatBytes ( lastProgressStat . bytesTotal ) } ; } ) ; } param url private static getSafeUrlForLogging ( url ) { const safeToLogDownloadUrl = new URL ( url ) ; if ( safeToLogDownloadUrl . search && safeToLogDownloadUrl . search . length > 0 ) safeToLogDownloadUrl . search = "..." ; if ( safeToLogDownloadUrl . hash && safeToLogDownloadUrl . hash . length > 0 ) safeToLogDownloadUrl . hash = "..." ; return safeToLogDownloadUrl . toString ( ) ; } param download param futureSeconds public static isUrlExpired ( downloadUrl , futureSeconds ? ) { const sasUrl = new URL ( downloadUrl ) ; const se = sasUrl . searchParams . get ( "se" ) ; if ( se ) { const expiryUTC = new Date ( se ) ; const now = new Date ( ) ; const currentUTC = new Date ( now . toUTCString ( ) ) ; if ( futureSeconds ) { currentUTC . setSeconds ( futureSeconds + currentUTC . getSeconds ( ) ) ; } return expiryUTC <= currentUTC ; } return false ; } param fileSize private useAzCopyForFileTransfer ( fileSize ? ) { if ( fileSize ) { const minFileSize = process . env . IMJS_AZ_MIN_FILESIZE_THRESHOLD ? Number ( process . env . IMJS_AZ_MIN_FILESIZE_THRESHOLD ) : 500 * 1024 * 1024 ; if ( fileSize < minFileSize ) return false ; } return AzCopy . isAvailable ; } param downloadUrl param downloadToPathname param fileSize param progressCallback throws throws public async downloadFile ( _accessToken , downloadUrl , downloadToPathname , fileSize ? , progressCallback ? , cancelRequest ? ) { const safeToLogUrl = AzureFileHandler . getSafeUrlForLogging ( downloadUrl ) ; Logger . logInfo ( loggerCategory , `Downloading file from ${ safeToLogUrl }` ) ; ArgumentCheck . defined ( "downloadUrl" , downloadUrl ) ; ArgumentCheck . defined ( "downloadToPathname" , downloadToPathname ) ; if ( AzureFileHandler . isUrlExpired ( downloadUrl ) ) { Logger . logError ( loggerCategory , `Sas url has expired ${ safeToLogUrl }` ) ; throw new SasUrlExpired ( 403 , "Download URL has expired" ) ; } if ( fs . existsSync ( downloadToPathname ) ) fs . unlinkSync ( downloadToPathname ) ; AzureFileHandler . makeDirectoryRecursive ( path . dirname ( downloadToPathname ) ) ; try { if ( this . useAzCopyForFileTransfer ( fileSize ) ) { await this . transferFileUsingAzCopy ( downloadUrl , downloadToPathname , progressCallback ) ; } else { await this . downloadFileUsingHttps ( downloadUrl , downloadToPathname , fileSize , progressCallback , cancelRequest ) ; } } catch ( err ) { if ( fs . existsSync ( downloadToPathname ) ) fs . unlinkSync ( downloadToPathname ) ; if ( ! ( err instanceof UserCancelledError ) ) Logger . logWarning ( loggerCategory , `Error downloading file: ${ err }` ) ; throw err ; } if ( fileSize && fs . existsSync ( downloadToPathname ) ) { if ( fs . lstatSync ( downloadToPathname ) . size !== fileSize ) { fs . unlinkSync ( downloadToPathname ) ; Logger . logError ( loggerCategory , `Downloaded file is of incorrect size ${ safeToLogUrl }` ) ; throw new DownloadFailed ( 403 , "Download failed. Expected filesize does not match" ) ; } } Logger . logTrace ( loggerCategory , `Downloaded file from ${ safeToLogUrl }` ) ; } /** Get encoded block id from its number. */ private getBlockId ( blockId ) { return Base64 . encode ( blockId . toString ( 16 ) . padStart ( 5 , "0" ) ) ; } private async uploadChunk ( uploadUrlString , fileDescriptor , blockId , callback ? ) { const chunkSize = 4 * 1024 * 1024 ; let buffer = Buffer . alloc ( chunkSize ) ; const bytesRead = fs . readSync ( fileDescriptor , buffer , 0 , chunkSize , chunkSize * blockId ) ; buffer = buffer . slice ( 0 , bytesRead ) ; const options = { method : "PUT" , headers : { "x-ms-blob-type" : "BlockBlob" , "Content-Type" : "application/octet-stream" , "Content-Length" : buffer . length } , body : buffer , progressCallback : callback , agent : this . agent , timeout : { deadline : 60000 , response : 60000 } } ; const uploadUrl = `${ uploadUrlString }&comp=block&blockid=${ this . getBlockId ( blockId ) }` ; await request ( uploadUrl , options ) ; } param uploadUrl param uploadFromPathname param progressCallback throws throws public async uploadFile ( _accessToken , uploadUrlString , uploadFromPathname , progressCallback ? ) { const safeToLogUrl = AzureFileHandler . getSafeUrlForLogging ( uploadUrlString ) ; Logger . logTrace ( loggerCategory , `Uploading file to ${ safeToLogUrl }` ) ; ArgumentCheck . defined ( "uploadUrlString" , uploadUrlString ) ; ArgumentCheck . defined ( "uploadFromPathname" , uploadFromPathname ) ; const fileSize = this . getFileSize ( uploadFromPathname ) ; if ( this . useAzCopyForFileTransfer ( fileSize ) ) { await this . transferFileUsingAzCopy ( uploadFromPathname , uploadUrlString , progressCallback ) ; } else { await this . uploadFileUsingHttps ( uploadUrlString , uploadFromPathname , fileSize , progressCallback ) ; } } private async uploadFileUsingHttps ( uploadUrlString , uploadFromPathname , fileSize , progressCallback ? ) { const file = fs . openSync ( uploadFromPathname , "r" ) ; const chunkSize = 4 * 1024 * 1024 ; try { let blockList = '<?xml version=\"1.0\" encoding=\"utf-8\"?><BlockList>' ; let i = 0 ; const callback = ( progress ) => { const uploaded = i * chunkSize + progress . loaded ; if ( progressCallback ) progressCallback ( { loaded : uploaded , percent : uploaded / fileSize , total : fileSize } ) ; } ; for ( ; i * chunkSize < fileSize ; ++ i ) { await this . uploadChunk ( uploadUrlString , file , i , progressCallback ? callback : undefined ) ; blockList += `<Latest>${ this . getBlockId ( i ) }</Latest>` ; } blockList += "</BlockList>" ; const options = { method : "PUT" , headers : { "Content-Type" : "application/xml" , "Content-Length" : blockList . length } , body : blockList , agent : this . agent , timeout : { response : 5000 , deadline : 60000 } } ; const uploadUrl = `${ uploadUrlString }&comp=blocklist` ; await request ( uploadUrl , options ) ; } finally { fs . closeSync ( file ) ; } } param filePath returns public getFileSize ( filePath ) { return fs . statSync ( filePath ) . size ; } param filePath returns public isDirectory ( filePath ) { return fs . statSync ( filePath ) . isDirectory ( ) ; } param filePath returns public exists ( filePath ) { return fs . existsSync ( filePath ) ; } param filePath public unlink ( filePath ) { fs . unlinkSync ( filePath ) ; } param filePath returns public basename ( filePath ) { return path . basename ( filePath ) ; } param paths returns public join ( ... paths ) { return path . join ( ... paths ) ; } }