标签: rx-java2

在JUnit环境中运行时,下面的代码不会崩溃.但它在应用程序中运行时崩溃了.我可以在控制台中看到错误日志,但测试标记为已通过.

  @Test
  public void test() {
    Observable observable = Observable.error(new RuntimeException());
    observable.subscribe();
  }

所以,问题是:如何在JUnit中使其崩溃.因为是的,如果某些内容在应用程序中不起作用,如果它在单元测试中也不起作用也是一件好事:)

在这个例子中,我可以直接访问observable.但在我的真实测试中,我没有那个.真正的可观察性只是被测试的类的内部细节.我能做的最多的事情就是注入调度程序或其他东西.

那么,如何在没有直接访问observable的情况下使其崩溃？

另外,我刚检查过这段代码也没有崩溃:

  @Test
  public void test() {
    Observable observable = Observable.error(new RuntimeException());
    observable.subscribe(new Consumer() {
      @Override
      public void accept(Object o) throws Exception {
        throw new RuntimeException();
      }
    }, new Consumer<Throwable>() {
      @Override
      public void accept(Throwable throwable) throws Exception {
        throw new RuntimeException();
      }
    });
  }

假设有包含方法的界面:

Observable<Data> makeHttpCall(int param1, boolean param2);

Completable storeInDatabase(Data data);

Completable combinedCall(int param1, boolean param2);

实现combineCall方法的最佳方法是:

1. 从makeHttpCall获取数据
2. 使用storeInDatabase存储它
3. return store可在storeInDatabase完成时完成吗？

似乎在RxJava 1.0中可以执行Completable.merge(Observable),但merge似乎不再接受Observable.

这是依赖项

compile 'com.squareup.retrofit2:retrofit:2.2.0'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'com.squareup.retrofit2:adapter-rxjava:2.2.0'
compile 'com.google.code.gson:gson:2.7'
compile 'com.squareup.retrofit2:converter-gson:2.0.0'

当我尝试运行应用程序时,我收到此错误

Error:Execution failed for task ':app:transformResourcesWithMergeJavaResForDebug'.

com.android.build.api.transform.TransformException:com.android.builder.packaging.DuplicateFileException:在APK META-INF/rxjava.properties File1中复制的重复文件:C:\ Users\abondarenco.gradle\caches\modules-2\files-2.1\io.reactivex.rxjava2\rxjava\2.0.7\21734c0092a5d3c3ec99510e50c1ff76bdf0c65a\rxjava-2.0.7.jar File2:C:\ Users\abondarenco.gradle\_caches\modules-2\files-2.1\io.reactivex\rxjava\1.2.0\42bfaf64c94f3848ebf5cf1c2ea4ec9d1b3ac6c8\rxjava-1.2.0.jar

我尝试运行应用程序没有adapter-rxjava它工作,但当我添加适配器它说有重复文件,它下载RxJava 1 ...

在我的应用程序中,我试图使用Retrofit 2和从我的网络服务器获取数据RxJava2.但是当我订阅我的API接口方法时,我得到错误

Cannot Resolve method 'subscribe(anonymous org.reactivestreams.Subscriber<*Model Name*>

API客户端:

public class ApiClient {
    public static final String BASE_URL = "*********";
    private static Retrofit retrofit = null;
    public static Retrofit getClient() {
        final OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .readTimeout(60, TimeUnit.SECONDS)
                .connectTimeout(60, TimeUnit.SECONDS)
                .build();
        if (retrofit==null) {
            Gson gson = new GsonBuilder()
                    .setLenient()
                    .create();
            retrofit = new Retrofit.Builder()
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .baseUrl(BASE_URL)
                    .client(okHttpClient)
                    .addConverterFactory(GsonConverterFactory.create(gson))
                    .build();
        }
        return retrofit;
    }


}

ApiInterface

public interface ApiInterface {
    @FormUrlEncoded
    @POST("user/get_user")
    Observable<UserResponse> getUsers(@Field("email") String …

我用Retrofit 2.0创建了两个不同的API请求

Observable<T> getOrganisationList()
Observable<T> getStaffList()

我正在使用CompositeDisposable包含多个其他一次性用品的容器.我希望上述两个可观察请求并行执行.

请在代码的帮助下回答问题,而不仅仅是理论.

Android Studio 3.0 Beta2

我创建了2个方法,一个创建了observable,另一个创建了订阅者.

但是,我遇到了一个问题,试图让订阅者订阅observable.在Java中,这将起作用,我试图让它在Kotlin中工作.

在我的onCreate(..)方法中,我试图设置它.这是正确的方法吗？

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        /* CANNOT SET SUBSCRIBER TO SUBCRIBE TO THE OBSERVABLE */
        createStringObservable().subscribe(createStringSubscriber())
    }


    fun createStringObservable(): Observable<String> {
        val myObservable: Observable<String> = Observable.create {
            subscriber ->
            subscriber.onNext("Hello, World!")
            subscriber.onComplete()
        }

        return myObservable
    }

    fun createStringSubscriber(): Subscriber<String> {
        val mySubscriber = object: Subscriber<String> {
            override fun onNext(s: String) {
                println(s)
            }

            override fun onComplete() {
                println("onComplete")
            }

            override fun onError(e: Throwable) {
                println("onError")
            } …

我有一个可观察的东西，它发出大量的数据，例如

[1,1,1,2,2,2,3,3,1,1,5,5 ......]

在RxJava中，我们可以使用distinctUntilChanged（）来获取一个独特的项目，直到它发生变化为止，并且它会

[1,2,3,1,5，......]

同样，有一种方法可以缓冲项目直到更改？例如，我希望输出像

[[1,1,1]，[2,2,2]，[3,3]，[1,1]，[5,5] ......]

我有一个简单的流:

Observable.error<Int>(Exception()).startWith(1).subscribe {
        println("Item is $it")
    }

一切都像预期的那样工作.首先使用整数1调用onNext,然后抛出异常,但是当我通过添加observeOn来改变流时:

Observable.error<Int>(Exception()).startWith(1).observeOn(AndroidSchedulers.mainThread()).subscribe {
        println("Item is $it")
    }

永远不会调用onNext.只抛出异常.我在这里错过了什么？

我正在学习本教程,然后添加以下行:

.retryWhen(errors -> errors.flatMap(error -> Observable.timer(30, TimeUnit.SECONDS)))

对我来说Transformer,但它会触发编译错误:

错误:不兼容的类型:无法推断类型变量R(参数不匹配; lambda表达式中的错误返回类型Observable无法转换为Publisher)其中R,T是类型变量:R extends方法flatMap中声明的Object(函数> )T扩展在类Flowable中声明的Object

而下面的波浪线error -> Observable.timer(30, TimeUnit.SECONDS)说:

没有类型变量R的实例存在,以便Observable符合Publisher

我究竟做错了什么？这个例子中缺少什么？

您好，我想问一下rxJava和Room。

这是我房间的道：

@Query("SELECT * from `Order` WHERE id = :ID")
fun findOrderById(ID: Int): Flowable<Order>

这是调用Dao的代码：

 mDB.orderDataDao().findOrderById(orderId)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe {
                    if (it.deliveryStatus == EnumOrder.IN_DELIVERY.name) {
                        mView.orderInProgress()
                        mView.setOrderFromDB(it, true)
                    } else {
                        mView.noOrderInProgress()
                        mView.setOrderFromDB(it, false)
                    }
                }

我的问题是为什么在调用Dao时会出现警告“ rxjava上未使用subscribe的结果”。为什么会发生这种情况以及我该如何做对？我在使用rxJava的每个代码中都得到了警告。