split-stream.mjs

import fs from 'fs'
import fsPromise from 'node:fs/promises'
import { Transform, pipeline } from 'node:stream'
import EventEmitter from 'node:events'
import path from 'node:path'
import crypto from 'crypto';
class SplitEmitter extends EventEmitter{}
const splitter=new SplitEmitter()

async function createReader(path) {
    const fd=await fsPromise.open(path,'r');
    const read= fd.createReadStream()
    return read
}
function createWriter(pathNum) {
    const outputDir="video-part"
    if(!fs.existsSync(outputDir)){
        fs.mkdirSync(outputDir)
    }
    const name=`${outputDir}/part${pathNum}.mp4`
    return fs.createWriteStream(name)
}
function createChecker(emitter, limit) {
    let acc = 0;
    return new Transform({
        transform(chunk, enc, cb) {
            let offset = 0;
            while (offset < chunk.length) {
                const remainingToLimit = limit - acc;
                if (chunk.length - offset >= remainingToLimit) {
                    // slice that completes the part
                    const piece = chunk.subarray(offset, offset + remainingToLimit);
                    this.push(piece);                 // emit piece downstream
                    emitter.emit('split');            // request a split
                    offset += remainingToLimit;
                    acc = 0;                          // reset for next part
                } else {
                    // not enough to reach limit, pass remainder
                    const piece = chunk.subarray(offset);
                    this.push(piece);
                    acc += piece.length;
                    offset = chunk.length;
                }
            }
            cb();
        }
    });
}

/* function createChecker(emitter,limit) {
    let bytes=0;
    return new Transform({
        transform(chunk,enc,cb){
            bytes+=chunk.length
            if (bytes>=limit) {
                emitter.emit('split')
                bytes=bytes%limit
            }
            cb(null,chunk)
        }
    })
} */
/* function handleSplit(emitter,getNewWriter){
    let currentWriter=getNewWriter()
    let swapping=false
    let pendingSwap=false
    let carryChunk = null
    emitter.on('split',()=>{
        if (pendingSwap) {
            return
        }
        pendingSwap=true
        console.log('split signal recieved (marked pending)//////////////////////////////////');
    })
    function doSwap(oldWriter) {
        if (swapping) {
            return
        }

        swapping=true
        const finishswap=()=>{
            oldWriter.end(()=>{
                console.log('oldWriter finished');
                currentWriter=getNewWriter()
                pendingSwap=false
                swapping=false
                if (carryChunk) {
                    currentWriter.write(carryChunk);
                    carryChunk = null;
                }
                console.log(' new writer created===============================================\n')
            })
            
        }
        if (oldWriter.writableNeedDrain) {
            oldWriter.once('drain',finishswap)
        }else{
            finishswap()
            console.log('new writer created immediately (already ended)==================')
        }}
    return new Transform({
        transform(chunk,enc,cb){
            if (currentWriter.writableEnded) {
                console.warn('attempt to writer has ended, skipping chunk')
                return cb()
            }
            const canWrite=currentWriter.write(chunk)
            const afterWrite=()=>{
                if (pendingSwap&&!swapping) {
                    carryChunk = chunk
                    doSwap(currentWriter)
                }
                cb()
            }
            if (!canWrite) {
                currentWriter.once('drain',()=>{
                    afterWrite()
                    console.log('finished writing for spliting')
                })
            }else{
                afterWrite()
                console.log('finished writing-----------------------------------')
            }
        },
        flush(cb){
            currentWriter.end(()=>{
                console.log('final writer closed')
                cb()
            })
        }
    })
} */
function handleSplit(emitter, getNewWriter) {
    let currentWriter = getNewWriter();
    let swapping = false;
    let pendingSwap = false;
    // track how many bytes have been written into the current part
    let partBytes = 0;

    // When checker emits split we mark pending; actual swap happens inside transform
    emitter.on('split', () => {
        if (pendingSwap) return;
        pendingSwap = true;
        console.log('split signal received (marked pending)');
    });

    async function finishOldAndMakeNew(oldWriter) {
        // prevent further writes to oldWriter by marking swapping true (already set by caller)
        // ensure all buffered data flushed before end: if needDrain true, wait for drain; then end.
        const waitForDrain = () => new Promise(res => {
            if (oldWriter.writableNeedDrain) {
                oldWriter.once('drain', res);
            } else {
                // small tick to allow any queued writes to be processed
                process.nextTick(res);
            }
        });

        try {
            await waitForDrain();
        } catch (e) { /* ignore */ }

        return new Promise((resolve, reject) => {
            oldWriter.end(() => {
                // create new writer after old one finished
                currentWriter = getNewWriter();
                partBytes = 0;           // reset counter for new part
                pendingSwap = false;
                swapping = false;
                console.log('oldWriter finished, new writer created');
                resolve();
            });
        });
    }

    return new Transform({
        transform(chunk, enc, cb) {
            // We will process the chunk possibly splitting it across parts.
            // If a swap is pending, we must finalize current part before writing remainder.
            (async () => {
                let offset = 0;

                while (offset < chunk.length) {
                    // if pending swap and not yet swapping: prepare to finalize current part before writing remainder
                    if (pendingSwap && !swapping) {
                        swapping = true;
                        // wait for old writer to flush and create new writer
                        await finishOldAndMakeNew(currentWriter);
                    }

                    // compute how many bytes we can write into current part before it reaches limit
                    // NOTE: we rely on createChecker to signal splits; here partBytes tracks bytes written to this writer
                    // but since checker already emitted split, pendingSwap would be true only when we've reached
                    // or passed the limit — so we aim to write remainder into the new writer.
                    const canWriteNow = currentWriter.write(chunk.subarray(offset));
                    offset = chunk.length; // we've handed remaining bytes to current writer (either old or new)

                    // if write returned false, wait for drain before continuing
                    if (!canWriteNow) {
                        await new Promise(res => currentWriter.once('drain', res));
                    }
                }

                cb();
            })().catch(err => cb(err));
        },

        flush(cb) {
            // finalize the last writer
            currentWriter.end(() => {
                console.log('final writer closed');
                cb();
            });
        }
    });
}

