Can't get the resulting file

[Umisyus]

I wrote this function to compress files

async function createAsyncZip(data: KeyValueStoreRecord<Buffer>[]): Promise<Uint8Array> {

    return new Promise(async (resolve, reject) => {
        let res = null
        let handler: AsyncFlateStreamHandler = function (err, dat, fin) {
            err ? reject(err) : null

            if (fin) {
                log.info("Finished")
                resolve(dat)
            }
        }

        // Create a new zip container
        let zipContainer = new Zip(handler)

        for (let index = 0; index < data.length; index++) {
            const element = data[index];
            // Create a new file stream for the file to compress
            let fileToCompress = new ZipDeflate(element.key)
            // Add the file `stream` to the zip container
            zipContainer.add(fileToCompress)

            // Is this the last file?
            let fin = index == data.length - 1
            if (fin) {
                log.info("LAST ONE!")
            }
            // Push the file to the file stream
            fileToCompress.push(element.value, fin)

        }

        zipContainer.end()
    })
}

Whenever I run this, the files get added, the callback gets called twice then never again. When it is called, the final parameter is always false. I looked at the example and found this:

// from: (just above of) https://github.com/101arrowz/fflate#bundle-size-estimates
// Streaming ZIP archives can accept asynchronous streams. This automatically
// uses multicore compression.
const zip = new Zip();
zip.ondata = (err, chunk, final) => { ... };
// The JSON and BMP are compressed in parallel
const exampleFile = new AsyncZipDeflate('example.json');
zip.add(exampleFile);
exampleFile.push(JSON.stringify({ large: 'object' }), true);
const exampleFile2 = new AsyncZipDeflate('example2.bmp', { level: 9 });
zip.add(exampleFile2);
exampleFile2.push(ec2a);
exampleFile2.push(ec2b);
exampleFile2.push(ec2c);
...
exampleFile2.push(ec2Final, true);
zip.end();

[101arrowz]

The fin boolean should always be true: that flag signifies the last buffer per file, not the last buffer for the entire ZIP. However, since you already have a bunch of buffers, using zipSync or zip is better:

import { zip } from 'fflate';

function createAsyncZip(data: KeyValueStoreRecord<Buffer>[]): Promise<Uint8Array> {
  return new Promise((resolve, reject) => {
    const toZip = {};
    for (const kv of data) {
      toZip[kv.key] = kv.value;
    }
    zip(toZip, (err, res) => err ? reject(err) : resolve(res));
  });
}

[101arrowz]

As an aside, your original code wouldn't have been multithreaded unless you used AsyncZipDeflate instead of ZipDeflate. On the other hand, the code I just provided is multithreaded by default.

[Umisyus]

Thank you for replying so quickly. I will use zip instead for the smaller files which I have. Thank you for your clarifications.

[Umisyus]

let zipFile = new Zip()

const zipStream = new Transform({

objectMode: true,
async transform(fileData: string, _encoding: BufferEncoding, callback: (err: Error | null, data: Uint8Array) => void) {
    // Push the file data to the zip file
    zipFile.ondata = (_err, chunk, _final) => {
        if (_err) {
            console.error(_err);
        }

        callback(null, chunk)
    }

    let fileName = fileData.split('/').pop();
    addFileToZip(fileName, fileData).then(() => {
        console.log("File Added");
    });
},

})

zipStream.on('close', () => {
console.log("END ZIP");
zipFile.end();
})

let dataStream = await (async () => Readable.from((await readFiles('./images')).slice(0, 2)))();

dataStream.on('close', () => {
console.log("CLOSE");
})

pipeline(dataStream,
zipStream,
fs.createWriteStream('text-streams-2-3-4.zip'),
(err: Error) => {
if (err) {
console.error("PIPELINE ERROR", err);
return;
}
console.log("PIPELINE SUCCESS");
})

[Umisyus]

Got it working now:

[Umisyus]

import { Zip, ZipPassThrough } from "fflate";
import fs, { createReadStream } from "fs";
import path from "path";
import { readdir, stat } from "fs/promises";

