Highland.js中的嵌套流操作

问题描述

我有一个来自readdirp模块的目录流.

I have a stream of directories from the readdirp module.

我要:-

• 在每个目录中使用正则表达式(例如README.*)搜索文件
• 读取不以#开头的文件的第一行
• 打印出每个目录以及该目录中README的第一行非标题行.

• search for a file using a regex (e.g. README.*) in each directory
• read the first line of that file that does not start with a #
• print out each directory and this first non-heading line of the README in the directory.

我正在尝试使用流和 highland.js .

我一直在努力处理每个目录中所有文件的流.

I am stuck trying to process a stream of all files inside each directory.

h = require 'highland'

dirStream = readdirp root: root, depth: 0, entryType: 'directories'

dirStream = h(dirStream)
  .filter (entry) -> entry.stat.isDirectory()
  .map (entry) ->

    # Search all files in the directory for README.
    fileStream = readdirp root: entry.fullPath, depth: 0, entryType: 'files', fileFilter: '!.DS_Store'
    fileStream = h(fileStream).filter (entry) -> /README\..*/.test entry.name
    fileStream.each (file) ->
      readmeStream = fs.createReadStream file
      _(readmeStream)
        .split()
        .takeUntil (line) -> not line.startsWith '#' and line isnt ''
        .last(1)
        .toArray (comment) ->
          # TODO: How do I access `comment` asynchronously to include in the return value of the map?

    return {name: entry.name, comment: comment}

推荐答案

最好将Highland流视为不可变的，并且filter和map之类的操作将返回依赖于旧流的新流，而不是对旧的流.

It's best to consider Highland streams as immutable, and operations like filter and map returning new streams that depend on the old stream, rather than modifications of the old stream.

此外，Highland方法是惰性的:仅当您现在绝对需要数据 时，才应调用each或toArray.

Also, Highland methods are lazy: you should only call each or toArray when you absolutely need the data right now.

异步映射流的标准方法是flatMap.就像map一样，但是您赋予它的函数应该返回一个流.从flatMap获得的流是所有返回的流的串联.由于新流按顺序依赖于所有旧流，因此可以用于对异步过程进行排序.

The standard way of asynchronously mapping a stream is flatMap. It's like map, but the function you give it should return a stream. The stream you get from flatMap is the concatenation of all the returned streams. Because the new stream depends on all the old streams in order, it can be used to sequence asynchronous process.

我将您的示例修改为以下内容(澄清了一些变量名称):

I'd modify your example to the following (clarified some variable names):

h = require 'highland'

readmeStream = h(readdirp root: root, depth: 0, entryType: 'directories')
  .filter (dir) -> dir.stat.isDirectory()
  .flatMap (dir) ->
    # Search all files in the directory for README.
    h(readdirp root: dir.fullPath, depth: 0, entryType: 'files', fileFilter: '!.DS_Store')
    .filter (file) -> /README\..*/.test file.name
    .flatMap (file) ->
      h(fs.createReadStream file.name)
        .split()
        .takeUntil (line) -> not line.startsWith '#' and line isnt ''
        .last(1)
        .map (comment) -> {name: file.name, comment}

我们来看一下这段代码中的类型.首先，请注意flatMap具有类型(用Haskellish表示法)Stream a → (a → Stream b) → Stream b，即，它接受包含某些类型为a的东西的流，以及一个期望包含a类型的东西并返回包含b s的流的函数. ，并返回包含b的流.集合类型(例如流和数组)的标准做法是实现flatMap来串联返回的集合.

Let's take a walk though the types in this code. First, note that flatMap has type (in Haskellish notation) Stream a → (a → Stream b) → Stream b, i.e. it takes a stream containing some things of type a, and a function expecting things of type a and returning streams containing bs, and returns a stream containing bs. It's standard for collection types (such as stream and array) to implement flatMap as concatenating the returned collections.

h(readdirp root: root, depth: 0, entryType: 'directories')

让我们说它的类型为Stream Directory. filter不会更改类型，因此flatMap将是Stream Directory → (Directory → Stream b) → Stream b.我们将看到函数返回的内容:

Let's say this has type Stream Directory. The filter doesn't change the type, so the flatMap will be Stream Directory → (Directory → Stream b) → Stream b. We'll see what the function returns:

h(readdirp root: dir.fullPath, depth: 0, entryType: 'files', fileFilter: '!.DS_Store')

将其称为Stream File，因此第二个flatMap是Stream File → (File → Stream b) → Stream b.

Call this a Stream File, so the second flatMap is Stream File → (File → Stream b) → Stream b.

h(fs.createReadStream file.name)

这是Stream String. split，takeUntil和last不会改变它，那么map会做什么? map与flatMap非常相似:其类型为Stream a → (a → b) → Stream b.在这种情况下，a是String，而b是对象类型{name : String, comment : String}.然后map返回该对象的流，这就是整个flatMap函数返回的内容.向上移动，第二个flatMap中的b是对象，因此第一个flatMap的函数还返回该对象的流，因此整个流为Stream {name : String, comment : String}.

This is a Stream String. split, takeUntil and last don't change that, so what does the map do? map is very similar to flatMap: its type is Stream a → (a → b) → Stream b. In this case a is String and b is an object type {name : String, comment : String}. Then map returns a stream of that object, which is what the overall flatMap function returns. Step up, and b in the second flatMap is the object, so the first flatMap's function also returns a stream of the object, so the entire stream is a Stream {name : String, comment : String}.

请注意，由于Highland的懒惰，实际上并没有启动任何流或处理.您需要使用each或toArray引起thunk并启动管道.在each中，回调将与您的对象一起调用.根据您想对注释进行的操作，最好再添加一些flatMap(例如，如果您要将它们写到文件中).

Note that because of Highland's laziness, this doesn't actually start any streaming or processing. You need to use each or toArray to cause a thunk and start the pipeline. In each, the callback will be called with your object. Depending on what you want to do with the comments, it might be best to flatMap some more (if you're writing them to a file for example).

好吧，我不是要写论文.希望这会有所帮助.

Well, I didn't mean to write an essay. Hope this helps.

这篇关于Highland.js中的嵌套流操作的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！