128 lines
3.1 KiB
TypeScript
128 lines
3.1 KiB
TypeScript
import axios from "axios";
|
|
import { UrlPool } from "./urlpool";
|
|
|
|
export class Downloader {
|
|
|
|
pool: UrlPool;
|
|
chunks: {[U: string]: (Buffer | null)} = {};
|
|
|
|
cache_size = 15;
|
|
|
|
loading: boolean = false;
|
|
destroyed: boolean = false;
|
|
|
|
start_position: number;
|
|
sent_position: number;
|
|
current_position: number;
|
|
|
|
constructor(pool: UrlPool, start_position: number) {
|
|
this.pool = pool;
|
|
this.start_position = start_position;
|
|
this.sent_position = start_position;
|
|
this.current_position = start_position;
|
|
console.log(pool.id, "| new downloader", start_position, "/", pool.total_size);
|
|
}
|
|
|
|
collect() {
|
|
let collected: Buffer[] = [];
|
|
while(true) {
|
|
let id = this.sent_position.toString();
|
|
let c = this.chunks[id];
|
|
if(!c) break;
|
|
this.sent_position += c.length;
|
|
collected.push(c);
|
|
delete this.chunks[id];
|
|
}
|
|
if(collected.length) return Buffer.concat(collected);
|
|
return null;
|
|
}
|
|
|
|
cache() {
|
|
if(this.current_position - this.sent_position > this.cache_size*1024*1024) return;
|
|
for(let i=0; i<this.cache_size; i++) {
|
|
this.download_next(Math.round(1*1024*1024));
|
|
}
|
|
}
|
|
|
|
async more(): Promise<Buffer | null | false> {
|
|
if(this.loading) return false;
|
|
this.loading = true;
|
|
if(this.sent_position > this.pool.total_size-1) return null;
|
|
|
|
this.cache();
|
|
let promise: Promise<Buffer | null> = new Promise((resolve) => {
|
|
let wait_for_result = () => {
|
|
let result = this.collect();
|
|
if(result || this.destroyed) {
|
|
this.loading = false;
|
|
resolve(result);
|
|
clearInterval(interval);
|
|
if(result) console.log(this.pool.id, "| sending", Math.round(result.length/1024), "kB");
|
|
}
|
|
};
|
|
let interval = setInterval(wait_for_result, 1000);
|
|
wait_for_result();
|
|
});
|
|
return promise;
|
|
}
|
|
|
|
async download_next(bytes: number) {
|
|
if(this.destroyed) return;
|
|
let to = this.current_position + bytes;
|
|
if(to > this.pool.total_size-1) to = this.pool.total_size-1;
|
|
let from = this.current_position;
|
|
this.current_position = to+1;
|
|
this.download_part(from, to);
|
|
}
|
|
|
|
async download_part(from: number, to: number) {
|
|
|
|
if(to > this.pool.total_size-1) to = this.pool.total_size-1;
|
|
if(from > this.pool.total_size-1) return;
|
|
|
|
let url = await this.pool.get();
|
|
if(!url) throw "No available URL!";
|
|
|
|
if(this.destroyed) {
|
|
this.pool.return(url[0]);
|
|
return;
|
|
}
|
|
|
|
let controller = new AbortController();
|
|
let tmo = setTimeout(() => {
|
|
console.log(this.pool.id, "| took so long");
|
|
controller.abort();
|
|
}, 10000);
|
|
|
|
try {
|
|
let r = await axios.get(url[1], {
|
|
responseType: 'arraybuffer',
|
|
headers: {
|
|
Range: `bytes=${from}-${to}`
|
|
},
|
|
signal: controller.signal
|
|
});
|
|
this.chunks[from.toString()] = r.data;
|
|
} catch(e) {
|
|
clearTimeout(tmo);
|
|
console.log(this.pool.id, "| failed, retrying");
|
|
setTimeout(() => {
|
|
this.download_part(from, to);
|
|
this.pool.return(url![0], false);
|
|
}, 2000);
|
|
return;
|
|
}
|
|
|
|
clearTimeout(tmo);
|
|
this.pool.return(url[0]);
|
|
|
|
}
|
|
|
|
destroy() {
|
|
this.destroyed = true;
|
|
let index = this.pool.downloaders.indexOf(this);
|
|
this.pool.downloaders.splice(index, 1);
|
|
console.log(this.pool.id, "| destroyed downloader", this.start_position, "/", this.pool.total_size);
|
|
}
|
|
|
|
} |