v2.0.0

src/common/utils/download/Downloader.ts

@@ -0,0 +1,358 @@
+import fs from 'fs'
+import path from 'path'
+import { EventEmitter } from 'events'
+import { performance } from 'perf_hooks'
+import { STATUS } from './util'
+import http from 'http'
+import { request, Options as RequestOptions } from './request'
+
+export interface Options {
+  forceResume: boolean
+  requestOptions: RequestOptions
+}
+
+const defaultChunkInfo = {
+  path: '',
+  startByte: '0',
+  endByte: '',
+}
+
+const defaultRequestOptions: Options['requestOptions'] = {
+  method: 'get',
+  headers: {},
+}
+const defaultOptions: Options = {
+  forceResume: true,
+  requestOptions: { ...defaultRequestOptions },
+}
+
+class Task extends EventEmitter {
+  resumeLastChunk: Buffer | null
+  downloadUrl: string
+  chunkInfo: { path: string, startByte: string, endByte: string }
+  status: typeof STATUS[keyof typeof STATUS]
+  options: Options
+  requestOptions: Options['requestOptions']
+  ws: fs.WriteStream | null = null
+  progress = { total: 0, downloaded: 0, speed: 0, progress: 0 }
+  statsEstimate = { time: 0, bytes: 0, prevBytes: 0 }
+  requestInstance: http.ClientRequest | null = null
+
+
+  constructor(url: string, savePath: string, filename: string, options: Partial<Options> = {}) {
+    super()
+
+    this.resumeLastChunk = null
+    this.downloadUrl = url
+    this.chunkInfo = Object.assign({}, defaultChunkInfo, {
+      path: path.join(savePath, filename),
+      startByte: '0',
+    })
+    // if (!this.chunkInfo.endByte) this.chunkInfo.endByte = ''
+
+    this.options = Object.assign({}, defaultOptions, options)
+    this.requestOptions = Object.assign({}, defaultRequestOptions, this.options.requestOptions || {})
+    this.requestOptions.headers = this.requestOptions.headers ? { ...this.requestOptions.headers } : {}
+
+    this.status = STATUS.idle
+  }
+
+  async __init() {
+    const { path, startByte, endByte } = this.chunkInfo
+    this.progress.downloaded = 0
+    this.progress.progress = 0
+    this.progress.speed = 0
+    if (startByte) this.requestOptions.headers!.range = `bytes=${startByte}-${endByte}`
+
+    if (!path) return
+    return new Promise<void>((resolve, reject) => {
+      fs.stat(path, (errStat, stats) => {
+        if (errStat) {
+          // console.log(errStat.code)
+          if (errStat.code !== 'ENOENT') {
+            this.__handleError(errStat)
+            reject(errStat)
+            return
+          }
+        } else if (stats.size >= 10) {
+          fs.open(path, 'r', (errOpen, fd) => {
+            if (errOpen) {
+              this.__handleError(errOpen)
+              reject(errOpen)
+              return
+            }
+            fs.read(fd, Buffer.alloc(10), 0, 10, stats.size - 10, (errRead, bytesRead, buffer) => {
+              if (errRead) {
+                this.__handleError(errRead)
+                reject(errRead)
+                return
+              }
+              fs.close(fd, errClose => {
+                if (errClose) {
+                  this.__handleError(errClose)
+                  reject(errClose)
+                  return
+                }
+
+                // resume download
+                // console.log(buffer)
+                this.resumeLastChunk = buffer
+                this.progress.downloaded = stats.size
+                this.requestOptions.headers!.range = `bytes=${stats.size - 10}-${endByte || ''}`
+                resolve()
+              })
+            })
+          })
+          return
+        }
+        resolve()
+      })
+    })
+  }
+
+  __httpFetch(url: string, options: Options['requestOptions']) {
+    // console.log(options)
+    this.requestInstance = request(url, options)
+      .on('response', response => {
+        if (response.statusCode !== 200 && response.statusCode !== 206) {
+          if (response.statusCode == 416) {
+            fs.unlink(this.chunkInfo.path, (err) => {
+              this.__handleError(new Error(response.statusMessage))
+              this.chunkInfo.startByte = '0'
+              this.resumeLastChunk = null
+              this.progress.downloaded = 0
+              if (err) this.__handleError(err)
+            })
+            return
+          }
+          this.status = STATUS.failed
+          this.emit('fail', response)
+          this.__closeRequest()
+          void this.__closeWriteStream()
+          return
+        }
+        this.emit('response', response)
+        try {
+          this.__initDownload(response)
+        } catch (error: any) {
+          return this.__handleError(error)
+        }
+        this.status = STATUS.running
+        response
+          .on('data', this.__handleWriteData.bind(this))
+          .on('error', err => this.__handleError(err))
+          .on('end', () => {
+            if (response.complete) {
+              this.__handleComplete()
+            } else {
+              // this.__handleError(new Error('The connection was terminated while the message was still being sent'))
+              void this.stop()
+            }
+          })
+      })
+      .on('error', err => this.__handleError(err))
+      .on('close', () => {
+        void this.__closeWriteStream()
+      })
+      .end()
+  }
+
+  __initDownload(response: http.IncomingMessage) {
+    this.progress.total = response.headers['content-length'] ? parseInt(response.headers['content-length']) : 0
+    if (!this.progress.total) return this.__handleError(new Error('Content length is 0'))
+    let options: any = {}
+    let isResumable = this.options.forceResume ||
+      response.headers['accept-ranges'] !== 'none' ||
+      (typeof response.headers['accept-ranges'] == 'string' &&
+        parseInt(response.headers['accept-ranges'].replace(/^bytes=(\d+)/, '$1')) > 0)
+
+    if (isResumable) {
+      options.flags = 'a'
+      if (this.progress.downloaded) this.progress.total -= 10
+    } else {
+      if (this.chunkInfo.startByte != '0') return this.__handleError(new Error('The resource cannot be resumed download.'))
+    }
+    this.progress.total += this.progress.downloaded
+    this.statsEstimate.prevBytes = this.progress.downloaded
+    if (!this.chunkInfo.path) return this.__handleError(new Error('Chunk save Path is not set.'))
+    this.ws = fs.createWriteStream(this.chunkInfo.path, options)
+
+    this.ws.on('finish', () => {
+      void this.__closeWriteStream()
+    })
+    this.ws.on('error', err => {
+      fs.unlink(this.chunkInfo.path, (unlinkErr: any) => {
+        this.__handleError(err)
+        this.chunkInfo.startByte = '0'
+        this.resumeLastChunk = null
+        this.progress.downloaded = 0
+        if (unlinkErr && unlinkErr.code !== 'ENOENT') this.__handleError(unlinkErr)
+      })
+    })
+  }
+
+  __handleComplete() {
+    if (this.status == STATUS.error) return
+    void this.__closeWriteStream().then(() => {
+      if (this.progress.downloaded == this.progress.total) {
+        this.status = STATUS.completed
+        this.emit('completed')
+      } else {
+        this.status = STATUS.stopped
+        this.emit('stop')
+      }
+    })
+    // console.log('end')
+  }
+
+  __handleError(error: Error) {
+    if (this.status == STATUS.error) return
+    this.status = STATUS.error
+    this.__closeRequest()
+    void this.__closeWriteStream()
+    if (error.message == 'aborted') return
+    this.emit('error', error)
+  }
+
+  async __closeWriteStream() {
+    return new Promise<void>((resolve, reject) => {
+      if (!this.ws) return resolve()
+      // console.log('close write stream')
+      this.ws.close(err => {
+        if (err) {
+          this.status = STATUS.error
+          this.emit('error', err)
+          reject(err)
+          return
+        }
+        this.ws = null
+        resolve()
+      })
+    })
+  }
+
+  __closeRequest() {
+    if (!this.requestInstance || this.requestInstance.destroyed) return
+    // console.log('close request')
+    this.requestInstance.destroy()
+    this.requestInstance = null
+  }
+
+  __handleWriteData(chunk: Buffer) {
+    if (this.resumeLastChunk) {
+      const result = this.__handleDiffChunk(chunk)
+      if (result) chunk = result
+      else {
+        this.__handleStop().finally(() => {
+          // this.__handleError(new Error('Resume failed, response chunk does not match.'))
+          // Resume failed, response chunk does not match, remove file and restart download
+          console.log('Resume failed, response chunk does not match.')
+          fs.unlink(this.chunkInfo.path, (unlinkErr: any) => {
+            // this.__handleError(err)
+            this.chunkInfo.startByte = '0'
+            this.resumeLastChunk = null
+            if (unlinkErr && unlinkErr.code !== 'ENOENT') return this.__handleError(unlinkErr)
+            void this.start()
+          })
+        })
+        return
+      }
+    }
+    // console.log('data', chunk)
+    if (this.status == STATUS.stopped || this.ws == null) return console.log('cancel write')
+    this.__calculateProgress(chunk.length)
+    this.ws.write(chunk, err => {
+      if (!err) return
+      console.log(err)
+      this.__handleError(err)
+      void this.stop()
+    })
+  }
+
+  __handleDiffChunk(chunk: Buffer): Buffer | null {
+    // console.log('diff', chunk)
+    let resumeLastChunkLen = this.resumeLastChunk!.length
+    let chunkLen = chunk.length
+    let isOk
+    if (chunkLen >= resumeLastChunkLen) {
+      isOk = chunk.slice(0, resumeLastChunkLen).toString('hex') === this.resumeLastChunk!.toString('hex')
+      if (!isOk) return null
+
+      this.resumeLastChunk = null
+      return chunk.slice(resumeLastChunkLen)
+    } else {
+      isOk = chunk.slice(0, chunkLen).toString('hex') === this.resumeLastChunk!.slice(0, chunkLen).toString('hex')
+      if (!isOk) return null
+      this.resumeLastChunk = this.resumeLastChunk!.slice(chunkLen)
+      return chunk.slice(chunkLen)
+    }
+  }
+
+  async __handleStop() {
+    return new Promise<void>((resolve, reject) => {
+      this.__closeRequest()
+      if (this.ws) {
+        this.ws.close(err => {
+          if (err) {
+            reject(err)
+            this.emit('error', err)
+            return
+          }
+          this.ws = null
+          resolve()
+        })
+      } else {
+        resolve()
+      }
+    })
+  }
+
+  __calculateProgress(receivedBytes: number) {
+    const currentTime = performance.now()
+    const elaspsedTime = currentTime - this.statsEstimate.time
+
+    const progress = this.progress
+    progress.downloaded += receivedBytes
+    progress.progress = progress.total ? (progress.downloaded / progress.total) * 100 : -1
+
+
+    // emit the progress every second or if finished
+    if (progress.downloaded === progress.total || elaspsedTime > 1000) {
+      this.statsEstimate.time = currentTime
+      this.statsEstimate.bytes = progress.downloaded - this.statsEstimate.prevBytes
+      this.statsEstimate.prevBytes = progress.downloaded
+      this.emit('progress', {
+        total: progress.total,
+        downloaded: progress.downloaded,
+        progress: progress.progress,
+        speed: this.statsEstimate.bytes,
+      })
+    }
+  }
+
+  async start() {
+    this.status = STATUS.init
+    await this.__init()
+    if (this.status !== STATUS.init) return
+    this.status = STATUS.running
+    this.__httpFetch(this.downloadUrl, this.requestOptions)
+    this.emit('start')
+  }
+
+  async stop() {
+    if (this.status == STATUS.stopped || this.status == STATUS.completed) return
+    this.status = STATUS.stopped
+    await this.__handleStop()
+    this.emit('stop')
+  }
+
+  refreshUrl(url: string) {
+    this.downloadUrl = url
+  }
+
+  updateSaveInfo(filePath: string, fileName: string) {
+    this.chunkInfo.path = path.join(filePath, fileName)
+  }
+}
+
+export default Task

