有条件地订阅rxJava2 Observable

Question

我有一个rxJava2 Observable，我想有条件地订阅它。我将进行网络调用的场景，只有在设备连接到网络时才会调用subscribe()。我想做类似下面的事情

     observable. 
     subsribeWhenConditionIsMet(

   if (connected to internet) {
      mPresentation.showNetworkError();
      return;
    }

   subscribe();
  }
)

有关如何做到这一点的任何建议？有更好的方法吗？

Answer 1

目前在RxJava2中没有这样的方法。但是如果你使用kotlin，你可以使用扩展功能。声明如下。

fun <T> Observable<T>.subscribeIf(predicate: () -> Boolean) {
    if (predicate()) {
        subscribe()
    }
}

在致电时：

anyObservable()
    .subscribeIf { isConnectedToInternet() }

额外

如果你想处理后备情况，可以像下面那样编写你的扩展，并使fallback lambda可选，这样我们可以省略它，如果不需要的话。

fun <T> Observable<T>.subscribeIf(predicate: () -> Boolean, fallback: () -> Unit = {}) {
    if (predicate()) {
        subscribe()
    } else {
        fallback()
    }
}

在致电时：

anyObservable()
    .subscribeIf(
        predicate = { isConnectedToInternet() },
        fallback = { showError() }
    )
}

注意：您也可以从Java调用它，请参阅此链接https://stackoverflow.com/a/28364983/3544839

Answer 2

拥有互联网访问不仅仅是一个简单的条件，如果你考虑它更像是一个布尔流，有时它是真的，有时它是假的。

你想要的是创建一个Observable，当互联网连接可用时，它会触发true。如果您使用的是Android，则可以在广播接收器中使用BehaviourSubject

public class NetworkChangeReceiver extends BroadcastReceiver {

    @Override
    public void onReceive(final Context context, final Intent intent) {
        final ConnectivityManager connMgr = (ConnectivityManager) context
                .getSystemService(Context.CONNECTIVITY_SERVICE);

        final android.net.NetworkInfo wifi = connMgr
                .getNetworkInfo(ConnectivityManager.TYPE_WIFI);

        final android.net.NetworkInfo mobile = connMgr
                .getNetworkInfo(ConnectivityManager.TYPE_MOBILE);

        boolean hasInternet = wifi.isAvailable() || mobile.isAvailable()
        subject.onNext(hasInternet);
    }
}

你仍然需要以某种方式将主题传递给你的广播接收器，但它不应该是一个大问题。

然后，为了仅在此主题返回true时订阅您的o​​bservable，您可以这样做：

 subject
                .filter(hasInternet ->
                        hasInternet // Don't continue if hasInternet is false
                )
                .flatMap(o ->
                        yourObservable // At this point, return the observable cause we have internet
                )
                .subscribe() // subscribe