Java 类android.arch.lifecycle.LiveDataReactiveStreams 实例源码

public static <T> ObservableTransformer<T, T> applyObservable(LifecycleOwner lifecycleOwner,
                                                              BackpressureStrategy strategy) {
    return observable -> {
        LiveData<T> liveData = LiveDataReactiveStreams.fromPublisher(observable.toFlowable(strategy));
        return Flowable.fromPublisher(LiveDataReactiveStreams
                .toPublisher(lifecycleOwner, liveData)).toObservable();
    };
}

@Override
public void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    LiveData<List<Dog>> dogs = monarchy.findAllMappedWithChanges(realm -> realm.where(RealmDog.class),
                                                           from -> Dog.create(from.getName()));
    this.dogs = Flowable.fromPublisher(LiveDataReactiveStreams.toPublisher(getActivity(), dogs));
}

@Override
protected void onStart() {
    super.onStart();
    LiveDataReactiveStreams.fromPublisher(viewModel.uiModels(viewState))
            .observe(this, new UIObserver<>(this, errorMessageFactory()));
    viewModel.processEvents(events());
}

@Override
public void onStart() {
    super.onStart();
    LiveDataReactiveStreams.fromPublisher(viewModel.uiModels(viewState))
            .observe(this, new UIObserver<>(this, errorMessageFactory()));
    viewModel.processEvents(events());
}

@Override
protected void onStart() {
    super.onStart();
    mLifecycleRegistry.markState(Lifecycle.State.STARTED);
    LiveDataReactiveStreams.fromPublisher(viewModel.uiModels(viewState))
            .observe(this, new UIObserver<>(this, errorMessageFactory()));
    viewModel.processEvents(events());
}

@Override
protected void onStart() {
    super.onStart();
    LiveDataReactiveStreams.fromPublisher(viewModel.uiModels(viewState))
            .observe(this, new UIObserver<>(this, errorMessageFactory()));
    viewModel.processEvents(events());
}

@Override
public void onStart() {
    super.onStart();
    mLifecycleRegistry.markState(Lifecycle.State.STARTED);
    LiveDataReactiveStreams.fromPublisher(viewModel.uiModels(viewState))
            .observe(this, new UIObserver<>(this, errorMessageFactory()));
    viewModel.processEvents(events());
}

public static <T> ObservableTransformer<T, T> applyObservable(LifecycleOwner lifecycleOwner) {
    return observable -> {
        LiveData<T> liveData = LiveDataReactiveStreams.fromPublisher(observable
                .toFlowable(BackpressureStrategy.BUFFER));
        return Flowable.fromPublisher(LiveDataReactiveStreams
                .toPublisher(lifecycleOwner, liveData)).toObservable();
    };
}

public static <T> SingleTransformer<T, T> applySingle(LifecycleOwner lifecycleOwner) {
    return single -> {
        LiveData<T> liveData = LiveDataReactiveStreams.fromPublisher(single.toFlowable());
        return Flowable.fromPublisher(LiveDataReactiveStreams
                .toPublisher(lifecycleOwner, liveData)).singleOrError();
    };
}

public static <T> MaybeTransformer<T, T> applyMaybe(LifecycleOwner lifecycleOwner) {
    return maybe -> {
        LiveData<T> liveData = LiveDataReactiveStreams.fromPublisher(maybe.toFlowable());
        return Flowable.fromPublisher(LiveDataReactiveStreams
                .toPublisher(lifecycleOwner, liveData)).firstElement();
    };
}

public static <T> FlowableTransformer<T, T> applyFlowable(LifecycleOwner lifecycleOwner) {
    return flowable -> {
        LiveData<T> liveData = LiveDataReactiveStreams.fromPublisher(flowable);
        return Flowable.fromPublisher(LiveDataReactiveStreams.toPublisher(lifecycleOwner, liveData));
    };
}