Comment créer des flux à partir d'une chaîne dans Node.Js?

J'utilise une bibliothèque, ya-csv , qui attend un fichier ou un flux en entrée, mais j'ai une chaîne.

Comment convertir cette chaîne en un flux dans Node?

À partir du nœud 10.17, stream.Readable a une fromméthode pour créer facilement des flux à partir de n'importe quel itérable (qui comprend des littéraux de tableau):

const { Readable } = require("stream")

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

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

Notez qu'au moins entre 10.17 et 12.3, une chaîne est elle-même un itérable, donc Readable.from("input string")fonctionnera, mais émettra un événement par caractère. Readable.from(["input string"])émettra un événement par élément du tableau (dans ce cas, un élément).

Notez également que dans les nœuds ultérieurs (probablement 12.3, puisque la documentation dit que la fonction a été modifiée à ce moment-là), il n'est plus nécessaire d'envelopper la chaîne dans un tableau.

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

Comme @substack m'a corrigé dans #node , la nouvelle API de flux dans Node v10 facilite les choses:

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

… Après quoi vous pouvez le canaliser librement ou le transmettre d'une autre manière à votre consommateur.

Ce n'est pas aussi propre que le one-liner de reprise , mais cela évite la dépendance supplémentaire.

( Mise à jour: de v0.10.26 à v9.2.1 jusqu'à présent, un appel à pushdirectement depuis l'invite REPL plantera avec une not implementedexception si vous ne l'avez pas défini _read. Il ne plantera pas à l'intérieur d'une fonction ou d'un script. Si une incohérence vous rend nerveux, incluez le noop.)

N'utilisez pas la réponse de reprise de Jo Liss. Cela fonctionnera dans la plupart des cas, mais dans mon cas, cela m'a fait perdre 4 ou 5 heures de recherche de bogues. Il n'y a pas besoin de modules tiers pour ce faire.

NOUVELLE RÉPONSE :

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

Il doit s'agir d'un flux lisible entièrement conforme. Voir ici pour plus d'informations sur la façon d'utiliser correctement les flux.

ANSWER : Utilisez simplement le flux PassThrough natif:

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

Notez que l'événement 'close' n'est pas émis (ce qui n'est pas requis par les interfaces de flux).

Créez simplement une nouvelle instance du streammodule et personnalisez-la en fonction de vos besoins:

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

ou

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');

Edit: La réponse de Garth est probablement meilleure.

Mon ancien texte de réponse est conservé ci-dessous.

Pour convertir une chaîne à un flux, vous pouvez utiliser un pause par flux:

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

Exemple:

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

Il existe un module pour cela: https://www.npmjs.com/package/string-to-stream

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

dans coffee-script:

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

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

utilise le:

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

Une autre solution est de passer la fonction de lecture au constructeur de Readable (cf doc stream readable options )

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

vous pouvez après utilisation sur le tuyau par exemple

J'étais fatigué de devoir réapprendre cela tous les six mois, alors je viens de publier un module npm pour résumer les détails de l'implémentation:

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

C'est le cœur du module:

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;

strest le stringqui doit être passé au constructeur lors de l'invocation, et sera produit par le flux en tant que données. optionssont les options typiques qui peuvent être passées à un flux, selon la documentation .

Selon Travis CI, il devrait être compatible avec la plupart des versions de node.

Voici une solution ordonnée dans 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...')

JavaScript est de type canard, donc si vous copiez simplement l'API d'un flux lisible , cela fonctionnera très bien. En fait, vous ne pouvez probablement pas implémenter la plupart de ces méthodes ou simplement les laisser sous forme de stubs; tout ce dont vous aurez besoin pour implémenter est ce que la bibliothèque utilise. Vous pouvez également utiliser la EventEmitterclasse pré-construite de Node pour gérer les événements, de sorte que vous n'ayez pas à implémenter addListenervous-même.

Voici comment vous pouvez l'implémenter dans 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'

Ensuite, vous pouvez l'utiliser comme ceci:

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