(async () => {
    try {
        const readStream=await createReader('./Node.js-Express-Course-Build-4-Projects.mp4')
        const limit=1024*1024+1
        let partNum=1
        const getNewWriter=()=>{
            const w=createWriter(partNum++)
            console.log(`created writer for part${partNum-1}.mp4`)
            return w
        }
        const checker=createChecker(splitter,limit)
        const switcher=handleSplit(splitter,getNewWriter)
        pipeline(readStream,checker,switcher,
            (err)=>{
                if (err) {
                    console.error('pipeline failed:',err)
                }else{
                    console.log('pipeline completed sucesfully')
                }
            }
        )
    } catch (error) {
        throw error
    }
})
/* function cheking() {

    const partsDir = './video-part';
    const files = fs.readdirSync(partsDir).filter(f => f.endsWith('.mp4'));
    const totalSize = files.reduce((sum, f) => sum + fs.statSync(path.join(partsDir, f)).size, 0);
    const finalSize = fs.statSync('./Node.js-Express-Course-Build-4-Projects.mp4').size;
    console.log('derived size:', totalSize);
    console.log('real size:', finalSize);
} */

function fileHash(file) {
    return new Promise((res, rej) => {
        const hash = crypto.createHash('sha256');
        const s = fs.createReadStream(file);
        s.on('error', rej);
        s.on('data', d => hash.update(d));
        s.on('end', () => res(hash.digest('hex')));
    });
}

// after pipeline completes:
const parts = fs.readdirSync('./video-part').sort((a,b)=>{
            const numA = parseInt(a.match(/part(\d+)\.mp4/)[1], 10);
            const numB = parseInt(b.match(/part(\d+)\.mp4/)[1], 10);
            return numA - numB});
const combinedHash = await (async () => {
    // stream parts into a single hash without creating a file
    const hash = crypto.createHash('sha256');
    for (const p of parts) {
        const s = fs.createReadStream(path.join('./video-part', p));
        await new Promise((res, rej) => {
            s.on('data', d => hash.update(d));
            s.on('error', rej);
            s.on('end', res);
        });
    }
    return hash.digest('hex');
})();
const originalHash = await fileHash('./Node.js-Express-Course-Build-4-Projects.mp4');
//console.log('originalHash', originalHash, 'combinedHash', combinedHash);
if (originalHash===combinedHash) {
    console.log("true")
}else{console.log('false')}