improve: refactor HLS handling in Gaana and SoundCloud sources

1Lucas1apk · 1Lucas1apk · commit e6ed5450b2ea · 2026-01-21T22:30:52.000-04:00
diff --git a/src/playback/hls/HLSHandler.js b/src/playback/hls/HLSHandler.js
@@ -12,7 +12,7 @@ export default class HLSHandler extends PassThrough {
     this.headers = options.headers || {}
     this.localAddress = options.localAddress || null
     this.onResolveUrl = options.onResolveUrl || null
-    this.strategy = options.type?.includes('fmp4') ? 'segmented' : 'streaming'
+    this.strategy = options.strategy || (options.type?.includes('fmp4') ? 'segmented' : 'streaming')
     
     this.fetcher = new SegmentFetcher({ 
       headers: this.headers,
@@ -24,12 +24,13 @@ export default class HLSHandler extends PassThrough {
     this.processedOrder = []
     this.MAX_HISTORY = 200
     this.segmentQueue = []
+    this.MAX_PARALLEL_FETCHES = this.strategy === 'segmented' ? 3 : (this.strategy === 'streaming' ? 2 : 1)
     this.isFetching = false
     this.stop = false
     this.lastMapUri = null
     this.isLive = false
     this.playlistTimer = null
-    this.activeSegmentStream = null
+    this.activeSegmentStreams = new Map()
     this.lastMediaSequence = -1
     this.highestSequence = -1
     this.maxGap = 30
@@ -54,10 +55,10 @@ export default class HLSHandler extends PassThrough {
       clearTimeout(this.playlistTimer)
       this.playlistTimer = null
     }
-    if (this.activeSegmentStream) {
-      this.activeSegmentStream.destroy()
-      this.activeSegmentStream = null
+    for (const stream of this.activeSegmentStreams.values()) {
+      stream.destroy()
     }
+    this.activeSegmentStreams.clear()
     this.segmentQueue = []
     this.processedSegments.clear()
     this.processedOrder = []
@@ -107,31 +108,31 @@ export default class HLSHandler extends PassThrough {
       }
 
       if (parsed.isMaster) {
-        const bestVariant = parsed.variants.reduce((prev, current) => {
-          const hasAudio = (v) => v.codecs?.includes('mp4a') || v.codecs?.includes('opus')
-          if (!hasAudio(prev)) return current
-          if (!hasAudio(current)) return prev
-          return (current.bandwidth > prev.bandwidth) ? current : prev
-        }, parsed.variants[0])
-
-        const itag = bestVariant.url.match(/[\/&]itag[\/](\d+)/)?.[1] || 
-                     bestVariant.url.match(/[?&]itag=(\d+)/)?.[1] || 'unknown'
-
-        logger('debug', 'HLSHandler', `Selected variant itag: ${itag}, bandwidth: ${bestVariant.bandwidth}, codecs: ${bestVariant.codecs}`)
+        const sortedVariants = parsed.variants.sort((a, b) => b.bandwidth - a.bandwidth)
+        const bestVariant = sortedVariants.find(v => 
+          (v.codecs?.includes('mp4a') || v.codecs?.includes('opus')) && 
+          !v.codecs?.includes('avc1')
+        ) || sortedVariants.find(v => 
+          v.codecs?.includes('mp4a') || v.codecs?.includes('opus')
+        ) || sortedVariants[0]
+
+        logger('debug', 'HLSHandler', `Selected variant bandwidth: ${bestVariant.bandwidth}, codecs: ${bestVariant.codecs}`)
         this.currentUrl = bestVariant.url
         return setImmediate(() => this._playlistLoop())
       }
 
       this.isLive = parsed.isLive
 
       if (this.lastMediaSequence !== -1 && (parsed.mediaSequence < this.lastMediaSequence || parsed.mediaSequence > this.lastMediaSequence + this.maxGap)) {
-        logger('warn', 'HLSHandler', `Playlist sequence discontinuity (${this.lastMediaSequence} -> ${parsed.mediaSequence}). Resetting to live edge.`)
-        this.segmentQueue = []
-        this.processedSegments.clear()
-        this.processedOrder = []
-        this.highestSequence = -1
-        this.preRolled = false
-        this.justResynced = true
+        if (this.isLive) {
+          logger('warn', 'HLSHandler', `Playlist sequence discontinuity (${this.lastMediaSequence} -> ${parsed.mediaSequence}). Resetting to live edge.`)
+          this.segmentQueue = []
+          this.processedSegments.clear()
+          this.processedOrder = []
+          this.highestSequence = -1
+          this.preRolled = false
+          this.justResynced = true
+        }
       }
       this.lastMediaSequence = parsed.mediaSequence
 
@@ -142,7 +143,7 @@ export default class HLSHandler extends PassThrough {
       }
 
       const isFirstLoad = this.processedSegments.size === 0
-      if ((isFirstLoad || this.justResynced) && this.isLive) {
+      if (this.isLive && (isFirstLoad || this.justResynced)) {
         if (this.justResynced) {
           this.processedSegments.clear()
           this.processedOrder = []
@@ -159,6 +160,8 @@ export default class HLSHandler extends PassThrough {
           if (seg.sequence !== -1 && seg.sequence > this.highestSequence) this.highestSequence = seg.sequence
         }
         this.justResynced = false
+      } else {
+        this.justResynced = false
       }
 
       const newSegments = parsed.segments.filter(s => {
@@ -209,45 +212,64 @@ export default class HLSHandler extends PassThrough {
     }
   }
 
+  async _fetchWithRetry(segment, attempt = 1) {
+    try {
+      if (this.strategy === 'segmented') {
+        const data = await this.fetcher.fetchSegment(segment, { stream: false })
+        return { segment, data }
+      }
+      const stream = await this.fetcher.fetchSegment(segment, { stream: true })
+      return { segment, stream }
+    } catch (err) {
+      if (this.stop) return null
+      const isRecoverable = err.message === 'aborted' || err.code === 'ECONNRESET' || err.code === 'ETIMEDOUT'
+      if (isRecoverable && attempt <= 3) {
+        const delay = Math.pow(2, attempt) * 500
+        logger('warn', 'HLSHandler', `Segment fetch failed (attempt ${attempt}/3): ${err.message}. Retrying in ${delay}ms...`)
+        await new Promise(r => setTimeout(r, delay))
+        return this._fetchWithRetry(segment, attempt + 1)
+      }
+      logger('error', 'HLSHandler', `Segment fetch permanently failed ${segment.sequence}: ${err.message}`)
+      return null
+    }
+  }
+
   async _fetchSegments() {
     if (this.isFetching || this.stop) return
     this.isFetching = true
 
-    let nextSegmentPromise = null
+    const fetchPool = new Map()
 
-    const getSegment = async (seg) => {
-      try {
-        if (this.strategy === 'segmented') {
-          return { segment: seg, data: await this.fetcher.fetchSegment(seg, { stream: false }) }
-        }
-        return { segment: seg, stream: await this.fetcher.fetchSegment(seg, { stream: true }) }
-      } catch (err) {
-        logger('error', 'HLSHandler', `Segment fetch error ${seg.sequence}: ${err.message}`)
-        return null
+    const fillPool = () => {
+      while (fetchPool.size < this.MAX_PARALLEL_FETCHES && this.segmentQueue.length > 0) {
+        const seg = this.segmentQueue.shift()
+        const key = seg.sequence !== -1 ? seg.sequence : seg.url
+        fetchPool.set(key, this._fetchWithRetry(seg))
       }
     }
 
-    while ((this.segmentQueue.length > 0 || nextSegmentPromise) && !this.stop) {
-      if (this.isLive && this.segmentQueue.length < 3 && !nextSegmentPromise) {
-        await new Promise(r => setTimeout(r, 500))
-        if (this.segmentQueue.length === 0) break
-      }
+    while ((this.segmentQueue.length > 0 || fetchPool.size > 0) && !this.stop) {
+      fillPool()
 
-      let current
-      if (nextSegmentPromise) {
-        current = await nextSegmentPromise
-        nextSegmentPromise = null
-      } else {
-        const seg = this.segmentQueue.shift()
-        current = await getSegment(seg)
+      if (this.isLive && fetchPool.size === 0 && this.segmentQueue.length === 0 && !this.preRolled) {
+        await new Promise(r => setTimeout(r, 500))
+        if (this.segmentQueue.length === 0 && fetchPool.size === 0) break
+        continue
       }
 
-      if (!current) continue
+      if (fetchPool.size === 0) break
 
-      if (this.segmentQueue.length > 0 && !this.stop) {
-        nextSegmentPromise = getSegment(this.segmentQueue.shift())
+      const [key, promise] = fetchPool.entries().next().value
+      fetchPool.delete(key)
+      
+      const current = await promise
+      if (!current) {
+        logger('warn', 'HLSHandler', `Skipping failed segment: ${key}`)
+        continue
       }
 
+      this.preRolled = true
+
       try {
         const { segment, data, stream } = current
         if (segment.map && segment.map.uri !== this.lastMapUri) {
@@ -263,20 +285,20 @@ export default class HLSHandler extends PassThrough {
             if (!this.write(data)) await new Promise(r => this.once('drain', r))
           }
         } else if (stream) {
-          this.activeSegmentStream = stream
+          this.activeSegmentStreams.set(key, stream)
           for await (const chunk of stream) {
             if (this.stop) break
             if (!this.write(chunk)) await new Promise(r => this.once('drain', r))
           }
-          this.activeSegmentStream = null
+          this.activeSegmentStreams.delete(key)
         }
       } catch (err) {
         logger('error', 'HLSHandler', `Segment processing error: ${err.message}`)
       }
     }
 
     this.isFetching = false
-    if (!this.isLive && this.segmentQueue.length === 0 && !this.stop) {
+    if (!this.isLive && this.segmentQueue.length === 0 && fetchPool.size === 0 && !this.stop) {
       this.emit('finishBuffering')
       this.end()
     }
diff --git a/src/sources/gaana.js b/src/sources/gaana.js
@@ -2,8 +2,8 @@
 * Credits: https://github.com/southctrl; adapted for NodeLink
 */
 
-import { PassThrough } from 'node:stream'
 import { encodeTrack, http1makeRequest, logger } from '../utils.js'
+import HLSHandler from '../playback/hls/HLSHandler.js'
 
 const USER_AGENT =
   'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36'
@@ -144,8 +144,11 @@ export default class GaanaSource {
 
   async loadStream(track, url, protocol, additionalData) {
     if (protocol === 'hls') {
-      const stream = new PassThrough()
-      this.streamHlsPlaylist(stream, url)
+      const stream = new HLSHandler(url, {
+        headers: BASE_HEADERS,
+        strategy: 'segmented',
+        localAddress: this.nodelink.routePlanner?.getIP()
+      })
       return { stream, type: 'mpegts' }
     }
 
@@ -155,35 +158,11 @@ export default class GaanaSource {
       return { stream, type: 'mp4' }
     }
 
-    const stream = new PassThrough()
-    this.streamUrl(stream, url)
-    return { stream, type: 'mp4' }
-  }
-
-  async streamUrl(outputStream, url) {
-    try {
-      const { stream, error, statusCode } = await http1makeRequest(url, { method: 'GET', streamOnly: true })
-
-      if (error || statusCode !== 200 || !stream) {
-        throw new Error(error?.message || `Stream status ${statusCode}`)
-      }
-
-      await new Promise((resolve, reject) => {
-        stream.on('data', (chunk) => {
-          if (!outputStream.destroyed) outputStream.write(chunk)
-        })
-        stream.on('end', resolve)
-        stream.on('error', reject)
-      })
-    } catch (e) {
-      logger('warn', 'Gaana', `Stream error: ${e.message}`)
-      if (!outputStream.destroyed) outputStream.emit('error', e)
-    } finally {
-      if (!outputStream.destroyed) {
-        outputStream.emit('finishBuffering')
-        outputStream.end()
-      }
+    const { stream, error, statusCode } = await http1makeRequest(url, { method: 'GET', streamOnly: true, headers: BASE_HEADERS })
+    if (error || statusCode !== 200 || !stream) {
+      throw new Error(error?.message || `Stream status ${statusCode}`)
     }
+    return { stream, type: 'mp4' }
   }
 
   async streamSegments(outputStream, initUrl, segments) {
@@ -206,79 +185,6 @@ export default class GaanaSource {
     }
   }
 
-  async streamHlsPlaylist(outputStream, playlistUrl) {
-    try {
-      const playlist = await this.fetchText(playlistUrl)
-      if (!playlist) throw new Error('Empty HLS playlist')
-
-      const lines = playlist
-        .split('\n')
-        .map((l) => l.trim())
-        .filter(Boolean)
-
-      if (lines.some((l) => l.startsWith('#EXTINF'))) {
-        await this.streamHlsSegments(outputStream, playlistUrl, lines)
-        return
-      }
-
-      const audioTags = lines.filter(
-        (l) => l.startsWith('#EXT-X-MEDIA') && l.includes('TYPE=AUDIO') && l.includes('URI="')
-      )
-
-      if (audioTags.length) {
-        const preferred = audioTags.find((l) => /DEFAULT=YES/.test(l)) || audioTags[audioTags.length - 1]
-        const uri = preferred.match(/URI="([^"]+)"/)?.[1]
-        if (uri) {
-          const audioUrl = new URL(uri, playlistUrl).toString()
-          const audioPlaylist = await this.fetchText(audioUrl)
-          const audioLines = audioPlaylist
-            .split('\n')
-            .map((l) => l.trim())
-            .filter(Boolean)
-
-          await this.streamHlsSegments(outputStream, audioUrl, audioLines)
-          return
-        }
-      }
-
-      const variantUrl = lines.find((l) => !l.startsWith('#'))
-      if (!variantUrl) throw new Error('No HLS variant found')
-
-      const resolved = new URL(variantUrl, playlistUrl).toString()
-      const variantPlaylist = await this.fetchText(resolved)
-      const variantLines = variantPlaylist
-        .split('\n')
-        .map((l) => l.trim())
-        .filter(Boolean)
-
-      await this.streamHlsSegments(outputStream, resolved, variantLines)
-    } catch (e) {
-      if (!outputStream.destroyed) outputStream.emit('error', e)
-    } finally {
-      if (!outputStream.destroyed) {
-        outputStream.emit('finishBuffering')
-        outputStream.end()
-      }
-    }
-  }
-
-  async streamHlsSegments(outputStream, baseUrl, lines) {
-    const segments = []
-
-    for (let i = 0; i < lines.length; i++) {
-      if (lines[i].startsWith('#EXTINF')) {
-        const uri = lines[i + 1]
-        if (uri && !uri.startsWith('#')) segments.push(new URL(uri, baseUrl).toString())
-      }
-    }
-
-    for (const segmentUrl of segments) {
-      if (outputStream.destroyed) break
-      const ok = await this.streamUrlChunk(outputStream, segmentUrl)
-      if (!ok) break
-    }
-  }
-
   async streamUrlChunk(outputStream, url) {
     try {
       const { stream, statusCode, error } = await http1makeRequest(url, {
@@ -307,20 +213,6 @@ export default class GaanaSource {
     }
   }
 
-  async fetchText(url) {
-    const { body, statusCode, error } = await http1makeRequest(url, {
-      method: 'GET',
-      headers: BASE_HEADERS,
-      disableBodyCompression: true
-    })
-
-    if (error || statusCode !== 200 || !body) {
-      throw new Error(error?.message || `Playlist status ${statusCode}`)
-    }
-
-    return typeof body === 'string' ? body : JSON.stringify(body)
-  }
-
   mapTrack(track) {
     const title = track?.title || track?.name
     if (!title) return null
diff --git a/src/sources/soundcloud.js b/src/sources/soundcloud.js
