Tengo este código para envolver una devolución de llamada en Rx Java 1 y compila bien, pero ahora que cambié a RX Java 2 no compila... ¿cuál es el equivalente en Rx Java 2?
return Observable.fromEmitter(new Action1<AsyncEmitter<Integer>>() { @Override public void call(AsyncEmitter<Integer> emitter) { transObs.setTransferListener(new TransferListener() { @Override public void onStateChanged(int id, TransferState state) { if (state == TransferState.COMPLETED) emitter.onCompleted(); } @Override public void onProgressChanged(int id, long bytesCurrent, long bytesTotal) { } @Override public void onError(int id, Exception ex) { emitter.onError(ex); } }); emitter.setCancellation(new AsyncEmitter.Cancellable() { @Override public void cancel() throws Exception { transObs.cleanTransferListener(); } }); } }, AsyncEmitter.BackpressureMode.BUFFER);ACTUALIZAR:
Se me ocurrió esto, pero ¿tienes que lidiar con la contrapresión ya que es una llamada oncreate?
return Observable.create(new ObservableOnSubscribe<List<DigitsUser>>() { @Override public void subscribe(final ObservableEmitter<List<DigitsUser>> emitter) throws Exception { mDigitFriends.findFriends((gotEm, users) -> { emitter.onNext(users); }); emitter.setCancellable(() -> { emitter.onNext(null); }); } });Si le preocupa la contrapresión, debe usar la clase Flowable. Aquí hay una cita de RxJava2 Wiki :
Prácticamente, el fromEmitter 1.x (anteriormente fromAsync) ha sido renombrado a Flowable.create.
Aquí está su ejemplo usando la clase Flowable:
return Flowable.create(new FlowableEmitter<List<DigitsUser>>() { @Override public void subscribe(final FlowableEmitter<List<DigitsUser>> emitter) throws Exception { mDigitFriends.findFriends((gotEm, users) -> { emitter.onNext(users); }); emitter.setCancellable(() -> { emitter.onNext(null); }); } }, BackpressureStrategy.BUFFER);