วิธีหยุดชั่วคราวและยกเลิกการหยุดสตรีมออบเจ็กต์โหนดชั่วคราวขณะประมวลผลเอาต์พุต

หากฉันเข้าใจคำถามถูกต้อง นี่คือแอป Node ง่ายๆ ที่ใช้ Dust.js ที่ช่วยแก้ปัญหาได้

Dust เป็นกลไกการสร้างเทมเพลต แต่หนึ่งในคุณสมบัติที่ดีที่สุดคือความเข้าใจดั้งเดิมของ Node Streams ตัวอย่างนี้ใช้ Dust 2.7.0

ฉันใช้ node-byline แทนสตรีม Transform ของคุณ แต่มันทำสิ่งเดียวกัน-- อ่านสตรีมทีละบรรทัด

var fs = require('fs'),
    byline = require('byline'),
    dust = require('dustjs-linkedin');

var stream = byline(fs.createReadStream('./test.txt', { encoding: 'utf8' }));

var template = dust.loadSource(dust.compile('{#byline}--> {.|s}{~n}{match}{/byline}'));

dust.stream(template, {
  byline: stream,
  match: function(chunk, context) {
    var currentLine = context.current();

    if(currentLine.match(/line\.match/g)) {
      return fs.createReadStream('./test.txt', 'utf8');
    }
    return chunk;
  }
}).pipe(process.stdout);

นี่คือผลลัพธ์จากโปรแกรมของฉัน:

$ node index.js
--> fs = require 'fs'
--> TestTransform = require './test-transform'
--> inStream = new TestTransform
--> fs.createReadStream("./test.coffee").pipe(inStream)
--> inStream.on 'line', (line) ->
-->   process.stdout.write "-->"
-->   if line.match /line\.match/g
fs = require 'fs'
TestTransform = require './test-transform'
inStream = new TestTransform
fs.createReadStream("./test.coffee").pipe(inStream)
inStream.on 'line', (line) ->
  process.stdout.write "-->"
  if line.match /line\.match/g
    process.stdout.write line
    console.error "PAUSE"
    inStream.pause()
    fs.createReadStream("./test.coffee").pipe(process.stdout).on 'end', ->
      console.error "UNPAUSE"
      inStream.resume()
  else
    process.stdout.write line

-->     process.stdout.write line
-->     console.error "PAUSE"
-->     inStream.pause()
-->     fs.createReadStream("./test.coffee").pipe(process.stdout).on 'end', ->
-->       console.error "UNPAUSE"
-->       inStream.resume()
-->   else
-->     process.stdout.write line

อย่างที่คุณเห็น มันมีการแทรกเอาต์พุตอย่างเหมาะสม หากฉันสามารถอธิบายเพิ่มเติมเกี่ยวกับวิธีการทำงานของส่วน Dust ได้ โปรดแจ้งให้เราทราบ

แก้ไข: นี่คือคำอธิบายของเทมเพลต Dust โดยเฉพาะ

