如何包装 Node `EventEmitter` 以便能够将其作为终止的`AsyncIterator` 使用？

Question

我正在尝试使用 XML 流解析 API (xml-flow)，它公开了一个节点 EventEmitter，它为我感兴趣的标签发出一堆 tag 事件，以及一个 end 事件当它完成阅读文档时。

我希望能够使用Interactive Extensions 处理这个问题，但我不知道如何将它转换为一个结束的异步迭代； ix 只有fromEvent/fromEventPattern 似乎没有办法处理“结束”事件。

尝试一下：

import * as aix from 'ix/asynciterable';
import flow from 'xml-flow';

const iterTags = aix.fromEvent(flow(...), 'tag:foo');
console.log('max', aix.max(iterTags));

不产生任何输出，而添加 .pipe(tap(console.debug)) 来打印被迭代的值表明流实际上正在正确处理。

有没有办法可以连接end 事件以使迭代器连接到return 以便正常工作？

Answer 1

我设法通过使用rxjs 中的某些部分而不是相应的ixjs 版本来创建一个异步迭代。

无论你用 rxjs 做什么，你都可以通过管道从 ix.from 将 observable 转换为 async iterable。

输入：

  <root>
      <foo>
        <name>Bill</name>
        <id>1</id>
        <age>27</age>
      </foo>
      <foo>
        <name>Sally</name>
        <id>2</id>
        <age>40</age>
      </foo>
      <foo>
        <name>Kelly</name>
        <id>3</id>
        <age>37</age>
      </foo>
    </root>

代码：

const {fromEvent} = require('rxjs');
const {takeUntil, tap, map} = require('rxjs/operators');
const ai = require('ix/asynciterable');

const [aiFrom, aiMax] = [ai.from, ai.max];


const flow = require('xml-flow');
const fs = require('fs');
const path = require('path');

const inFile = fs.createReadStream(path.join(__dirname, 'test.xml'));


const eventEmitter = flow(inFile);
const iterTags = aiFrom(
        fromEvent(eventEmitter, 'tag:foo')
            .pipe(takeUntil(fromEvent(eventEmitter, 'end')))
            .pipe(tap(console.log), 
                  map(o=>o.age))
    );


aiMax(iterTags).then((v)=>console.log("max age:", v));

// This works also
// (async()=>{
//     for await(el of iterTags){
//         console.log(el)
//     }
// })();

输出：

{ '$name': 'foo', name: 'Bill', id: '1', age: '27' }
{ '$name': 'foo', name: 'Sally', id: '2', age: '40' }
{ '$name': 'foo', name: 'Kelly', id: '3', age: '37' }
max age: 40