<commit_before>import {Socket} from "net";
import mkdirp = require('mkdirp');
import fs = require('fs');
import once = require('once');
import {FtpFile} from "../objects/FtpFile";
import {TransferSpeedAverage} from "../objects/TransfersSpeedAverage";

const logger = require('./Logger');
const FtpController = require('./FtpController');

class FtpDownloader {
    private _file: FtpFile;
    private _destinationDirectory: string;

    private _downloadDoneCallback;

    private _downloadSpeed = new TransferSpeedAverage();

    constructor(file: FtpFile, destinationDirectory: string) {
        this._file = file;
        this._destinationDirectory = destinationDirectory;
    }

    public start(callback) {
        var self = this;

        this._downloadDoneCallback = callback;
        
        const file = this._file;
        const localDirectory = this._destinationDirectory;

        file.destinationRoot = localDirectory;

        file.downloading = true;

        logger.info("mkdirp", localDirectory);

        //Create the full path. jsftp will not error if the directory doesn't exist.
        mkdirp.sync(localDirectory, function (err) {
            if (err) logger.error(err);
            else logger.info('dir created')
        });

        let localPath = FtpFile.appendSlash(localDirectory) + file.name;

        logger.info("Downloading", file.fullPath);

        var ftp = FtpController.ftpForDownloading();

        let fd;
        let socket;

        let downloadDone = function(err) {
            if (err) {
                logger.error("Error downloading file\r\n", err);
                self._downloadDoneCallback(err, file);

                //Make sure no more data can come in
                if (socket) {
                    socket.destroy();
                }
            }

            FtpController.doneWithFtpObj(ftp);

            if (fd) {
                try {
                    fs.closeSync(fd);
                } catch(exception) {
                    logger.error('Error closing file descriptor.?', exception);
                    err = exception;
                }
            }

            if (!err) {
                logger.info("File downloaded succesfully", localPath);
            }

            self._file.downloading = false;

            self._downloadDoneCallback(err, file);
        };

        //There's a bunch of error listeners and one success listener tied to this, only call once.
        downloadDone = once(downloadDone);


        ftp.on('error', downloadDone)
            .on('timeout', function() {
                downloadDone('timeout');
            });



        // Retrieve the file using async streams
        ftp.getGetSocket(file.fullPath, function(err: Error, sock: Socket) {
            socket = sock;
            
            if (err) {
                logger.error('Error calling ftp.getGetSocket');
                downloadDone(err);
                return;
            }

            fd = fs.openSync(localPath, "w+");

            // `sock` is a stream. attach events to it.
            sock.on("data", function(p) {
                fs.writeSync(fd, p, 0, p.length, null);

                //Or should this be fs.bytesWritten?
                file.transferred = sock.bytesRead;

                self._downloadSpeed.dataReceived(p.length);

                file.downloadRate = self._downloadSpeed.average();
            });
            sock.on("close", function(err) {
                if (err) {
                    downloadDone(err);
                    return;
                }

                downloadDone(null)
            });

            // The sock stream is paused. Call resume() on it to start reading.
            sock.resume();
        });
    }
}

module.exports = FtpDownloader;<commit_msg>Downloads are now named *.tmp to prevent other apps from grabbing them before they are ready. This closes #27.<commit_after>import {Socket} from "net";
import mkdirp = require('mkdirp');
import fs = require('fs');
import once = require('once');
import {FtpFile} from "../objects/FtpFile";
import {TransferSpeedAverage} from "../objects/TransfersSpeedAverage";

const logger = require('./Logger');
const FtpController = require('./FtpController');

class FtpDownloader {
    private _file: FtpFile;
    private _destinationDirectory: string;

    private _downloadDoneCallback;

    private _downloadSpeed = new TransferSpeedAverage();

    constructor(file: FtpFile, destinationDirectory: string) {
        this._file = file;
        this._destinationDirectory = destinationDirectory;
    }

    public start(callback) {
        var self = this;

        this._downloadDoneCallback = callback;
        
        const file = this._file;
        const localDirectory = this._destinationDirectory;

        file.destinationRoot = localDirectory;

        file.downloading = true;

        logger.info("mkdirp", localDirectory);

        //Create the full path. jsftp will not error if the directory doesn't exist.
        mkdirp.sync(localDirectory, function (err) {
            if (err) logger.error(err);
            else logger.info('dir created')
        });

        let localPath = FtpFile.appendSlash(localDirectory) + file.name;
        let tempPath = localPath + ".tmp";

        logger.info("Downloading", file.fullPath);

        var ftp = FtpController.ftpForDownloading();

        let fd;
        let socket;

        let downloadDone = function(err) {
            if (err) {
                logger.error("Error downloading file\r\n", err);
                self._downloadDoneCallback(err, file);

                //Make sure no more data can come in
                if (socket) {
                    socket.destroy();
                }
            }

            FtpController.doneWithFtpObj(ftp);

            if (fd) {
                try {
                    fs.closeSync(fd);
                } catch(exception) {
                    logger.error('Error closing file descriptor.?', exception);
                    err = exception;
                }
            }

            try {
                fs.renameSync(tempPath, localPath);
            } catch (exception) {
                logger.error('Error renaming temp file', tempPath, exception);
                err = exception;
            }

            if (!err) {
                logger.info("File downloaded succesfully", localPath);
            }

            self._file.downloading = false;

            self._downloadDoneCallback(err, file);
        };

        //There's a bunch of error listeners and one success listener tied to this, only call once.
        downloadDone = once(downloadDone);


        ftp.on('error', downloadDone)
            .on('timeout', function() {
                downloadDone('timeout');
            });



        // Retrieve the file using async streams
        ftp.getGetSocket(file.fullPath, function(err: Error, sock: Socket) {
            socket = sock;
            
            if (err) {
                logger.error('Error calling ftp.getGetSocket');
                downloadDone(err);
                return;
            }

            fd = fs.openSync(tempPath, "w+");

            // `sock` is a stream. attach events to it.
            sock.on("data", function(p) {
                fs.writeSync(fd, p, 0, p.length, null);

                //Or should this be fs.bytesWritten?
                file.transferred = sock.bytesRead;

                self._downloadSpeed.dataReceived(p.length);

                file.downloadRate = self._downloadSpeed.average();
            });
            sock.on("close", function(err) {
                if (err) {
                    downloadDone(err);
                    return;
                }

                downloadDone(null)
            });

            // The sock stream is paused. Call resume() on it to start reading.
            sock.resume();
        });
    }
}

module.exports = FtpDownloader;
{'fraction_non_alphanumeric': 0.1036999748301032, 'fraction_numerical': 0.0005033979360684622, 'mean_word_length': 1.6632037533512065, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 49, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7133987', 'n_tokens_mistral': 2092, 'n_tokens_neox': 1927, 'n_words': 516}