複数の API リクエストからの複数の読み取り可能なストリームを単一の書き込み可能なストリームにパイプする方法は? 質問する

- 望ましい行動
- 実際の行動
- 私が試したこと
- 再現する手順
- 研究

望ましい行動

複数の API リクエストから受信した複数の読み取り可能なストリームを、単一の書き込み可能なストリームにパイプします。

APIレスポンスはIBM-WatsonのtextToSpeech.合成()方法。

複数のリクエストが必要な理由は、サービスに5KBテキスト入力の制限があるためです。

18KBしたがって、たとえばの文字列を完了するには 4 つのリクエストが必要になります。

実際の行動

書き込み可能なストリーム ファイルが不完全で文字化けしています。

アプリケーションがハングしているようです。

.mp3不完全なファイルをオーディオ プレーヤーで開こうとすると、破損しているというメッセージが表示されます。

ファイルを開いたり閉じたりするプロセスによってファイル サイズが増加するようです。ファイルを開くと、何らかの理由でより多くのデータが流入するようです。

望ましくない動作は、たとえば 4000 バイト以下の 4 つの文字列など、入力が大きい場合に顕著になります。

私が試したこと

npmパッケージを使用して、読み取り可能なストリームを単一の書き込み可能なストリームまたは複数の書き込み可能なストリームにパイプするいくつかの方法を試しました。複合ストリーム、複合ストリーム2、マルチストリームそしてアーカイバこれらすべての結果は不完全なファイルになります。私の最後の試みはパッケージを一切使用せず、Steps To Reproduce以下のセクションに示されています。

したがって、私はアプリケーション ロジックの各部分に疑問を抱いています。

01.Watson Text to Speech API リクエストの応答タイプは何ですか?

のテキスト読み上げドキュメントたとえば、API 応答タイプは次のようになります。

Response type: NodeJS.ReadableStream|FileObject|Buffer

応答タイプが 3 つのうちの 1 つであることがわかり、困惑しています。

これまでのすべての試みにおいて、私はそれが であると想定してきましたreadable stream。

02.マップ関数で複数の API リクエストを行うことはできますか?

03.各リクエストを 内にラップしてpromise()を解決できますかresponse?

04.結果の配列をpromises変数に割り当てることはできますか?

05.申告できますかvar audio_files = await Promise.all(promises)？

06.この宣言の後、すべての応答は「完了」しますか?

07.各応答を書き込み可能なストリームに正しくパイプするにはどうすればよいですか?

08.すべてのパイプが終了したことを検出して、ファイルをクライアントに送り返すにはどうすればよいですか?

質問 2 - 6 については、答えは「はい」であると想定しています。

私の失敗は質問 7 と 8 に関係していると思います。

再現する手順

3975このコードは、それぞれバイトサイズが、3863、バイトの3974ランダムに生成された 4 つのテキスト文字列の配列でテストできます。3629ここにその配列のペーストビンがあります。

// route handler
app.route("/api/:api_version/tts")
    .get(api_tts_get);

// route handler middleware
const api_tts_get = async (req, res) => {

    var query_parameters = req.query;

    var file_name = query_parameters.file_name;
    var text_string_array = text_string_array; // eg: https://pastebin.com/raw/JkK8ehwV

    var absolute_path = path.join(__dirname, "/src/temp_audio/", file_name);
    var relative_path = path.join("./src/temp_audio/", file_name); // path relative to server root

    // for each string in an array, send it to the watson api  
    var promises = text_string_array.map(text_string => {

        return new Promise((resolve, reject) => {

            // credentials
            var textToSpeech = new TextToSpeechV1({
                iam_apikey: iam_apikey,
                url: tts_service_url
            });

            // params  
            var synthesizeParams = {
                text: text_string,
                accept: 'audio/mp3',
                voice: 'en-US_AllisonV3Voice'
            };

            // make request  
            textToSpeech.synthesize(synthesizeParams, (err, audio) => {
                if (err) {
                    console.log("synthesize - an error occurred: ");
                    return reject(err);
                }
                resolve(audio);
            });

        });
    });

    try {
        // wait for all responses
        var audio_files = await Promise.all(promises);
        var audio_files_length = audio_files.length;

        var write_stream = fs.createWriteStream(`${relative_path}.mp3`);

        audio_files.forEach((audio, index) => {

            // if this is the last value in the array, 
            // pipe it to write_stream, 
            // when finished, the readable stream will emit 'end' 
            // then the .end() method will be called on write_stream  
            // which will trigger the 'finished' event on the write_stream    
            if (index == audio_files_length - 1) {
                audio.pipe(write_stream);
            }
            // if not the last value in the array, 
            // pipe to write_stream and leave open 
            else {
                audio.pipe(write_stream, { end: false });
            }

        });

        write_stream.on('finish', function() {

            // download the file (using absolute_path)  
            res.download(`${absolute_path}.mp3`, (err) => {
                if (err) {
                    console.log(err);
                }
                // delete the file (using relative_path)  
                fs.unlink(`${relative_path}.mp3`, (err) => {
                    if (err) {
                        console.log(err);
                    }
                });
            });

        });


    } catch (err) {
        console.log("there was an error getting tts");
        console.log(err);
    }

}

textToSpeech.synthesize(synthesizeParams)
  .then(audio => {
    audio.pipe(fs.createWriteStream('hello_world.mp3'));
  })
  .catch(err => {
    console.log('error:', err);
  });

const server = http.createServer((req, res) => {
// `req` is an http.IncomingMessage, which is a Readable Stream
// `res` is an http.ServerResponse, which is a Writable Stream

let body = '';
// get the data as utf8 strings.
// if an encoding is not set, Buffer objects will be received.
req.setEncoding('utf8');

// readable streams emit 'data' events once a listener is added
req.on('data', (chunk) => {
body += chunk;
});

// the 'end' event indicates that the entire body has been received
req.on('end', () => {
try {
const data = JSON.parse(body);
// write back something interesting to the user:
res.write(typeof data);
res.end();
} catch (er) {
// uh oh! bad json!
res.statusCode = 400;
return res.end(`error: ${er.message}`);
}
});
});

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
  writer.write(`hello, #${i}!\n`);
}
writer.end('This is the end\n');
writer.on('finish', () => {
  console.log('All writes are now complete.');
});

var ws = fs.createWriteStream('output.pdf');

fs.createReadStream('pdf-sample1.pdf').pipe(ws, { end: false });
fs.createReadStream('pdf-sample2.pdf').pipe(ws, { end: false });
fs.createReadStream('pdf-sample3.pdf').pipe(ws);

var a = fs.createReadStream('a');
var b = fs.createReadStream('b');
var c = fs.createWriteStream('c');
a.pipe(c, {end:false});
a.on('end', function() {
  b.pipe(c)
}