RxJava（或 Rx.NET）等价于 ReactiveCocoa 的 RACObserve答案

【问题标题】：RxJava (or Rx.NET) equivalent of ReactiveCocoa's RACObserveRxJava（或 Rx.NET）等价于 ReactiveCocoa 的 RACObserve
【发布时间】：2014-02-07 16:37:38
【问题描述】：

给定 Java 对象上的任意字段，我想创建一个 Observable 来监视该字段，并在每次字段值更改时将新结果推送到 Observer。 ReactiveCocoa 有一个名为 RACObserve 的宏，它似乎正是这样做的。

我想知道如何使用 RxJava 实现类似的功能。

例如，假设我有以下简单的类：

public class Foo {
    enum State {
        Idle,
        Ready,
        Error
    }

    private State currentState = State.Idle;

    //methods that can change currentState
}

我想创建一个Observable<State>，它会在每次更改currentState 的值时将新状态推送到观察者。

在 ReactiveCocoa 中，看起来我会写一些类似以下的东西（请原谅我的伪 Objective-C）：

[RACObserve(self, currentState) subscribeNext:^(NSString *newState) {
    NSLog(@"%@", newState);
}];

如何在 RxJava 中实现类似的功能？我在想我可能需要将所有对 currentState 的更改包装在一个 setter 中，但我不清楚我应该在哪里调用 Observable.create 以及如何将 currentState 的更改提供给 Observer。

【问题讨论】：

标签： system.reactive reactive-programming reactive-cocoa rx-java

【解决方案1】：

ReactiveCocoa 实际上比普通的 Rx 更类似于 ReactiveUI (http://www.reactiveui.net)。而在 ReactiveUI 中，你可以使用 this.WhenAnyValue(x => x.PropName) 来做你想做的事。

【讨论】：

【解决方案2】：

我最近偶然发现了同样的问题，我最终使用了 PropertyChangeListener，它会在属性更改时发出一个对象，请参见以下内容：

更新监听器：

public class GameUpdateListener {

public static Observable<Object> changed(Game game) {
    final BehaviorSubject<Object> subject = BehaviorSubject.create((Object)game);

    game.addPropertyChangeListener(new PropertyChangeListener() {
        @Override
        public void propertyChange(PropertyChangeEvent propertyChangeEvent) {
          subject.onNext( (Object)propertyChangeEvent.getNewValue());
        }
    });
    return subject;
  }
}

一些自定义对象：

public class Game {
 private PropertyChangeSupport pcs = new PropertyChangeSupport(this);
...
  public setSomeField(String field){
       this.field = field;
       pcs.firePropertyChange("field", this.field, field);

}

public void addPropertyChangeListener(PropertyChangeListener propertyChangeListener) {
    pcs.addPropertyChangeListener(propertyChangeListener);
}

...

}

观察：

Game game = new Game();
GameUpdateListener listener = new GameUpdateListener();

final Observable<Object> gameObserver = listener.changed(game);

    gameObserver.subscribe(new Action1<Object>() {
        @Override
        public void call(Object o) {
            Log.e(TAG, "Object Changed");
        }
    });


game.setSomeField("New value");

只要您不需要再次实例化您的对象，它就可以正常工作。也许解决这个问题的方法是创建一个本地 setter 方法并在那里发出更改。

【讨论】：

【解决方案3】：

由于您的问题标题包含“或 Rx.NET”，因此这是我的建议（我不知道 RxJava，您可能会发现类似的内容）。

您可能必须利用 setter 中的某种机制。 .NET 中的标准方法是使用INotifyPropertyChanged 接口。然后通过触发事件，您可以使用以下方法从此流中创建 IObservable<T> Observable.FromEvent<TEvent, TArgs>()

你可以找到一个很好的例子来说明你想做的事情 (.NET) here。

（感谢Rob Foncesa-Ensor）

【讨论】：

【解决方案4】：

我认为你所追求的是Subject<T>。它实现了IObserver<T>，因此您可以调用OnNext(T) 来触发一个新值，也可以调用IObservable<T>，您可以将其公开以便订阅。

如果您需要它为新订阅者触发最新值，您可以使用缓冲区大小为 1 的ReplaySubject<T>。

这是一个基本的实现：

public class SomeService
{
    private Subject<int> values = new Subject<int>();

    public IObservable<T> Values
    {
        get
        {
            // AsObservable prevents it from being cast back to Subject
            return values.AsObservable();
        }
    }

    // Private; called by some internal mechanism
    private void SetValue(int newValue)
    {
        newValue.OnNext(newValue);
    }
}

【讨论】：