Socket issue tried to fix

This commit is contained in:
killer069 2021-08-30 15:58:23 +05:30
parent 5bef17528a
commit 7b1f082c27

View File

@ -1,7 +1,7 @@
import { PassThrough } from 'stream' import { PassThrough } from 'stream'
import got from 'got' import got from 'got'
import Request from 'got/dist/source/core';
import { StreamType } from '../stream'; import { StreamType } from '../stream';
import { Socket } from 'net'
export interface FormatInterface{ export interface FormatInterface{
url : string; url : string;
@ -17,6 +17,7 @@ export class LiveStreaming{
private interval : number private interval : number
private packet_count : number private packet_count : number
private timer : NodeJS.Timer | null private timer : NodeJS.Timer | null
private socket : Socket | null
private segments_urls : string[] private segments_urls : string[]
constructor(dash_url : string, target_interval : number){ constructor(dash_url : string, target_interval : number){
this.type = StreamType.Arbitrary this.type = StreamType.Arbitrary
@ -26,6 +27,7 @@ export class LiveStreaming{
this.segments_urls = [] this.segments_urls = []
this.packet_count = 0 this.packet_count = 0
this.timer = null this.timer = null
this.socket = null
this.interval = target_interval * 1000 || 0 this.interval = target_interval * 1000 || 0
this.stream.on('close', () => { this.stream.on('close', () => {
this.cleanup() this.cleanup()
@ -45,6 +47,8 @@ export class LiveStreaming{
private cleanup(){ private cleanup(){
clearTimeout(this.timer as NodeJS.Timer) clearTimeout(this.timer as NodeJS.Timer)
this.socket?.destroy()
this.socket = null
this.timer = null this.timer = null
this.url = '' this.url = ''
this.base_url = '' this.base_url = ''
@ -88,6 +92,7 @@ export class LiveEnded{
private base_url : string; private base_url : string;
private packet_count : number private packet_count : number
private segments_urls : string[] private segments_urls : string[]
private socket : Socket | null
constructor(dash_url : string){ constructor(dash_url : string){
this.type = StreamType.Arbitrary this.type = StreamType.Arbitrary
this.url = dash_url this.url = dash_url
@ -95,6 +100,7 @@ export class LiveEnded{
this.stream = new PassThrough({ highWaterMark : 10 * 1000 * 1000 }) this.stream = new PassThrough({ highWaterMark : 10 * 1000 * 1000 })
this.segments_urls = [] this.segments_urls = []
this.packet_count = 0 this.packet_count = 0
this.socket = null
this.stream.on('close', () => { this.stream.on('close', () => {
this.cleanup() this.cleanup()
}) })
@ -112,6 +118,8 @@ export class LiveEnded{
} }
private cleanup(){ private cleanup(){
this.socket?.destroy()
this.socket = null
this.url = '' this.url = ''
this.base_url = '' this.base_url = ''
this.segments_urls = [] this.segments_urls = []
@ -126,6 +134,10 @@ export class LiveEnded{
await this.dash_getter() await this.dash_getter()
if(this.packet_count === 0) this.packet_count = Number(this.segments_urls[0].split('sq/')[1].split('/')[0]) if(this.packet_count === 0) this.packet_count = Number(this.segments_urls[0].split('sq/')[1].split('/')[0])
for await (let segment of this.segments_urls){ for await (let segment of this.segments_urls){
if(this.stream.destroyed){
this.cleanup()
break
}
if(Number(segment.split('sq/')[1].split('/')[0]) !== this.packet_count){ if(Number(segment.split('sq/')[1].split('/')[0]) !== this.packet_count){
continue continue
} }
@ -151,6 +163,7 @@ export class Stream {
private per_sec_bytes : number; private per_sec_bytes : number;
private duration : number; private duration : number;
private timer : NodeJS.Timer | null private timer : NodeJS.Timer | null
private socket : Socket | null
constructor(url : string, type : StreamType, duration : number){ constructor(url : string, type : StreamType, duration : number){
this.url = url this.url = url
this.type = type this.type = type
@ -158,12 +171,14 @@ export class Stream {
this.bytes_count = 0 this.bytes_count = 0
this.per_sec_bytes = 0 this.per_sec_bytes = 0
this.timer = null this.timer = null
this.duration = duration; this.duration = duration
this.socket = null;
(duration > 300) ? this.loop_start() : this.normal_start() (duration > 300) ? this.loop_start() : this.normal_start()
} }
private cleanup(){ private cleanup(){
clearTimeout(this.timer as NodeJS.Timer) clearTimeout(this.timer as NodeJS.Timer)
this.socket?.destroy()
this.timer = null this.timer = null
this.url = '' this.url = ''
this.bytes_count = 0 this.bytes_count = 0
@ -176,6 +191,7 @@ export class Stream {
return return
} }
let stream = got.stream(this.url) let stream = got.stream(this.url)
this.socket = stream.socket as Socket
stream.pipe(this.stream) stream.pipe(this.stream)
} }
@ -193,9 +209,10 @@ export class Stream {
this.bytes_count += chunk.length this.bytes_count += chunk.length
this.stream.write(chunk) this.stream.write(chunk)
}) })
this.socket = stream.socket as Socket
stream.on('data', () => { stream.on('data', () => {
if(this.bytes_count > (this.per_sec_bytes * 300)){ if(this.bytes_count > (this.per_sec_bytes * 300)){
this.socket?.destroy()
stream.destroy() stream.destroy()
} }
}) })
@ -222,9 +239,10 @@ export class Stream {
this.bytes_count += chunk.length this.bytes_count += chunk.length
this.stream.write(chunk) this.stream.write(chunk)
}) })
this.socket = stream.socket as Socket
stream.on('data', () => { stream.on('data', () => {
if(absolute_bytes > (this.per_sec_bytes * 300)){ if(absolute_bytes > (this.per_sec_bytes * 300)){
this.socket?.destroy()
stream.destroy() stream.destroy()
} }
}) })