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

JavaScript 是鸭子类型的，所以如果你只是复制一个可读流的 API，它就会正常工作。事实上，您可能无法实现其中的大多数方法，或者只是将它们保留为存根;您需要实现的只是库使用的内容。您也可以使用 Node 预构建的 EventEmitter 类来处理事件，因此您不必自己实现此类事件。addListener

以下是如何在 CoffeeScript 中实现它：

class StringStream extends require('events').EventEmitter
  constructor: (@string) -> super()

  readable: true
  writable: false

  setEncoding: -> throw 'not implemented'
  pause: ->    # nothing to do
  resume: ->   # nothing to do
  destroy: ->  # nothing to do
  pipe: -> throw 'not implemented'

  send: ->
    @emit 'data', @string
    @emit 'end'

然后你可以像这样使用它：

stream = new StringStream someString
doSomethingWith stream
stream.send()

只需创建一个模块的新实例，并根据您的需要对其进行自定义：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');

编辑：Garth的答案可能更好。

我的旧答案文本保存在下面。

---

若要将字符串转换为流，可以使用暂停的通过流：

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()

正如@substack在 #node 中纠正我的那样，Node v10 中的新流 API 使这更容易：

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

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

它不像恢复器单行那么干净，但它确实避免了额外的依赖。

(更新：到目前为止，在 v0.10.26 到 v9.2.1 中，如果您没有设置 ，直接从 REPL 提示符调用将崩溃并出现异常。它不会在函数或脚本中崩溃。如果不一致让您感到紧张，请包括 .）pushnot implemented_readnoop

在 coffee-script 中：

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

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

使用它：

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

不要使用 Jo Liss 的简历答案。它在大多数情况下都可以工作，但在我的情况下，它让我失去了 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”事件（流接口不需要）。

有一个模块可以做到这一点：https://www.npmjs.com/package/string-to-stream

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

我厌倦了每六个月重新学习一次，所以我刚刚发布了一个 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是调用时必须传递给构造函数，并将由流作为数据输出。 是根据文档可以传递给流的典型选项。stringoptions

根据 Travis CI 的说法，它应该与大多数版本的 node 兼容。

另一种解决方案是将 read 函数传递给 Readable 的构造函数（cf doc stream readeable options)

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

您可以在使用 S.pipe 作为示例后

这是 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...')

从节点 10.17 开始，流。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

在 NodeJS 中，您可以通过以下几种方式创建可读流：

解决方案 1

你可以用模块来做。该函数允许您打开一个可读的流，您所要做的就是传递文件的路径以开始流式传输。fsfs.createReadStream()

const fs = require('fs');

const readable_stream = fs.createReadStream('file_path');

解决方案 2

如果您不想创建文件，可以创建一个内存流并对其进行某些操作（例如，将其上传到某个位置）。你可以用模块来做到这一点。您可以从模块导入，也可以创建可读的流。创建对象时，还可以实现用于从内部缓冲区中读取数据的方法。如果没有可供读取的数据，则返回。可选参数指定要读取的特定字节数。如果未指定该参数，则将返回内部缓冲区中包含的所有数据。streamReadablestreamread()nullsizesize

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

const readable_stream = new Readable({
  ​read(size) {
   ​// ...
​  }
});

解决方案 3

当您通过网络获取某些内容时，可以像流一样获取（例如，您正在从某个 API 获取 PDF 文档）。

const axios = require('axios');

const readable_stream = await axios({
  method: 'get',
  url: "pdf_resource_url",
  responseType: 'stream'
}).data;

解决方案 4

第三方软件包可以支持将流创建为一项功能。这是一种通常用于将文件上传到的包方式。aws-sdkS3

const file = await s3.getObject(params).createReadStream();