长时间后如何保持请求打开以使用 write() 方法

Question

我需要保持连接打开，所以在完成音乐后我会写入新数据。问题是我这样做的方式，流在第一首歌之后就停止了。 如何保持连接打开并播放下一首歌曲？

const fs = require('fs');
const express = require('express');
const app = express();
const server = require('http').createServer(app)
const getMP3Duration = require('get-mp3-duration')

let sounds = ['61880.mp3', '62026.mp3', '62041.mp3', '62090.mp3', '62257.mp3', '60763.mp3']

app.get('/current', async (req, res) => {
    let readStream = fs.createReadStream('sounds/61068.mp3')
    let duration = await getMP3Duration(fs.readFileSync('sounds/61068.mp3'))

    let pipe = readStream.pipe(res, {end: false})

    async function put(){
        let file_path = 'sounds/'+sounds[Math.random() * sounds.length-1]

        duration = await getMP3Duration(fs.readFileSync(file_path))

        readStream = fs.createReadStream(file_path)

        readStream.on('data', chunk => {
            console.log(chunk)
            pipe.write(chunk)
        })

        console.log('Current Sound: ', file_path)

        setTimeout(put, duration)
    }

    setTimeout(put, duration)
})

server.listen(3005, async function () {
    console.log('Server is running on port 3005...')
});

Answer 1

经过几次修改后的最终解决方案：

我怀疑您的主要问题在于您的随机数组元素生成器。你需要用Math.floor 包裹你所拥有的东西以四舍五入以确保你最终得到一个整数：

sounds[Math.floor(Math.random() * sounds.length)]

另外，Readstream.pipe 返回目的地，所以你所做的事情是有意义的。但是，在您已经通过管道传输之后，在您的可读设备上调用 on('data') 可能会得到意想不到的结果。 node.js streams docs 提到了这一点。我在本地机器上测试了您的代码，这似乎不是问题，但更改它可能是有意义的，这样您将来就不会遇到问题。

选择一种 API 样式

可读流 API 跨越多个 Node.js 版本发展，并提供多种消费流数据的方法。一般来说，开发人员应该选择一种消费数据的方法，并且永远不应该使用多种方法来消费来自单个流的数据。具体来说，结合使用 on('data')、on('readable')、pipe() 或异步迭代器可能会导致不直观的行为。

我不再调用on('data') 和res.write，而是将readStream 再次通过管道连接到res。此外，除非您真的想获得持续时间，否则我会将该库拉出并使用readStream.end 事件对put() 进行额外调用。这是有效的，因为您在管道时传递了false 选项，这会禁用写入流上的默认end 事件功能并使其保持打开状态。但是，它仍然会发出，因此您可以将其用作标记来了解可读文件何时完成管道。这是重构后的代码：

const fs = require('fs');
const express = require('express');
const app = express();
const server = require('http').createServer(app)
//const getMP3Duration = require('get-mp3-duration') no longer needed

let sounds = ['61880.mp3', '62026.mp3', '62041.mp3', '62090.mp3', '62257.mp3', '60763.mp3']

app.get('/current', async (req, res) => {
    let readStream = fs.createReadStream('sounds/61068.mp3')
    let duration = await getMP3Duration(fs.readFileSync('sounds/61068.mp3'))

    let pipe = readStream.pipe(res, {end: false})

    function put(){
        let file_path = 'sounds/'+sounds[Math.floor(Math.random() * sounds.length)]

        readStream = fs.createReadStream(file_path)
        
        // you may also be able to do readStream.pipe(res, {end: false})
        readStream.pipe(pipe, {end: false})

        console.log('Current Sound: ', file_path)

        readStream.on('end', () => {
            put()
        });
    }

    readStream.on('end', () => {
        put()
    });
})

server.listen(3005, async function () {
    console.log('Server is running on port 3005...')
});

Answer 2

Express 通过对单个请求返回单个响应来工作。发送请求后，需要立即生成新请求以触发新响应。

但是，在您的情况下，您希望继续从单个请求中生成新的响应。

可以使用两种方法来解决您的问题：

1. 更改创建响应的方式以满足您的用例。
2. 使用即时通信框架 (websocket)。我想到的最好和最简单的是socket.io

适配快递

这里的解决方案是按照这个程序进行：

1. 端点/current上的请求进来
2. 音频序列已准备好
3. 返回整个序列的流

所以你的处理程序看起来像这样：

const fs = require('fs');
const express = require('express');
const app = express();
const server = require('http').createServer(app);
// Import the PassThrough class to concatenate the streams
const { PassThrough } = require('stream');
// The array of sounds now contain all the sounds
const sounds = ['61068.mp3','61880.mp3', '62026.mp3', '62041.mp3', '62090.mp3', '62257.mp3', '60763.mp3'];


// function which concatenate an array of streams
const concatStreams = streamArray => {
  let pass = new PassThrough();
  let waiting = streamArray.length;
  streamArray.forEach(soundStream => {
    pass = soundStream.pipe(pass, {end: false});
    soundStream.once('end', () => --waiting === 0 && pass.emit('end'));
  });
  return pass;
};

// function which returns a shuffled array
const shuffle = (array) => {
  const a = [...array]; // shallow copy of the array
  for (let i = a.length - 1; i > 0; i--) {
    const j = Math.floor(Math.random() * (i + 1));
    [a[i], a[j]] = [a[j], a[i]];
  }
  return a;
};


server.get('/current', (req, res) => {
  // Start by shuffling the array
  const shuffledSounds = shuffle(sounds);

  // Create a readable stream for each sound
  const streams = shuffledSounds.map(sound => fs.createReadStream(`sounds/${sound}`));

  // Concatenate all the streams into a single stream
  const readStream = concatStreams(streams);

  // This will wait until we know the readable stream is actually valid before piping
  readStream.on('open', function () {
    // This just pipes the read stream to the response object (which goes to the client)
    // the response is automatically ended when the stream emits the "end" event
    readStream.pipe(res);
  });
});

请注意，该函数不再需要 async 关键字。该过程仍然是异步的，但编码是基于发射器而不是基于承诺的。

如果您想循环播放声音，您可以创建额外的洗牌/映射到流/连接的步骤。

为了简单起见，我没有包含 socketio 替代方案。

Answer 3

您应该使用库或查看源代码，看看它们做了什么。 一个好的是： https://github.com/obastemur/mediaserver

提示：
始终通过从其他项目中学习来开始您的研究。（如果可能或当您没有发明轮子时；））您不是第一个这样做或遇到这个问题的人 :)

用短语“nodejs stream mp3 github”快速搜索给了我一些方向.. 祝你好运！