File Transport begin

frontend/P2PFileSender.ts

@@ -0,0 +1,262 @@
+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);
+    }
+};

frontend/WebRTC.ts

@@ -1,3 +1,4 @@
+import P2PFileSender from "./P2PFileSender";
 import Peer from "./Peer";
 import "webrtc-adapter";
 interface TransferStreamInfo
@@ -27,6 +28,8 @@ export default class WebRTC
 
     public peer? : Peer;
 
+    public FileTransportChannel? : P2PFileSender;
+
     constructor()
     {
         this.rtc = new RTCPeerConnection({
@@ -197,7 +200,7 @@ export default class WebRTC
         if(this.rtc)
         {
             this.rtc.close();
-            // this.rtc = undefined;
+            //this.rtc = undefined;
         };
         this.emit('disconnected');
         WebRTC.channels.delete(this.id);
@@ -367,6 +370,46 @@ export default class WebRTC
     
         this.sendingStream.clear();
     }
+    public async SendFile(file:File, meta: object)
+    {
+        if(!this.peer)
+        {
+            throw new Error("Peer is not ready");
+        }
+        this.FileTransportChannel = new P2PFileSender(this, this.peer);
+
+        await this.FileTransportChannel.SendFile(file, meta);
+    }
+    public async RecaiveFile(
+        chnlCount:number,
+        filemeta: {
+            name: string;
+            type: string;
+        },
+        totalSize: number
+    ) : Promise<File>
+    {
+        if(!this.peer)
+        {
+            throw new Error("Peer is not ready");
+        }
+        this.FileTransportChannel = new P2PFileSender(this, this.peer);
+
+        return await new Promise(recaivedFile => {
+            if(this.FileTransportChannel)
+            {
+                this.FileTransportChannel.RecaiveFile(
+                    this.rtc,
+                    filemeta,
+                    chnlCount,
+                    totalSize,
+                    (file: File) => {
+                        recaivedFile(file)
+                    }
+                );
+            }
+        })
+    }
 }
 
 WebRTC.requireGC = false;