Stop downloading the torrent file when I stop streaming the file

[rudnypc]

When i stop downloading the file, the torrent may stop downloading and be removed. I see that it is not the correct file and I will download another one, but the old one keeps downloading.

[KiraLT]

File stopping is implemented:

torrent-stream-server/src/api/stream.ts

Lines 78 to 80 in 8adae33

	return pump(file.createReadStream(range), res, () => {
	file.stop()
	})

But I also noticed that it doesn't work in some cases. I will need to find a better solution for this.

[KiraLT]

Added fix for frontend player, now it's destroyed when react component is destoyed. However, file stopping is buggy webtorrent/webtorrent#164. There is a pending PR that could fix it (webtorrent/webtorrent#2115) or I need to find a workaround.

[omgbox]

The issue you're describing is that the download does not stop when the user stops streaming. This is likely due to the fact that the pump function continues to stream data even after the user has stopped requesting it. The pump function does not inherently handle the cancellation of streams when the client disconnects.

To address this, you can use the res.on('close', ...) event to detect when the client has disconnected and then stop the streaming process. Here's how you can modify your code to handle this:

import { Readable } from 'stream'
import rangeParser from 'range-parser'
import { HttpError, HttpStatusCodes } from 'common-stuff'

import { TorrentClient, filterFiles } from '../services/torrent-client'
import { Globals } from '../config'
import { verifyJwtToken, getSteamUrl } from '../helpers'
import { createRoute, Route, getRouteUrl } from '../helpers/openapi'

/**
 * Creates a router for handling streaming requests.
 * @param {Globals} config - The global configuration object.
 * @param {TorrentClient} client - The TorrentClient instance for handling torrent operations.
 * @returns {Route[]} An array of routes for handling streaming requests.
 */
export function getStreamRouter({ config }: Globals, client: TorrentClient): Route[] {
    // Get the encoding token from the configuration
    const encodeToken = config.security.streamApi.key || config.security.apiKey

    /**
     * Type definition for the parameters used in the streaming requests.
     * @typedef {Object} Params
     * @property {string} torrent - The torrent link.
     * @property {string} [file] - The file name.
     * @property {string} [fileType] - The file type.
     * @property {number} [fileIndex] - The file index.
     */
    type Params = { torrent: string; file?: string; fileType?: string; fileIndex?: number }

    /**
     * Parses the parameters for the streaming request.
     * If the encoding token is present, it verifies the JWT token and returns the decoded data.
     * Otherwise, it returns the provided parameters.
     * @param {Params} params - The parameters to parse.
     * @returns {Params} The parsed parameters.
     * @throws {HttpError} If the parameters are invalid or the JWT token is incorrect.
     */
    const parseParams = (params: Params): Params => {
        if (encodeToken) {
            if (params.file || params.fileIndex || params.fileType) {
                throw new HttpError(
                    HttpStatusCodes.BAD_REQUEST,
                    `All parameters should be encoded with JWT`
                )
            }

            const data = verifyJwtToken<Params>(
                params.torrent,
                encodeToken,
                config.security.streamApi.maxAge
            )

            if (!data) {
                throw new HttpError(HttpStatusCodes.FORBIDDEN, 'Incorrect JWT encoding')
            }

            return data
        }

        return params
    }

    return [
        /**
         * Route for handling the main streaming request.
         * @param {Request} req - The HTTP request object.
         * @param {Response} res - The HTTP response object.
         * @returns {Promise<void>} A promise that resolves when the streaming is complete.
         */
        createRoute('getStream', async (req, res) => {
            // Parse the parameters from the request
            const { torrent: link, ...params } = parseParams({
                torrent: req.params.torrent,
                ...req.query,
            })

            // Add the torrent and get the torrent object
            const torrent = await client.addTorrent(link)

            // Filter the files based on the parameters and get the first file
            const file = filterFiles(torrent.files, params)[0]

            // If no file is found, throw a 404 error
            if (!file) {
                throw new HttpError(HttpStatusCodes.NOT_FOUND)
            }

            // Set the response headers
            res.setHeader('Accept-Ranges', 'bytes')
            res.attachment(file.name)
            req.connection.setTimeout(3600000)

            // Parse the range header if present
            const parsedRange = req.headers.range
                ? rangeParser(file.length, req.headers.range)
                : undefined
            const range = parsedRange instanceof Array ? parsedRange[0] : undefined

            // Set the response headers based on the range
            if (range) {
                res.statusCode = 206
                res.setHeader('Content-Length', range.end - range.start + 1)
                res.setHeader(
                    'Content-Range',
                    'bytes ' + range.start + '-' + range.end + '/' + file.length
                )
            } else {
                res.setHeader('Content-Length', file.length)
            }

            // If the request method is HEAD, end the response
            if (req.method === 'HEAD') {
                return res.end()
            }

            // Create a readable stream for the file
            const stream = file.createReadStream(range)

            // Handle the 'close' event to stop the stream when the client disconnects
            res.on('close', () => {
                stream.destroy()
                file.stop()
            })

            // Pipe the stream to the response and handle errors
            stream.pipe(res).on('error', (err) => {
                console.error('Stream error:', err)
                file.stop()
            })
        }),

        /**
         * Route for redirecting to the main streaming route.
         * @param {Request} req - The HTTP request object.
         * @param {Response} res - The HTTP response object.
         * @returns {Promise<void>} A promise that resolves when the redirection is complete.
         */
        createRoute('getStream2', async (req, res) => {
            const { torrent, ...query } = req.query

            return res.redirect(getRouteUrl('getStream', { torrent }, query), 301)
        }),

        /**
         * Route for generating a playlist file.
         * @param {Request} req - The HTTP request object.
         * @param {Response} res - The HTTP response object.
         * @returns {Promise<void>} A promise that resolves when the playlist is generated.
         */
        createRoute('getPlaylist', async (req, res) => {
            const domain = req.protocol + '://' + req.get('host')
            const { torrent: link, ...params } = parseParams({
                torrent: req.params.torrent,
                ...req.query,
            })

            // Add the torrent and get the torrent object
            const torrent = await client.addTorrent(link)

            // Filter the files based on the parameters and get the video and audio files
            const files = filterFiles(torrent.files, params).filter(
                (v) => v.type.includes('video') || v.type.includes('audio')
            )

            // Set the response timeout and headers
            req.connection.setTimeout(10000)
            res.attachment(torrent.name + `.m3u`)

            // Generate the playlist content
            return res.send(
                [
                    '#EXTM3U',
                    ...files.flatMap((f) => [
                        `#EXTINF:-1,${f.name}`,
                        `${domain}${getSteamUrl(link, f.path, encodeToken)}`,
                    ]),
                ].join('\n')
            )
        }),

        /**
         * Route for redirecting to the playlist generation route.
         * @param {Request} req - The HTTP request object.
         * @param {Response} res - The HTTP response object.
         * @returns {Promise<void>} A promise that resolves when the redirection is complete.
         */
        createRoute('getPlaylist2', async (req, res) => {
            const { torrent, ...query } = req.query

            return res.redirect(getRouteUrl('getPlaylist', { torrent }, query), 301)
        }),
    ]
}

Explanation:
res.on('close', ...) Event: This event is triggered when the client disconnects from the server. By listening to this event, you can detect when the client has stopped streaming and then stop the streaming process.

stream.destroy(): This method is used to destroy the stream, which will stop any further data from being sent.

file.stop(): This method is used to stop the file from being read further.

By adding these lines, the streaming process will be properly stopped when the client disconnects, preventing unnecessary resource consumption.