如何使用RxJS進行節點漫步？

Question

Node walk提供了一個API，其中包含一些類似的事件。

walker.on('file', (root, filestats, next) => { 
    // next should be called to goto next file 
    next(); 
}); 
walker.on('end',() => { 
    // the end of the stream 
}); 

它是反應如果從subscriber你調用一個函數來通知source去流中的下一個項目？事件不會等待用戶對它做出反應，對吧？

如何將其轉換爲Rx Observable？

Answer 1

最好的辦法是創建一個圍繞它的包裝：

Rx.Observable.fromWalk = function(root, options, scheduler) { 
    scheduler = scheduler || Rx.Scheduler.currentThread; 

    return Rx.Observable.create(function(observer) { 
    var walker = walk.walk(root, options); 

    function fileHandler(x) { 
     observer.onNext({stats : x.stats, root : x.root}); 
     scheduler.scheduleWithState(x, function(s, i) { 
     i.next(); 
     }); 
    } 

    var files = Rx.Observable.fromEvent(walker, 'file', 
     function(arr) { 
     return { root : arr[0], stats : arr[1], next : arr[2] }; 
    }); 
    var ended = Rx.Observable.fromEvent(walker, 'end'); 

    return new Rx.CompositeDisposable(
     files.subscribe(fileHandler), 
     ended.subscribe(observer.onCompleted.bind(observer)) 
    ); 
    }); 
}; 

我更新了example相應

Answer 2

嘗試Rx.Observable.fromCallback：

var walkerOn = Rx.Observable.fromCallback(walker.on, walker) 
var source = walkerOn('file'); 
var subscription = source.subscribe(observer);