Java技巧分享之利用RxJava打造可观测数据RxLiveData

class User {
    //...... Java Bean 的字段略
}
interface Listener {
    void onUserUpdated(User user);
}
class UserManager {
    private static UserManager manager = new UserManager();
    private UserManager() {
    }
    public static UserManager getInstance() {
        return manager;
    }
    private User user;
    private List<Listener> listeners = new LinkedList<>();
    public void addUserListener(Listener listener) {
        listeners.add(listener);
    }
    public void removeUserListener(Listener listener) {
        listeners.remove(listener);
    }
    public User getUser() {
        return user;
    }
    public void setUser(User user) {
        this.user = user;
        for (Listener listener : listeners) {
            listener.onUserUpdated(this.user);
        }
    }
}

/* 测试用的 Java Bean 数据类*/
class User {
    //...... Java Bean 的字段略
}
/* 一个单例 */
class UserManager {
    private static final UserManager manager = new UserManager();
    private UserManager() {
    }
    public static UserManager getInstance() {
        return manager;
    }
    private final RxLiveData<User> userData = new RxLiveData<>();
    public RxLiveData<User> getUserData() {
        return userData;
    }
}
class A {
    public void init() {
        //订阅可观测对象，使得数据发生改变时可以被回调
        UserManager.getInstance().getUserData().getObservable()
                .subscribe((User user) -> {//使用lambda版
                    update(user);// 每次用户信息改变，这里会被调用
                });
        update(UserManager.getInstance().getUserData().getValue());
    }
    private void update(User user) {
        System.out.println("user changed");
    }
}
class B{
    public B() {
        UserManager.getInstance().getUserData().getObservable().subscribe(this::update);//方法引用版
    }
    private void update(User user) {
        System.out.println("user changed");
    }
}
public class Main {
    public static void main(String[] args) throws InterruptedException {
        A a = new A();
        a.init();
        B b = new B();
        //更新UserManager中的数据，这时候A和B中的对应方法会被调用
        UserManager.getInstance().getUserData().postData(new User());
    }
}

private final RxLiveData<Long> timeData = new RxLiveData<>();
public RxLiveData<Long> getTimeData() {
    return timeData;
}

userManager.getUserData().getObservable()
        .compose(bindUntilEvent(ActivityEvent.DESTROY))//指定在onDestroy回调时取消订阅
        .observeOn(AndroidSchedulers.mainThread())//指定主线程
        .subscribe(user -> {
        }, Throwable::printStackTrace);

import io.reactivex.rxjava3.core.Observable;//如果用的是RxJava2的请改为该版本的包名
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.disposables.Disposable;
import java.util.Optional;
public class RxLiveData<T> {
    private final Observable<T> observable;
    private Disposable disposable;
    private T value;
    private ObservableEmitter<T> emitter;
    public RxLiveData() {
        observable = Observable
                .create((ObservableEmitter<T> emitter) -> this.emitter = emitter)
                .publish()
                .autoConnect(0, disposable -> this.disposable = disposable);
    }
    public Observable<T> getObservable() {
        return observable;
    }
    public void postData(T value) {
        this.value = value;
        if (emitter != null && value != null) {
            emitter.onNext(value);
        }
    }
    public T getValue() {
        return value;
    }
    public Optional<T> optValue() {
        return Optional.ofNullable(value);
    }
}