{#byline} {! look for the context variable named `byline` !}
{! okay, it's a stream. For each `data` event, output this stuff once !}
-->
{.|s} {! output the current `data`. Use |s to turn off HTML escaping !}
{~n} {! a newline !}
{match} {! look up the variable called `match` !}
{! okay, it's a function. Run it and insert the result !}
{! if the result is a stream, stream it in. !}
{/byline} {! done looping !}

จริงๆ แล้วฉันก็พบคำตอบแยกต่างหากสำหรับเรื่องนี้เช่นกัน ไม่สวยเท่าแต่ก็ใช้งานได้

โดยพื้นฐานแล้ว pause() หยุดเอาต์พุตชั่วคราวจากสตรีมแบบไปป์เท่านั้น (ในโหมด "ไหล") เนื่องจากฉันกำลังฟังกิจกรรม 'line' จึงไม่ลื่นไหล และแน่นอน pause จึงไม่ทำอะไรเลย ดังนั้นวิธีแก้ปัญหาแรกคือใช้ removeListener แทน pause ซึ่งจะหยุดการสตรีมได้อย่างมีประสิทธิภาพ ตอนนี้ไฟล์ดูเหมือนว่า:

fs = require 'fs'
TestTransform = require './test-transform'
inStream = new TestTransform
fs.createReadStream("./test.coffee").pipe(inStream)
c = (line) ->
  process.stdout.write "-->"
  if line.match /line\.match/g
    process.stdout.write line
    console.error "PAUSE"
    inStream.removeListener 'line', c
    f = fs.createReadStream("./test.coffee")
    f.on 'end', ->
      console.error "UNPAUSE"
      inStream.on 'line', c
    f.pipe(process.stdout)
  else
    process.stdout.write line
inStream.on 'line', c

และสิ่งนี้สร้างผลลัพธ์ที่ เกือบ ใช้งานได้:

-->fs = require 'fs'
-->TestTransform = require './test-transform'
-->inStream = new TestTransform
-->fs.createReadStream("./test.coffee").pipe(inStream)
-->c = (line) ->
-->  process.stdout.write "-->"
-->  if line.match /line\.match/g
PAUSE
fs = require 'fs'
TestTransform = require './test-transform'
inStream = new TestTransform
fs.createReadStream("./test.coffee").pipe(inStream)
c = (line) ->
  process.stdout.write "-->"
  if line.match /line\.match/g
    process.stdout.write line
    console.error "PAUSE"
    inStream.removeListener 'line', c
    f = fs.createReadStream("./test.coffee")
    f.on 'end', ->
      console.error "UNPAUSE"
      inStream.on 'line', c
    f.pipe(process.stdout)
  else
    process.stdout.write line
inStream.on 'line', c
UNPAUSE

อย่างไรก็ตาม ดูเหมือนว่าสตรีมที่อ่านได้ดั้งเดิมจะหยุดลงเมื่อฉันลบ Listener ออก สิ่งนี้ทำให้รู้สึกบิดเบี้ยว (ฉันเดาว่าโหนดขยะจะรวบรวมสตรีมที่อ่านได้เมื่อผู้ฟังทั้งหมดถูกลบออก) ดังนั้นวิธีแก้ปัญหาสุดท้ายที่ฉันพบจึงอาศัยการวางท่อแทน เนื่องจากสตรีม Transform ที่ฉันแสดงไว้ด้านบนยังส่งเอาต์พุตทีละบรรทัดไปยังผู้ฟัง 'data' ใดๆ ก็ตาม pause() จึงสามารถนำมาใช้ที่นี่ได้อย่างมีประสิทธิภาพตามวัตถุประสงค์ดั้งเดิม โดยไม่ต้องฆ่าสตรีมเลย ผลลัพธ์สุดท้าย:

fs = require 'fs'
TestTransform = require './test-transform'
inStream = new TestTransform
fs.createReadStream("./test.coffee").pipe(inStream)
inStream.on 'data', (chunk) ->
  line = chunk.toString()
  process.stdout.write "-->#{line}"
  if line.match /line\.match/g
    inStream.pause()
    f = fs.createReadStream("./test.coffee")
    f.on 'end', ->
      inStream.resume()
    f.pipe(process.stdout)

มีเอาต์พุต:

-->fs = require 'fs'
-->TestTransform = require './test-transform'
-->inStream = new TestTransform
-->fs.createReadStream("./test.coffee").pipe(inStream)
-->inStream.on 'data', (chunk) ->
-->  line = chunk.toString()
-->  process.stdout.write "-->#{line}"
-->  if line.match /line\.match/g
fs = require 'fs'
TestTransform = require './test-transform'
inStream = new TestTransform
fs.createReadStream("./test.coffee").pipe(inStream)
inStream.on 'data', (chunk) ->
  line = chunk.toString()
  process.stdout.write "-->#{line}"
  if line.match /line\.match/g
    inStream.pause()
    f = fs.createReadStream("./test.coffee")
    f.on 'end', ->
      inStream.resume()
    f.pipe(process.stdout)
-->    inStream.pause()
-->    f = fs.createReadStream("./test.coffee")
-->    f.on 'end', ->
-->      inStream.resume()
-->    f.pipe(process.stdout)
-->

ซึ่งเป็นผลตามที่ตั้งใจไว้