<commit_before>'use strict'

/* @flow */

import FS from 'fs'
import URL from 'url'
import Path from 'path'
import {Emitter} from 'sb-event-kit'
import {promisedRequest, getRange} from './helpers'
import type {Disposable} from 'sb-event-kit'
import type {RangePool, PoolWorker} from 'range-pool'
import type {Readable} from 'stream'

export class Connection {
  emitter: Emitter;
  url: string;
  pool: RangePool;
  worker: PoolWorker;
  fileInfo: {
    size: number
  };
  response: Readable;
  started: boolean;

  constructor(url: string, pool: RangePool) {
    this.url = url
    this.pool = pool
    this.started = false
    this.emitter = new Emitter()
  }
  async activate(): Promise<Connection> {
    if (this.started) {
      return ;
    }

    this.started = true
    this.fileInfo = { size: 0 }
    this.worker = this.pool.createWorker()
    this.response = await promisedRequest({
      url: this.url,
      headers: {
        'User-Agent': 'sb-downloader for Node.js',
        'Range': getRange(this.worker)
      }
    })
    this.fileInfo.size = parseInt(this.response.headers['content-length']) || 0
    return this
  }
  start(fd: number) {
    this.response.on('data', chunk => {
      const chunkLength = chunk.length
      const remaining = this.worker.getRemaining()
      const shouldClose = remaining <= chunkLength

      if (chunkLength > remaining) {
        chunk = chunk.slice(0, remaining)
      }

      FS.write(fd, chunk, 0, chunk.length, this.worker.getCurrentIndex(), error => {
        if (error) {
          this.emitter.emit('did-error', error)
        }
      })
      this.worker.advance(chunk.length)
      this.emitter.emit('did-progress', this.worker.getCompletionPercentage())
      if (shouldClose) {
        this.emitter.emit('did-close')
        this.dispose()
      }
    })
    this.response.resume()
  }
  getResponse(): Object {
    return this.response
  }
  getFileSize(): number {
    return this.fileInfo.size
  }
  getFileName(): string {
    const parsed = URL.parse(this.url, true)
    return Path.basename(parsed.pathname || '')
  }
  onDidClose(callback: Function): Disposable {
    return this.emitter.on('did-close', callback)
  }
  onDidProgress(callback: Function): Disposable {
    return this.emitter.on('did-progress', callback)
  }
  onDidError(callback: Function): Disposable {
    return this.emitter.on('did-error', callback)
  }
  dispose() {
    this.emitter.dispose()
    this.worker.dispose()
    if (this.response) {
      // $FlowIgnore: It's an internal function for readable streams, but exposed by request API
      this.response.destroy()
    }
    this.started = false
    this.emitter = new Emitter()
  }
}
<commit_msg>:new: Add supportsResume properly of connection<commit_after>'use strict'

/* @flow */

import FS from 'fs'
import URL from 'url'
import Path from 'path'
import {Emitter} from 'sb-event-kit'
import {promisedRequest, getRange} from './helpers'
import type {Disposable} from 'sb-event-kit'
import type {RangePool, PoolWorker} from 'range-pool'

export class Connection {
  emitter: Emitter;
  url: string;
  pool: RangePool;
  worker: PoolWorker;
  fileInfo: {
    size: number
  };
  response: Object;
  started: boolean;
  supportsResume: boolean;

  constructor(url: string, pool: RangePool) {
    this.emitter = new Emitter()
    this.url = url
    this.pool = pool
    this.started = false
    this.supportsResume = true
  }
  async activate(): Promise<Connection> {
    if (this.started) {
      return ;
    }
    const range = getRange(this.worker)

    this.started = true
    this.fileInfo = { size: 0 }
    this.worker = this.pool.createWorker()
    this.response = await promisedRequest({
      url: this.url,
      headers: {
        'User-Agent': 'sb-downloader for Node.js',
        'Range': range
      }
    })
    this.fileInfo.size = parseInt(this.response.headers['content-length']) || 0
    this.supportsResume = range === null || this.response.statusCode === 206
    return this
  }
  start(fd: number) {
    this.response.on('data', chunk => {
      const chunkLength = chunk.length
      const remaining = this.worker.getRemaining()
      const shouldClose = remaining <= chunkLength

      if (chunkLength > remaining) {
        chunk = chunk.slice(0, remaining)
      }

      FS.write(fd, chunk, 0, chunk.length, this.worker.getCurrentIndex(), error => {
        if (error) {
          this.emitter.emit('did-error', error)
        }
      })
      this.worker.advance(chunk.length)
      this.emitter.emit('did-progress', this.worker.getCompletionPercentage())
      if (shouldClose) {
        this.emitter.emit('did-close')
        this.dispose()
      }
    })
    this.response.resume()
  }
  getResponse(): Object {
    return this.response
  }
  getFileSize(): number {
    return this.fileInfo.size
  }
  getFileName(): string {
    const parsed = URL.parse(this.url, true)
    return Path.basename(parsed.pathname || '')
  }
  onDidClose(callback: Function): Disposable {
    return this.emitter.on('did-close', callback)
  }
  onDidProgress(callback: Function): Disposable {
    return this.emitter.on('did-progress', callback)
  }
  onDidError(callback: Function): Disposable {
    return this.emitter.on('did-error', callback)
  }
  dispose() {
    this.emitter.dispose()
    this.worker.dispose()
    if (this.response) {
      this.response.destroy()
    }
    this.started = false
    this.emitter = new Emitter()
  }
}

{'fraction_non_alphanumeric': 0.13166086797289875, 'fraction_numerical': 0.0020142830983336385, 'mean_word_length': 2.849189570119803, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1410199', 'n_tokens_mistral': 1742, 'n_tokens_neox': 1689, 'n_words': 393}