src/common/utils/download/index.ts

@@ -0,0 +1,100 @@
+import Downloader, { Options as DownloaderOptions } from './Downloader'
+import { getRequestAgent } from './util'
+import { sizeFormate } from '@common/utils'
+import http from 'http'
+
+// these are the default options
+// const options = {
+//   method: 'GET', // Request Method Verb
+//   // Custom HTTP Header ex: Authorization, User-Agent
+//   headers: {},
+//   fileName: '', // Custom filename when saved
+//   override: false, // if true it will override the file, otherwise will append '(number)' to the end of file
+//   forceResume: false, // If the server does not return the "accept-ranges" header, can be force if it does support it
+//   // httpRequestOptions: {}, // Override the http request options
+//   // httpsRequestOptions: {}, // Override the https request options, ex: to add SSL Certs
+// }
+
+export interface Options {
+  url: string
+  path: string
+  fileName: string
+  method?: DownloaderOptions['requestOptions']['method']
+  headers?: DownloaderOptions['requestOptions']['headers']
+  forceResume?: boolean
+  proxy?: { host: string, port: number }
+  onCompleted?: () => void
+  onError?: (error: Error) => void
+  onFail?: (response: http.IncomingMessage) => void
+  onStart?: () => void
+  onStop?: () => void
+  onProgress?: (progress: LX.Download.ProgressInfo) => void
+}
+const noop = () => {}
+export const createDownload = ({
+  url,
+  path,
+  fileName,
+  method = 'get',
+  forceResume,
+  proxy,
+  // resumeTime = 5000,
+  onCompleted = noop,
+  onError = noop,
+  onFail = noop,
+  onStart = noop,
+  onStop = noop,
+  onProgress = noop,
+}: Options) => {
+  const dl = new Downloader(url, path, fileName, {
+    requestOptions: {
+      method,
+      headers: {
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36',
+      },
+      agent: getRequestAgent(url, proxy),
+      timeout: 60 * 1000,
+    },
+
+    forceResume,
+  })
+
+  dl.on('completed', () => {
+    onCompleted()
+  }).on('error', (err: any) => {
+    if (err.message === 'socket hang up') return
+    onError(err)
+  }).on('start', () => {
+    onStart()
+    // pauseResumeTimer(dl, resumeTime)
+  }).on('progress', (stats) => {
+    const speed = sizeFormate(stats.speed)
+    onProgress({
+      progress: parseInt(stats.progress.toFixed(2)),
+      speed,
+      downloaded: stats.downloaded,
+      total: stats.total,
+    })
+    // if (debugDownload) {
+    //   const downloaded = sizeFormate(stats.downloaded)
+    //   const total = sizeFormate(stats.total)
+    //   console.log(`${speed}/s - ${progress}% [${downloaded}/${total}]`)
+    // }
+  }).on('stop', () => {
+    onStop()
+    // debugDownload && console.log('paused')
+  }).on('fail', resp => {
+    onFail(resp)
+    // debugDownload && console.log('fail')
+  })
+
+  // debugDownload && console.log('Downloading: ', url)
+
+  dl.start().catch(err => {
+    onError(err)
+  })
+
+  return dl
+}
+
+export type DownloaderType = Downloader

