如何连续执行异步功能并非只有最新的流

-1

我一直坚持一个rxjs问题几天了。我在REACT中使用rxjs，并尝试获取属性更改流以运行异步函数以快速前进/后退播放器。然后进一步等待任何函数调用，直到回调返回。如果在等待回调期间流中发生任何属性更改，只需执行最后一个属性。然后重复这个过程。如何连续执行异步功能并非只有最新的流

下面是一个示例代码如下。 http://jsbin.com/jagoworawu/edit?js,console

const setCurrentTimeStream = Observable.bindNodeCallback(player.setCurrentTime); 

// Instant execute setCurrentTimeStream on first iteration. 
// Prevent next setCurrentTimeStream if not callback of previous setCurrentTimeStream was called. 
// Skip all except last while waiting for setCurrentTimeStream callback and then execute it. 
// Then repeat the process. 
const lifecycle$ = props$ 
    .distinctUntilKeyChanged('currentTime') 
    // tried with audit and throttle but it won't execute the last one 
    .audit(({ currentTime }) => setCurrentTimeStream(currentTime)); 
    // Probably needs something more here?

编辑：阅读一些资料后，我想出了这个。请提供反馈或其他解决方案，最好使用rxjs的本地方法。

const latestExecAsync = (input, action) => Observable.create((observer) => { 
    let queued = false; 
    let latestValue; 
    const dequeue = (reference) => { 
    queued = true; 
    action(reference.value, (error, value) => { 
     if (error) return output.error(error); 
     observer.next(reference.value); 
     if (reference !== latestValue) { 
     dequeue(latestValue); 
     } else { 
     queued = false; 
     } 
    }); 
    }; 
    const subscription = input.subscribe({ 
    next: (value) => { 
     const reference = latestValue = { value }; 
     if (!queued) dequeue(reference); 
    }, 
    error: e => observer.error(e), 
    complete:() => observer.complete(), 
    }); 
    return() => subscription.unsubscribe(); 
}); 

// Usage: 
latestExecAync(
    props$.distinctUntilKeyChanged('currentTime'), 
    ({ currentTime }, cb) => player.setCurrentTime(currentTime, cb) 
)

来源

2017-02-21 Olle Bröms

我终于想出了我一直在寻找的行为。仅使用rxjs本机方法。 :)

const currentTime$ = props$.pluck('currentTime'); 
const $trailingCurrentTime = new Subject(); 

const stream$ = Observable 
    .merge(currentTime$, $trailingCurrentTime) 
    .audit(currentTime => setCurrentTime$(currentTime)) 
    .filter(currentTime => { 
    if (currentTime !== player.currentTime) { 
     $trailingCurrentTime.next(currentTime); 
    } 
    return currentTime === player.currentTime; 
    });

http://jsbin.com/wedefojuyo/edit?js,console,output

来源

2017-03-24 16:31:35

如何连续执行异步功能并非只有最新的流

回答

相关问题