RxAndroid Подписаться код никогда не назывался

Question

Я довольно новичок в RxJava и RxAndroid, и, хотя некоторые вещи работают, я теперь полностью потрясен тем, что, как я вижу, не работает.

У меня есть подписка слово субъекта, который никогда не кажется, чтобы бежать, и я не могу понять, почему:

public class PairManager implements DiscoveryManagerListener { 

    private Subscription wifiAvailableSubscription; 
    private Subscription debugSubscription; 
    private DiscoveryManager discoveryManager; 
    private AsyncSubject<Map<String, ConnectableDevice>> availableDevices; 

    public PairManager(Context appContext) { 
     DiscoveryManager.init(appContext); 
     discoveryManager = DiscoveryManager.getInstance(); 
     discoveryManager.addListener(this); 
     availableDevices = AsyncSubject.<Map<String, ConnectableDevice>> create(); 

     // 
     // This subscription doesn't work 
     // 
     debugSubscription = availableDevices 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<Map<String, ConnectableDevice>>() { 
      @Override 
      public void call(Map<String, ConnectableDevice> stringConnectableDeviceMap) { 
       // 
       // This code is never run ! 
       // 
       Timber.d(">> Available devices changed %s", stringConnectableDeviceMap); 
      } 
     }, new Action1<Throwable>() { 
      @Override 
      public void call(Throwable throwable) { 
       Timber.d("Subscription failed %s", throwable); 
      } 
     }); 

     availableDevices.onNext(Collections.<String, ConnectableDevice>emptyMap()); 

     wifiAvailableSubscription = ReactiveNetwork.observeNetworkConnectivity(appContext) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<Connectivity>() { 
       @Override 
       public void call(Connectivity connectivity) { 
        if (connectivity.getState().equals(NetworkInfo.State.CONNECTED) && connectivity.getType() == ConnectivityManager.TYPE_WIFI) { 
         discoveryManager.start(); 
        } else { 
         discoveryManager.stop(); 
         availableDevices.onNext(Collections.<String, ConnectableDevice>emptyMap()); 
        } 
       } 
      }); 
    } 

    public AsyncSubject<Map<String, ConnectableDevice>> getAvailableDevices() { 
     return availableDevices; 
    } 

    @Override 
    public void onDeviceAdded(DiscoveryManager manager, ConnectableDevice device) { 
     Timber.d("onDeviceAdded %s", device); 
     availableDevices.onNext(manager.getAllDevices()); 
     Timber.d("Sanity check %s", availableDevices.getValue()); 
    } 

    // ... 

} 

Есть ли способ отладки, что происходит не так? Я пробовал создавать базовые вызовы типа Observable.from и записывать их, и это работает так, как ожидалось. Журнал проверки работоспособности в onDeviceAdded также распечатывает и указывает, что availableDevices фактически обновлен, как ожидалось. Что я делаю не так?

Answer 1

Я нашел проблему, я использовал AsyncSubjects, которые только когда-либо генерируют значения, когда они завершены, где я ожидаю функциональность BehaviorSubjects.

Answer 2

От doccumentation:

Когда подключения изменяется, абонент будет уведомлен. Связь может изменить его состояние или тип.

Вы говорите:

У меня есть подписка слово субъекта

объект, не будет возвращать тэ последнее значение. Я верну только значение, когда вызывается onNext. Я предполагаю, что Связь никогда не меняется, поэтому она никогда не срабатывает.