如何在Node.Js中从字符串创建流？

Question

我正在使用一个库,ya-csv,它要求文件或流作为输入,但我有一个字符串.

如何将该字符串转换为Node中的流？

Answer 1

作为@substack在纠正我#node,新的流API在节点V10使这更容易:

const Readable = require('stream').Readable;
const s = new Readable();
s._read = () => {}; // redundant? see update below
s.push('your text here');
s.push(null);

...之后,您可以自由地管道或以其他方式将其传递给您的目标消费者.

它不像resumer one-liner 一样干净,但确实避免了额外的依赖.

(更新:在v0.10.26到v9.2.1到目前为止,如果你没有设置,push直接从REPL提示符调用会因not implemented异常_read而崩溃.它不会在函数或脚本中崩溃.如果不一致使你紧张,包括noop.)

Answer 2

不要使用Jo Liss的resumer答案.它可以在大多数情况下工作,但在我的情况下,它失去了我4或5小时的错误发现.第三方模块不需要这样做.

新答案:

var Readable = require('stream').Readable

var s = new Readable()
s.push('beep')    // the string you want
s.push(null)      // indicates end-of-file basically - the end of the stream

这应该是完全兼容的可读流.有关如何正确使用流的更多信息,请参见此处.

旧答案:只需使用本机PassThrough流:

var stream = require("stream")
var a = new stream.PassThrough()
a.write("your string")
a.end()

a.pipe(process.stdout) // piping will work as normal
/*stream.on('data', function(x) {
   // using the 'data' event works too
   console.log('data '+x)
})*/
/*setTimeout(function() {
   // you can even pipe after the scheduler has had time to do other things
   a.pipe(process.stdout) 
},100)*/

a.on('end', function() {
    console.log('ended') // the end event will be called properly
})

请注意,不会发出'close'事件(流接口不需要).

Answer 3

从节点 10.17 开始，stream.Readable 有一种from方法可以轻松地从任何可迭代对象（包括数组文字）创建流：

const { Readable } = require("stream")

const readable = Readable.from(["input string"])

readable.on("data", (chunk) => {
  console.log(chunk) // will be called once with `"input string"`
})

请注意，至少在 10.17 和 12.3 之间，字符串本身是可迭代的，因此Readable.from("input string")可以工作，但每个字符发出一个事件。Readable.from(["input string"])将为数组中的每一项（在本例中为一项）发出一个事件。

另请注意，在以后的节点中（可能是 12.3，因为文档说当时函数已更改），不再需要将字符串包装在数组中。

https://nodejs.org/api/stream.html#stream_stream_readable_from_iterable_options

Answer 4

只需创建stream模块的新实例并根据您的需要进行自定义:

var Stream = require('stream');
var stream = new Stream();

stream.pipe = function(dest) {
  dest.write('your string');
  return dest;
};

stream.pipe(process.stdout); // in this case the terminal, change to ya-csv

要么

var Stream = require('stream');
var stream = new Stream();

stream.on('data', function(data) {
  process.stdout.write(data); // change process.stdout to ya-csv
});

stream.emit('data', 'this is my string');

Answer 5

编辑: 加思的答案可能更好.

我的旧答案文本保留在下面.

要将字符串转换为流,您可以使用暂停的直通流:

through().pause().queue('your string').end()

例:

var through = require('through')

// Create a paused stream and buffer some data into it:
var stream = through().pause().queue('your string').end()

// Pass stream around:
callback(null, stream)

// Now that a consumer has attached, remember to resume the stream:
stream.resume()

Answer 6

有一个模块:https://gist.github.com/kawanet/8aea35dc4a578f09757d

var str = require('string-to-stream')
str('hi there').pipe(process.stdout) // => 'hi there' 

Answer 7

在咖啡脚本:

class StringStream extends Readable
  constructor: (@str) ->
    super()

  _read: (size) ->
    @push @str
    @push null

用它:

new StringStream('text here').pipe(stream1).pipe(stream2)

Answer 8

另一个解决方案是将read函数传递给Readable的构造函数（参见doc 流readeable选项）

var s = new Readable({read(size) {
    this.push("your string here")
    this.push(null)
  }});

您可以使用s.pipe为例

Answer 9

我厌倦了必须每六个月重新学习一次，所以我刚刚发布了一个npm模块来抽象实现细节：

https://www.npmjs.com/package/streamify-string

这是模块的核心：

const Readable = require('stream').Readable;
const util     = require('util');

function Streamify(str, options) {

  if (! (this instanceof Streamify)) {
    return new Streamify(str, options);
  }

  Readable.call(this, options);
  this.str = str;
}

util.inherits(Streamify, Readable);

Streamify.prototype._read = function (size) {

  var chunk = this.str.slice(0, size);

  if (chunk) {
    this.str = this.str.slice(size);
    this.push(chunk);
  }

  else {
    this.push(null);
  }

};

module.exports = Streamify;

str是string在调用时必须传递给构造函数的，并将由流作为数据输出。根据文档options，是可以传递到流的典型选项。

根据Travis CI，它应该与大多数版本的节点兼容。

Answer 10

这是 TypeScript 中一个简洁的解决方案：

import { Readable } from 'stream'

class ReadableString extends Readable {
    private sent = false

    constructor(
        private str: string
    ) {
        super();
    }

    _read() {
        if (!this.sent) {
            this.push(Buffer.from(this.str));
            this.sent = true
        }
        else {
            this.push(null)
        }
    }
}

const stringStream = new ReadableString('string to be streamed...')