标签: rx-java2

我正在使用版本2.0.0的RxJava,似乎我无法访问AndroidSchedulers.我无法通过RxJava访问mainthread

我是新架构组件WorkManager的新手,我通过Retrofit和RxJava进行API调用.

我的用例是从后端获取新帖子,然后显示通知,并更新小部件.

所以来自Worker类的doWork()方法中的代码可能看起来像这样.

@NonNull
  @Override
  public Result doWork() {
    AppDependencies appDependencies = new AppDependencies((Application) getApplicationContext());
    Repository repository = appDependencies.getRepository();

    repository.getNewPosts()
        .flatMap(newPosts -> repository.inserPosts(newPosts).toObservable())
        .doOnError(Timber::e)
        //if success - > return  Result.SUCCESS,
        // -> show notification
        // -> update widget
        // error-> return Result.Failure
        .dontKnowWhatBestNextThing; //blocking or subscribing

    //if we reached here then Retry
    return Result.RETRY;
  }

我的问题是在Worker类中使用RxJava代码的正确方法是什么,因为doWork()方法有一个返回值,所以我必须使Rx代码同步.

如果我使用非阻塞Rx方法,我怎样才能返回值(成功 - 失败 - 重试)

我想使用rxjava在后台运行一个方法.我不关心结果.

void myHeavyMethod() { (...) }

到目前为止,我唯一的解决方案是将返回类型修改为例如boolean.

boolean myHeavyMethod() { (...) return true; }

然后我跑:

Completable.defer(() -> Completable.fromCallable(this::myHeavyMethod))
        .subscribeOn(Schedulers.computation())
        .subscribe(
                () -> {},
                throwable -> Log.e(TAG, throwable.getMessage(), throwable)
        );

有没有办法保持void返回类型？

我正在执行网络请求,我发送文件和消息.我想有一个取消当前请求的选项.我发现了两个类似的问题,并且都建议observable.subscribe(Observer)返回具有方法的Subscription对象unsubscribe().

这是第一个

第二个

在我的情况下,我使用observable.subscribe(Observer)哪个是无效的.这是我的代码:

Observable<MessengerRaw> observable = mModel.sendMessage(message, companion, description, multiParts);
        observable.subscribe(new Observer<MessengerRaw>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(MessengerRaw value) {
                if (getView() != null) {
                    ((MessengerActivity) getView()).resetMessegeView();
                    ((MessengerActivity) getView()).updateMessageList();
                }
            }

            @Override
            public void onError(Throwable e) {
                getData().remove(0);
                if (getView() != null) {
                    ((MessengerActivity) getView()).updateMessageList();
                }
            }

            @Override
            public void onComplete() {
                hideProgress();
            }
        });

那么如何取消订阅/取消我的请求？谢谢.

嘿,我正在使用Dagger2,Retrofit而且OkHttp我正面临依赖循环问题.

提供时OkHttp:

@Provides
@ApplicationScope
OkHttpClient provideOkHttpClient(TokenAuthenticator auth,Dispatcher dispatcher){
    return new OkHttpClient.Builder()
            .connectTimeout(Constants.CONNECT_TIMEOUT, TimeUnit.SECONDS)
            .readTimeout(Constants.READ_TIMEOUT,TimeUnit.SECONDS)
            .writeTimeout(Constants.WRITE_TIMEOUT,TimeUnit.SECONDS)
            .authenticator(auth)
            .dispatcher(dispatcher)
            .build();
}

提供时Retrofit:

@Provides
@ApplicationScope
Retrofit provideRetrofit(Resources resources,Gson gson, OkHttpClient okHttpClient){
    return new Retrofit.Builder()
            .baseUrl(resources.getString(R.string.base_api_url))
            .addConverterFactory(GsonConverterFactory.create(gson))
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .client(okHttpClient)
            .build();
}

提供时APIService:

@Provides
@ApplicationScope
APIService provideAPI(Retrofit retrofit) {
    return retrofit.create(APIService.class);
}

我的APIService界面:

public interface  APIService {
@FormUrlEncoded
@POST("token")
Observable<Response<UserTokenResponse>> refreshUserToken();

--- other methods like login, register ---

}

我的TokenAuthenticator …

我试图通过Jake Warthon了解这个库的目的:https: //github.com/JakeWharton/RxRelay

基本上:除了无法调用onComplete或onError之外的主题.主体以有害的方式处于有状态:当它们收到onComplete或onError时,它们不再可用于移动数据.

我知道,这是一个有效的用例,但上面似乎很容易实现只使用现有的主题.

1.不要向主题转发errors/ completions事件:

`observable.subscribe({ subject.onNext(it) }, { log error / throw exception },{ ... })`

2.不要暴露主题,使你的方法签名返回一个可观察的.

fun(): Observable<> { return subject }

我显然在这里遗漏了一些东西,我很好奇它是什么!

class MyPublishRelay<I> : Consumer<I> {

    private val subject: Subject<I> = PublishSubject.create<I>()

    override fun accept(intent: I) = subject.onNext(intent)

    fun subscribe(): Disposable = subject.subscribe()
    fun subscribe(c: Consumer<in I>): Disposable = subject.subscribe(c)
    //.. OTHER SUBSCRIBE OVERLOADS
}

PagedList<Object>用于Android的酷分页库.要使问题尽可能小:如果我有一个字符串列表,如

  List<String> stringList; // it consists of 200 strings

我想转换 stringList为PagedList<String>类似的类型

  PagedList<String> pagedStringList;

而且,如果我有一个PagedList<Object>如何将其转换为List<Object>？我经历了这个以供参考

如果我尝试相反的方式....

我怎么能转换List<Object>成DataSource.Factory<Integer, Object>..所以间接我可以把它转换成PagedList<>？

从DataSource.Factory<Integer, Object>我可以转换为PagedList<>但我怎么能转换list成PagedList<>？

使用时出现 UndeliverableExceptioncompletable

public Completable createBucketWithStorageClassAndLocation() {
        return Completable.complete()
                .doFinally(() -> {
            Bucket bucket =
                    storage.create(
                            BucketInfo.newBuilder(googleUploadObjectConfiguration.bucketName())
                                    .setStorageClass(storageClass)
                                    .setLocation(googleUploadObjectConfiguration.locationName())
                                    .build());       
        }).doOnError(error -> LOG.error(error.getMessage()));
    }

异常是从 Google 存储中抛出的，这是正确的，但尝试处理doOnError方法

Caused by: com.google.cloud.storage.StorageException: You already own this bucket. Please select another name.

RXJava 异常

io.reactivex.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | com.google.cloud.storage.StorageException: You already own this bucket. …

我像这样初始化我的变量: -

 val user: BehaviorSubject<User?> user = BehaviorSubject.create()

但我不能这样做.IDE抛出错误: -

user.onNext(null)

这样做,IDE说你永远不会为空: -

user.filter( u -> u!=null)

我不明白使用RxJava时花括号和Kotlin中的普通括号之间的真正区别.例如,我有以下代码按预期工作:

someMethodThatReturnsCompletable()
    .andThen(anotherMethodThatReturnsACompletable())
    .subscribe(...)

但以下不起作用:

someMethodThatReturnsCompletable()
    .andThen { anotherMethodThatReturnsACompletable() }
    .subscribe(...)

注意andThen()链条部分与花括号的区别.我无法理解两者之间的区别是什么.我看过一些文章,但不幸的是我仍然难以理解这种微妙的差异.