Цепь обратной связи без предмета в RX

Question

У меня есть следующие уравнения движения

move = target_position - position 
position = position + move 

, где target_position представляет собой поток и положение инициализируется в нуле. Я хотел бы иметь поток позиции. Я пытался что-то вроде этого (в приёмном псевдокоде)

moves = Subject() 
position = moves.scan(sum,0) 
target_position.combine_latest(position,diff).subscribe(moves.on_next) 

Это работает, но я читал, что с помощью Subject следует избегать. Можно ли вычислить поток позиции без объекта?

В питона полная реализация выглядит следующим образом

from pprint import pprint 
from rx.subjects import Subject 

target_position = Subject() 

moves = Subject() 

position = moves.scan(lambda x,y: x+y,0.0) 

target_position\ 
    .combine_latest(position,compute_next_move)\ 
    .filter(lambda x: abs(x)>0)\ 
    .subscribe(moves.on_next) 

position.subscribe(lambda x: pprint("position is now %s"%x)) 

moves.on_next(0.0) 
target_position.on_next(2.0) 
target_position.on_next(3.0) 
target_position.on_next(4.0) 

Answer 1

Вы можете использовать expand operator

targetPosition.combineLatest(position, (target, current) => [target, current]) 
    .expand(([target, current]) => { 
    // if you've reached your target, stop 
    if(target === current) { 
     return Observable.empty() 
    } 
    // otherwise, calculate the new position, emit it 
    // and pump it back into `expand` 
    let newPosition = calcPosition(target, current); 
    return Observable.just(newPosition) 
    }) 
    .subscribe(updateThings);