callback - qué - Rx Java Android: Cómo convertir este bloque de devolución de llamada a Observer

rxjava android retrofit

Question

Estoy intentando subir un archivo a través del S3 Android SDK de Amazon. He usado RX Java un poco, pero no estoy seguro de cómo convertir este método a un método que devuelve un Observable porque quiero encadenar el resultado de este método a otra llamada observable. Me confunde, supongo, por el hecho de que esto no vuelve de inmediato y no puede volver hasta que OnError o OnState cambien. ¿Cómo manejo estas situaciones de forma RX?

public void uploadFile(TransferObserver transferObserver){

    transferObserver.setTransferListener(new TransferListener() {
        @Override
        public void onStateChanged(int id, TransferState state) {

        }

        @Override
        public void onProgressChanged(int id, long bytesCurrent, long bytesTotal) {

        }

        @Override
        public void onError(int id, Exception ex) {

        }
  });

}

Si alguien pudiera responder con RX Java 2 y lambdas eso sería genial porque sigo quedándome corto en esta

Answer 1

En general, este es el enfoque correcto para establecer un puente entre el workd asíncrono / de devolución de llamada y el reactivo, pero ahora se desaconseja el uso de Observable.create() , ya que requiere conocimiento avanzado para hacerlo bien.
Debería usar el método de creación más reciente Observable.fromEmitter() , que se verá bastante igual:

    return Observable.fromEmitter(new Action1<Emitter<Integer>>() {
        @Override
        public void call(Emitter<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 Cancellable() {
                @Override
                public void cancel() throws Exception {
                    // Deal with unsubscription:
                    // 1. unregister the listener to avoid memory leak
                    // 2. cancel the upload 
                }
            });
        }
    }, Emitter.BackpressureMode.DROP);

Lo que se agregó aquí es: tratar con la anulación de la suscripción: cancelar la carga y anular el registro para evitar fugas de memoria y especificar la estrategia de contrapresión.
puedes leer más aquí .

Notas adicionales:

• si está interesado en el progreso, puede llamar a Siguiente () con progreso en onProgressChanged() y convertir el Observable a Observable<Integer> .
• de lo contrario, es posible que desee considerar el uso de Completable que es observable sin emisiones en onNext() , pero solo en onCompleted() esto puede onCompleted() a su caso si no está interesado con las indicaciones de progreso.

Answer 2

@Yosriz No pude obtener tu código para compilar, pero me ayudaste bastante, de modo que basándote en tu respuesta, aquí está lo que ahora tengo:

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.DROP);