RxJava Наблюдаемой альтернативы для создания в асинхронном вызове

Question

я слушал этого разговор https://www.youtube.com/watch?v=QdmkXL7XikQ&feature=youtu.be&t=274

И ушастый, что следует избегать создания наблюдаемого с помощью метода создания, поскольку он не обрабатывает отписку и противодавление автоматически, но я могу найти альтернативу для использования в приведенном ниже коде.

compositeSubscription.add(
    Observable.create(new Observable.OnSubscribe<DTOCompaniesCallback>() { 
     @Override 
     public void call(final Subscriber<? super DTOCompaniesCallback> subscriber) { 

      modelTrainStrike.getCompaniesFromServer(new CompaniesCallback() { 
       @Override 
       public void onResult(DTOCompaniesCallback dtoCompaniesCallback) { 
        try { 
         if (!subscriber.isUnsubscribed()) { 
          subscriber.onNext(dtoCompaniesCallback); 
          subscriber.onCompleted(); 
         } 
        } catch (Exception e) { 
         if (!subscriber.isUnsubscribed()) { 
          subscriber.onError(e); 
         } 
        } 
       } 
      }); 

     } 
    }) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Action1<DTOCompaniesCallback>() { 
     @Override 
     public void call(DTOCompaniesCallback dtoCompaniesCallback) { 
      Log.i("TAG", "onResult: " + dtoCompaniesCallback.getCompaniesList().size()); 
     } 
    }, new Action1<Throwable>() { 
     @Override 
     public void call(Throwable throwable) { 
      throw new OnErrorNotImplementedException("Source!", throwable); 
     } 
    }) 
); 

И я призываю очистить CompositeSubscription в методе OnDestroy

@Override 
public void onDestroy() { 
    if (compositeSubscription != null) { 
     compositeSubscription.clear(); 
    } 
} 

ли вы видите какие-либо альтернативы методу создания, которые я мог бы использовать здесь? Вы видите какую-либо потенциальную опасность или этот подход безопасен? Благодаря

Answer 1

Вы можете использовать + Отложить AsyncSubject:

Observable.defer(() -> { 
    AsyncSubject<DTOCompaniesCallback> async = AsyncSubject.create(); 
    modelTrainStrike.getCompaniesFromServer(v -> { 
     async.onNext(v); 
     async.onComplete(); 
    }); 
    return async; 
}) 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
... 

В случае getCompaniesFromServer поддерживает отмену, вы можете:

Observable.defer(() -> { 
    AsyncSubject<DTOCompaniesCallback> async = AsyncSubject.create(); 
    Closeable c = modelTrainStrike.getCompaniesFromServer(v -> { 
     async.onNext(v); 
     async.onComplete(); 
    }); 
    return async.doOnUnsubscribe(() -> { 
     try { c.close(); } catch (IOException ex) { } 
    }); 
})