import * as stream from 'stream' ; import { DriverConstructor , DriverStatics } from './driverModel' ; import S3Driver from './drivers/S3Driver' ; import AzDriver from './drivers/AzDriver' ; import GcDriver from './drivers/GcDriver' ; import DiskDriver from './drivers/diskDriver' ; import { promisify } from 'util' ; import * as winston from 'winston' ; import { DriverName } from './config' ; import nanoid = require ( 'nanoid/generate' ) ; const alphabet = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz' ; /**
 * Generates a random 10 char string using uppercase & lowercase alpha numeric alphabet.
 */ export function generateUniqueID ( ) { const id = nanoid ( alphabet , 10 ) ; return id ; } export const pipelineAsync = promisify ( stream . pipeline ) ; export const logger = winston . createLogger ( ) ; export function getDriverClass ( driver ) { if ( driver === 'aws' ) { return S3Driver ; } else if ( driver === 'azure' ) { return AzDriver ; } else if ( driver === 'disk' ) { return DiskDriver ; } else if ( driver === 'google-cloud' ) { return GcDriver ; } else { throw new Error ( `Failed to load driver: driver was set to ${ driver }` ) ; } } export function megabytesToBytes ( megabytes ) { return megabytes * 1024 * 1024 ; } export function bytesToMegabytes ( bytes , decimals = 2 ) { return Number . parseFloat ( ( bytes / ( 1024 / 1024 ) ) . toFixed ( decimals ) ) ; } export function dateToUnixTimeSeconds ( date ) { return Math . round ( date . getTime ( ) / 1000 ) ; } class MemoryStream extends stream . Writable { buffers : Buffer [ ] ; constructor ( opts ? ) { super ( opts ) ; this . buffers = [ ] ; } _write ( chunk , encoding , callback ) { this . buffers . push ( Buffer . from ( chunk , encoding ) ) ; callback ( null ) ; } getData ( ) { if ( this . buffers . length === 1 ) { return this . buffers [ 0 ] ; } return Buffer . concat ( this . buffers ) ; } } export async function readStream ( stream ) { const memStream = new MemoryStream ( ) ; await pipelineAsync ( stream , memStream ) ; return memStream . getData ( ) ; } export function timeout ( milliseconds ) { return new Promise ( resolve => { setTimeout ( ( ) => { resolve ( ) ; } , milliseconds ) ; } ) ; } export async function tryFor < T > ( fn , retryIntervalMS , totalTimeMS ) { const startTime = process . hrtime ( ) ; const getElapsedMilliseconds = ( ) => { const [ elapsedSeconds , elapsedNS ] = process . hrtime ( startTime ) ; return ( elapsedSeconds * 1000 ) + parseFloat ( ( elapsedNS * 1e-6 ) . toPrecision ( 4 ) ) ; } ; const getRemainingMS = ( ) => { const elapsedMs = getElapsedMilliseconds ( ) ; return totalTimeMS - elapsedMs ; } ; let lastError ; do { try { return await fn ( ) ; } catch ( error ) { lastError = error ; } const result = await Promise . race ( [ timeout ( retryIntervalMS ) , timeout ( getRemainingMS ( ) ) . then ( ( ) => ( { timeout : true } ) ) ] ) ; if ( result && result . timeout && lastError === undefined ) { lastError = new Error ( 'Timed out' ) ; } } while ( getRemainingMS ( ) > 0 ) ; throw lastError ; } export interface StreamProgressCallback { param totalBytes param chunkBytes ( totalBytes , chunkBytes ) : void ; } export interface MonitorStreamResult { monitoredStream : stream . Readable ; pipelinePromise : Promise < void > ; } export function monitorStreamProgress ( inputStream , progressCallback ) { let monitoredContentSize = 0 ; const monitorStream = new stream . PassThrough ( { transform : ( chunk , _encoding , callback ) => { monitoredContentSize += chunk . length ; try { progressCallback ( monitoredContentSize , chunk . length ) ; callback ( null , chunk ) ; } catch ( error ) { callback ( error ) ; } } } ) ; const monitorPipeline = pipelineAsync ( inputStream , monitorStream ) ; const result = { monitoredStream : monitorStream , pipelinePromise : monitorPipeline } ; return result ; } export class AsyncMutexScope { private readonly _opened = new Set < string > ( ) ; public get openedCount ( ) { return this . _opened . size ; } param id param spawnOwner returns public tryAcquire ( id , spawnOwner ) { if ( this . _opened . has ( id ) ) { return false ; } this . _opened . add ( id ) ; try { const owner = spawnOwner ( ) ; owner . finally ( ( ) => { this . _opened . delete ( id ) ; } ) ; } catch ( error ) { this . _opened . delete ( id ) ; throw error ; } return true ; } }