let readFiles = async (path = '.') => {
    let folders = await readdir(path);
    let filePaths: string[] = []
    // Read folders
    for await (const folder of folders) {

        if (folder.startsWith('.'))
            continue;

        // Read folders within folders
        await stat(`${path}/${folder}`).then(async (stats) => {

            if (!stats.isDirectory())
                return;
            let files = await readdir(path + '/' + folder)
            // Read files within folders
            files.forEach(async (file) => {
                filePaths.push(`${path}/${folder}/${file}`)
                // console.log('>>>', file);
            })
        })

    }
    return filePaths
}

readFiles('./images').then(processFiles()).then(() => {
    console.log('Done...')
})

function processFiles(): (value: string[]) => void | PromiseLike<void> {
    return async (files) => {
        files = files.map(file => path.resolve(file));
        let zipFile = new Zip();

        let zipWriteStream = fs.createWriteStream("test.zip");

        zipFile.ondata = (err, chunk, final) => {

            zipWriteStream.write(chunk);

            if (err || final) {
                return zipWriteStream.close();
            }
        };

        files
            // .slice()
            .forEach(file => addToZip(file));

        zipFile.end();

        console.log("END ZIP");

        function addToZip(file: string) {
            const fileName = file.split('/').pop();
            const zf = new ZipPassThrough(fileName);
            zipFile.add(zf);
            const fileStream = createReadStream(file);

            fileStream.on('readable', () => {
                let chunk = fileStream.read();
                while (chunk && chunk.length) {
                    zf.push(chunk);
                    chunk = fileStream.read();
                }
            });

            fileStream.on('end', () => {
                zf.push(new Uint8Array(0), true);
            });
        }
    };
}
Edit: Cleaned up code, removed unnecessary comments.

[Umisyus]

This works but there's an off by one error in it.

import { Zip, ZipPassThrough } from "fflate";

import fs, { createReadStream } from "fs";
import path from "path";
import { readdir, stat } from "fs/promises";
import { chunk } from "crawlee";

let readFiles = async (path = '.') => {
    let folders = await readdir(path);
    let filePaths: string[] = []
    // Read folders
    for await (const folder of folders) {

        if (folder.startsWith('.'))
            continue;

        // Read folders within folders
        await stat(`${path}/${folder}`).then(async (stats) => {

            if (!stats.isDirectory())
                return;
            let files = await readdir(path + '/' + folder)
            // Read files within folders
            files.forEach(async (file) => {
                filePaths.push(`${path}/${folder}/${file}`)
                // console.log('>>>', file);
            })
        })

    }
    return filePaths
}
let zipFile = new Zip(

)

zipFile.ondata = (err, chunk, final) => {
    zipWriteStream.write(chunk);

    if (err || final) {
        zipWriteStream.close();
    }
};
let zipWriteStream = fs.createWriteStream("test.zip");

let files = await readFiles('./images')
let flen = files.length;
let current = 0;
for await (const iterator of chunk(files.slice(0,500), 250)) {

    console.log('PROCESSING... ' + (current += iterator.length) + ' / ' + flen);

    await processFiles(iterator).catch(err => console.log(err));
}
zipFile.end();


async function processFiles(files: string[]) {

    files = files.map(file => path.resolve(file));

    console.log("PROCESS ID: " + process.pid);

    await Promise.all(files.map(file => addToZip(file)))
        .then(() => console.log('DONE'))
        .catch(err => console.log(err));

    async function addToZip(file: string): Promise<void> {
        return new Promise<void>(async (resolve, reject) => {

            const fileName = file.split('/').pop();
            const zf = new ZipPassThrough(fileName);

            zipFile.add(zf);
            const fileStream = createReadStream(file);

            fileStream.on('readable', () => {
                let chunk = fileStream.read();
                while (chunk !== undefined && chunk !== null && chunk.length) {
                    zf.push(chunk);
                    chunk = fileStream.read();
                }
            });

            fileStream.on('end', () => {

                zf.push(new Uint8Array(0), true);
                fileStream.close();
                console.log('File added: ' + fileName);
                resolve();
            });
        })
    };
}

[Umisyus]

For some reason I am missing the last file compressed. I get 49 instead of 50.

[Umisyus]

Turns out on Windows Explorer it will not show some files. On 7-zip, it does.