Как использовать узел-прогулку с 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);