src/common/utils/download/request.ts

@@ -0,0 +1,73 @@
+import { URL } from 'url'
+import http from 'http'
+import https from 'https'
+
+export interface Options {
+  method: 'get' | 'head' | 'delete' | 'patch' | 'post' | 'put'
+  params?: Record<string, string>
+  // body?: Record<string, string>
+  headers?: Record<string, string>
+  timeout?: number
+  agent?: http.Agent
+}
+
+const defaultOptions: Options = {
+  method: 'get',
+}
+
+type HttpCallback = (res: http.IncomingMessage) => void
+
+const sendRequest = (url: string, options: Options, callback?: HttpCallback) => {
+  const urlParse = new URL(url)
+  const httpOptions: http.RequestOptions | https.RequestOptions = {
+    host: urlParse.hostname,
+    port: urlParse.port,
+    path: urlParse.pathname + urlParse.search,
+    method: options.method,
+  }
+
+  if (options.params) {
+    (httpOptions.path as string) += `${urlParse.search ? '&' : '?'}${Object.entries(options.params)
+      .map(([key, value]) => `${key}=${encodeURIComponent(value)}`)
+      .join('&')}`
+  }
+
+  if (options.headers) httpOptions.headers = { ...options.headers }
+
+  if (options.agent) httpOptions.agent = options.agent
+
+  return urlParse.protocol == 'https:'
+    ? https.request(httpOptions, callback)
+    : http.request(httpOptions, callback)
+}
+
+const applyTimeout = (request: http.ClientRequest, time: number) => {
+  let timeout: NodeJS.Timeout | null = setTimeout(() => {
+    timeout = null
+    if (request.destroyed) return
+    request.destroy(new Error('Request timeout'))
+  }, time)
+  request.on('response', () => {
+    if (!timeout) return
+    clearTimeout(timeout)
+    timeout = null
+  })
+}
+
+// const isRequireRedirect = (response: http.IncomingMessage) => {
+//   return response.statusCode &&
+//     response.statusCode > 300 &&
+//     response.statusCode < 400 &&
+//     Object.hasOwn(response.headers, 'location') &&
+//     response.headers.location
+// }
+
+// export function request(url: string, callback: HttpCallback)
+// export function request(url: string, options: Partial<Options>, callback: HttpCallback)
+export function request(url: string, _options: Partial<Options>, callback?: HttpCallback) {
+  let options: Options = { ...defaultOptions, ..._options }
+  const request = sendRequest(url, options, callback)
+  if (options.timeout) applyTimeout(request, options.timeout)
+  return request
+}
+

src/common/utils/download/util.ts

@@ -0,0 +1,26 @@
+import { httpOverHttp, httpsOverHttp } from 'tunnel'
+
+export const STATUS = {
+  idle: 'IDLE',
+  init: 'INIT',
+  running: 'RUNNING',
+  paused: 'PAUSED',
+  stopped: 'STOPPED',
+  completed: 'COMPLETED',
+  error: 'ERROR',
+  failed: 'FAILED',
+} as const
+
+const httpsRxp = /^https:/
+export const getRequestAgent = (url: string, proxy?: { host: string, port: number }) => {
+  let options
+  if (proxy) {
+    options = {
+      proxy: {
+        host: proxy.host,
+        port: proxy.port,
+      },
+    }
+  }
+  return options ? (httpsRxp.test(url) ? httpsOverHttp : httpOverHttp)(options) : undefined
+}