MWSE/frontend/P2PFileSender.ts

import "webrtc-adapter";
import WebRTC from "./WebRTC";
import Peer from "./Peer";



export default class P2PFileSender
{
    public rtc : RTCPeerConnection;
    public peer : Peer;
    public webrtc : WebRTC;

    public totalSize : number = 0;
    public isReady : boolean = false;
    public isStarted : boolean = false;
    public isSending : boolean = false;
    public isRecaiving : boolean = false;
    public processedSize : number = 0;
    public recaivedFile? : File;

    public bufferSizePerChannel : number = 10e6;
    public bufferSizePerPack : number = 10e3;
    public safeBufferSizePerPack : number = 10e3 - 1;

    public constructor(webrtc : WebRTC, peer : Peer)
    {
        this.webrtc = webrtc;
        this.rtc = webrtc.rtc;
        this.peer = peer;
    }
    public async RecaiveFile(
        _rtc: RTCPeerConnection,
        fileMetadata: {name:string, type:string},
        channelCount: number,
        _totalSize: number,
        onEnded: Function
    )
    {
        //let totals = {};
       // let index = 0;
        /*setChannelStatus(Array.from({length:channelCount}).map((e, index) => {
            return {
                name: `${index+1}. Kanal`,
                current: 0,
                currentTotal: 0,
                total: 0
            }
        }));*/
        let parts : Blob[] = [];
        this.webrtc.on('datachannel',(datachannel:RTCDataChannel) => {
            //let channelIndex = index++;
            let current =  0;
            let totalSize = 0;
            let currentPart = 0;
            let bufferAmount : ArrayBuffer[] = [];
            datachannel.onmessage = function({data}){
                if(totalSize == 0)
                {
                    let {
                        size,
                        part,
                    } = JSON.parse(data);
                    totalSize = size;
                    currentPart = part;
                    /*updateChannelStatus(channelIndex, n => {
                        return {
                            ...n,
                            total: totalSize,
                            current: 0
                        }
                    });*/
                    datachannel.send("READY");
                }else{
                    current += data.byteLength;
                    bufferAmount.push(data);
                    /*updateChannelStatus(channelIndex, n => {
                        return {
                            ...n,
                            current: data.byteLength + n.current,
                            currentTotal: data.byteLength + n.currentTotal,
                        }
                    });
                    setProcessedSize(n => n + data.byteLength);*/
                    if(current == totalSize)
                    {
                        parts[currentPart] = new Blob(bufferAmount);
                        bufferAmount = []; 
                        //totals[datachannel.label] += totalSize;
                        totalSize = 0;
                        currentPart = 0;
                        current = 0;
                        datachannel.send("TOTAL_RECAIVED");
                    }
                }
            };
            datachannel.onclose = () => {
                channelCount--;
                if(channelCount == 0)
                {
                    let file = new File(parts, fileMetadata.name, {
                        type: fileMetadata.type,
                        lastModified: +new Date
                    });
                    onEnded(file);
                }
            };
        })
    }
    public async SendFile(
        file: File,
        metadata: object
    )
    {
        this.isSending = true;
        this.isStarted = true;


        let buffer = await file.arrayBuffer();
        let partCount = Math.ceil(buffer.byteLength / 10e6);
        let channelCount = Math.min(5, partCount);

        if(this.webrtc.iceStatus != "connected")
        {
            throw new Error("WebRTC is a not ready")
        }

        this.peer.send({
            type: 'file',
            name: file.name,
            size: file.size,
            mimetype: file.type,
            partCount,
            channelCount,
            metadata: metadata
        });

        let channels : RTCDataChannel[] = [];

        for(let channelIndex = 0; channelIndex < channelCount; channelIndex++)
        {
            let channel = this.rtc.createDataChannel("\\?\\file_" + channelIndex);
            channel.binaryType = "arraybuffer";
            await new Promise(ok => {
                channel.onopen = () => {
                    ok(void 0);
                }
            });
            channels.push(channel);
        };

        let currentPart = 0;
        let next = () => {
            if(currentPart < partCount)
            {
                let bufferPart = buffer.slice(currentPart * 10e6, currentPart * 10e6 + 10e6)
                currentPart++;
                return [bufferPart, currentPart - 1];
            };
            return [false,0];
        };
        let spyChannelIndex = channels.length;
        await new Promise(ok => {
            for (let channelIndex = 0; channelIndex < channels.length; channelIndex++)
            {
                this.sendPartition(
                    channels[channelIndex],
                    next,
                    channelIndex,
                    () => {
                        spyChannelIndex--;
                        if(spyChannelIndex == 0)
                        {
                            this.isSending = false;
                            this.isStarted = false;
                            ok(undefined)
                        }
                    }
                );
            }
        })
    }
    protected sendPartition(
        channel: RTCDataChannel,
        nextblob10mb: () => (number | ArrayBuffer)[] | (number | boolean)[],
        _channelIndex: number,
        onEnded: Function
    )
    {
        let [currentBuffer,currentPartition] = nextblob10mb();
        let currentPart = 0;
        let next = () => {
            if(!(currentBuffer instanceof ArrayBuffer))
            {
                return;
            }
            let bufferPart = currentBuffer.slice(currentPart * 16e3, currentPart * 16e3 + 16e3)
            currentPart++;
            if(bufferPart.byteLength != 0)
            {
                /*
                updateChannelStatus(channelIndex, n => {
                    return {
                        ...n,
                        current: bufferPart.byteLength + n.current,
                        currentTotal: bufferPart.byteLength + n.currentTotal
                    }
                });
                setProcessedSize(n => n + bufferPart.byteLength);
                */
                return bufferPart
            }
        };
        channel.addEventListener("message",({data}) => {
            if(data == "READY")
            {
                this.sendFileChannel(channel, next)
            }
            if(data == "TOTAL_RECAIVED")
            {
                [currentBuffer,currentPartition] = nextblob10mb();
                currentPart = 0;
                if(currentBuffer != false)
                {
                    /*updateChannelStatus(channelIndex, n => {
                        return {
                            ...n,
                            total: currentBuffer.byteLength,
                            current: 0,
                        }
                    });*/
                    channel.send(JSON.stringify({
                        size: (currentBuffer as ArrayBuffer).byteLength,
                        part: currentPartition
                    }))
                }else{
                    channel.close();
                    onEnded();
                }
            }
        });
        channel.send(JSON.stringify({
            size: (currentBuffer as ArrayBuffer).byteLength,
            part: currentPartition
        }))
    }
    protected sendFileChannel(
        channel: RTCDataChannel,
        getNextBlob: () => ArrayBuffer | undefined
    )
    {
        channel.addEventListener("bufferedamountlow",function(){
            let buffer = getNextBlob();
            if(buffer)
            {
                channel.send(buffer);
            }
        });
        channel.bufferedAmountLowThreshold = 16e3 - 1;
        let c = getNextBlob();
        c && channel.send(c);
    }
};