¿Cuándo se debe usar RxJava Observable y cuándo es simple Callback en Android?

Estoy trabajando en redes para mi aplicación. Así que decidí probar Square's Retrofit . Veo que soportan simplesCallback

@GET("/user/{id}/photo")
void getUserPhoto(@Path("id") int id, Callback<Photo> cb);

y de RxJava Observable

@GET("/user/{id}/photo")
Observable<Photo> getUserPhoto(@Path("id") int id);

Ambos parecen bastante similares a primera vista, pero cuando se trata de la implementación se vuelve interesante ...

Mientras que con una simple implementación de devolución de llamada sería similar a esto:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess() {
    }
});

lo cual es bastante simple y directo. Y con Observableeso rápidamente se vuelve detallado y bastante complicado.

public Observable<Photo> getUserPhoto(final int photoId) {
    return Observable.create(new Observable.OnSubscribeFunc<Photo>() {
        @Override
        public Subscription onSubscribe(Observer<? super Photo> observer) {
            try {
                observer.onNext(api.getUserPhoto(photoId));
                observer.onCompleted();
            } catch (Exception e) {
                observer.onError(e);
            }

            return Subscriptions.empty();
        }
    }).subscribeOn(Schedulers.threadPoolForIO());
}

Y eso no es todo. Todavía tienes que hacer algo como esto:

Observable.from(photoIdArray)
        .mapMany(new Func1<String, Observable<Photo>>() {
            @Override
            public Observable<Photo> call(Integer s) {
                return getUserPhoto(s);
            }
        })
        .subscribeOn(Schedulers.threadPoolForIO())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
                //save photo?
            }
        });

¿Me estoy perdiendo de algo? ¿O es este un caso incorrecto para usar Observables? ¿Cuándo preferiría / debería preferirse Observableuna devolución de llamada simple?

Actualizar

El uso de la modificación es mucho más simple que el ejemplo anterior como lo mostró @Niels en su respuesta o en el proyecto de ejemplo U2020 de Jake Wharton . Pero, esencialmente, la pregunta sigue siendo la misma: ¿cuándo se debe usar de una forma u otra?

Para cosas de redes simples, las ventajas de RxJava sobre Callback son muy limitadas. El simple ejemplo de getUserPhoto:

RxJava:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
               // do some stuff with your photo 
            }
     });

Llamar de vuelta:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess(Photo photo, Response response) {
    }
});

La variante RxJava no es mucho mejor que la variante Callback. Por ahora, ignoremos el manejo de errores. Tomemos una lista de fotos:

RxJava:

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            list.add(photo)
        }
    });

Llamar de vuelta:

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
        List<Photo> filteredPhotos = new ArrayList<Photo>();
        for(Photo photo: photos) {
            if(photo.isPNG()) {
                filteredList.add(photo);
            }
        }
    }
});

Ahora, la variante RxJava todavía no es más pequeña, aunque con Lambdas se acercaría más a la variante Callback. Además, si tiene acceso al feed JSON, sería un poco extraño recuperar todas las fotos cuando solo se muestran los PNG. Simplemente ajuste el feed para que solo muestre PNG.

Primera conclusión

No hace que su base de código sea más pequeña cuando está cargando un JSON simple que preparó para estar en el formato correcto.

Ahora, hagamos las cosas un poco más interesantes. Supongamos que no solo desea recuperar el userPhoto, sino que tiene un clon de Instagram y desea recuperar 2 JSON: 1. getUserDetails () 2. getUserPhotos ()

Desea cargar estos dos JSON en paralelo, y cuando se cargan ambos, se debe mostrar la página. La variante de devolución de llamada se volverá un poco más difícil: debe crear 2 devoluciones de llamada, almacenar los datos en la actividad y, si todos los datos están cargados, mostrar la página:

Llamar de vuelta:

api.getUserDetails(userId, new Callback<UserDetails>() {
    @Override
    public void onSuccess(UserDetails details, Response response) {
        this.details = details;
        if(this.photos != null) {
            displayPage();
        }
    }
});

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
        this.photos = photos;
        if(this.details != null) {
            displayPage();
        }
    }
});

RxJava:

private class Combined {
    UserDetails details;
    List<Photo> photos;
}


Observable.zip(api.getUserDetails(userId), api.getUserPhotos(userId), new Func2<UserDetails, List<Photo>, Combined>() {
            @Override
            public Combined call(UserDetails details, List<Photo> photos) {
                Combined r = new Combined();
                r.details = details;
                r.photos = photos;
                return r;
            }
        }).subscribe(new Action1<Combined>() {
            @Override
            public void call(Combined combined) {
            }
        });

Estamos llegando a alguna parte! El código de RxJava ahora es tan grande como la opción de devolución de llamada. El código RxJava es más robusto; ¿Piensa en lo que sucedería si necesitáramos cargar un tercer JSON (como los últimos videos)? El RxJava solo necesitaría un pequeño ajuste, mientras que la variante de devolución de llamada debe ajustarse en varios lugares (en cada devolución de llamada debemos verificar si se recuperan todos los datos).

Otro ejemplo; queremos crear un campo de autocompletar, que carga datos usando Retrofit. No queremos hacer una llamada web cada vez que un EditText tenga un TextChangedEvent. Al escribir rápido, solo el último elemento debe activar la llamada. En RxJava podemos usar el operador antirrebote:

inputObservable.debounce(1, TimeUnit.SECONDS).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                // use Retrofit to create autocompletedata
            }
        });

No crearé la variante de devolución de llamada, pero comprenderá que esto es mucho más trabajo.

Conclusión: RxJava es excepcionalmente bueno cuando los datos se envían como una secuencia. El Retrofit Observable empuja todos los elementos en la secuencia al mismo tiempo. Esto no es particularmente útil en sí mismo en comparación con la devolución de llamada. Pero cuando hay varios elementos empujados en la transmisión y en diferentes momentos, y necesita hacer cosas relacionadas con el tiempo, RxJava hace que el código sea mucho más fácil de mantener.

El material Observable ya está hecho en Retrofit, por lo que el código podría ser este:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
         @Override
            public void call(Photo photo) {
                //save photo?
            }
     });

En el caso de getUserPhoto () las ventajas para RxJava no son geniales. Pero tomemos otro ejemplo cuando obtenga todas las fotos para un usuario, pero solo cuando la imagen sea PNG y no tenga acceso al JSON para hacer el filtrado en el lado del servidor.

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            // on main thread; callback for each photo, add them to a list or something.
            list.add(photo)
        }
    }, 
    new Action1<Throwable>() {
    @Override
        public void call(Throwable throwable) {
            // on main thread; something went wrong
            System.out.println("Error! " + throwable);
        }
    }, 
    new Action0() {
        @Override
        public void call() {
            // on main thread; all photo's loaded, time to show the list or something.
        }
    });

Ahora el JSON devuelve una lista de fotos. Los asignaremos a elementos individuales. Al hacerlo, podremos usar el método de filtro para ignorar las fotos que no son PNG. Después de eso, nos suscribiremos y recibiremos una devolución de llamada para cada foto individual, un errorHandler y una devolución de llamada cuando se hayan completado todas las filas.

Punto TLDR aquí siendo; la devolución de llamada solo le devuelve una devolución de llamada por éxito y fracaso; RxJava Observable le permite hacer mapas, reducir, filtrar y muchas cosas más.

Con rxjava puedes hacer más cosas con menos código.

Supongamos que desea implementar la búsqueda instantánea en su aplicación. Con las devoluciones de llamada, le ha preocupado darse de baja de la solicitud anterior y suscribirse a la nueva, manejar el cambio de orientación usted mismo ... Creo que es mucho código y demasiado detallado.

Con rxjava es muy simple.

public class PhotoModel{
  BehaviorSubject<Observable<Photo>> subject = BehaviorSubject.create(...);

  public void setUserId(String id){
   subject.onNext(Api.getUserPhoto(photoId));
  }

  public Observable<Photo> subscribeToPhoto(){
    return Observable.switchOnNext(subject);
  }
}

si desea implementar la búsqueda instantánea, solo tiene que escuchar TextChangeListener y llamar al photoModel.setUserId(EditText.getText());

En el método de Fragmento o actividad onCreate, se suscribe al Observable que devuelve photoModel.subscribeToPhoto (), devuelve un Observable que siempre emite los elementos emitidos por el último Observable (solicitud).

AndroidObservable.bindFragment(this, photoModel.subscribeToPhoto())
                 .subscribe(new Action1<Photo>(Photo photo){
      //Here you always receive the response of the latest query to the server.
                  });

Además, si PhotoModel es Singleton, por ejemplo, no necesita preocuparse por los cambios de orientación, porque BehaviorSubject emite la última respuesta del servidor, independientemente de cuándo se suscriba.

Con estas líneas de código hemos implementado una búsqueda instantánea y manejamos cambios de orientación. ¿Crees que puedes implementar esto con devoluciones de llamada con menos código? Lo dudo.

Generalmente vamos con la siguiente lógica:

1. Si se trata de una simple llamada de respuesta única, entonces Callback o Future es mejor.
2. Si es una llamada con múltiples respuestas (flujo), o cuando hay una interacción compleja entre diferentes llamadas (vea la respuesta de @Niels ), entonces los Observables son mejores.

Según las muestras y conclusiones en otras respuestas, creo que no hay una gran diferencia para las tareas simples de uno o dos pasos. Sin embargo, la devolución de llamada es simple y directa. RxJava es más complicado y demasiado grande para una tarea simple. Existe la tercera solución por: AbacusUtil . Permítanme implementar los casos de uso anteriores con las tres soluciones: Callback, RxJava, CompletableFuture (AbacusUtil) con Retrolambda :

Recuperar fotos de la red y guardar / mostrar en el dispositivo:

// By Callback
api.getUserPhoto(userId, new Callback<Photo>() {
    @Override
    public void onResponse(Call<Photo> call, Response<Photo> response) {
        save(response.body()); // or update view on UI thread.
    }

    @Override
    public void onFailure(Call<Photo> call, Throwable t) {
        // show error message on UI or do something else.
    }
});

// By RxJava
api.getUserPhoto2(userId) //
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(photo -> {
            save(photo); // or update view on UI thread.
        }, error -> {
            // show error message on UI or do something else.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserPhoto(userId))
        .thenRunOnUI((photo, error) -> {
            if (error != null) {
                // show error message on UI or do something else.
            } else {
                save(photo); // or update view on UI thread.
            }
        });

Cargue los detalles del usuario y la foto en paralelo

// By Callback
// ignored because it's little complicated

// By RxJava
Observable.zip(api.getUserDetails2(userId), api.getUserPhoto2(userId), (details, photo) -> Pair.of(details, photo))
        .subscribe(p -> {
            // Do your task.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserDetails(userId))
          .runOnUIAfterBoth(TPExecutor.execute(() -> api.getUserPhoto(userId)), p -> {
    // Do your task
});

Personalmente prefiero usar Rx para obtener respuestas de API en casos en los que tengo que hacer un filtro, un mapa o algo así en los datos O en casos en que tengo que hacer otras llamadas de API basadas en respuestas de llamadas anteriores

Parece que estás reinventando la rueda, lo que estás haciendo ya está implementado en la actualización.

Por ejemplo, podría mirar RestAdapterTest.java de retrofit , donde definen una interfaz con Observable como tipo de retorno, y luego usarla .

Cuando está creando una aplicación por diversión, un proyecto favorito, un POC o el primer prototipo, utiliza clases básicas de Android / Java como devoluciones de llamada, tareas asíncronas, bucles, hilos, etc. Son fáciles de usar y no requieren ninguna Integración de terceros con lib. Una gran integración de la biblioteca solo para construir un proyecto inmutable de pequeño tamaño es ilógico cuando se puede hacer algo similar desde el principio.

Sin embargo, estos son como un cuchillo muy afilado. Usarlos en un entorno de producción siempre es genial, pero también tendrán consecuencias. Es difícil escribir código concurrente seguro si no está bien versado en codificación limpia y principios SÓLIDOS. Deberá mantener una arquitectura adecuada para facilitar los cambios futuros y mejorar la productividad del equipo.

Por otro lado, las bibliotecas de concurrencia como RxJava, Co-rutinas, etc., se prueban y prueban más de mil millones de veces para ayudar a escribir código concurrente listo para producción. Ahora, de nuevo, no es que al usar estas bibliotecas no esté escribiendo código concurrente o abstrayendo toda la lógica de concurrencia. Aún lo eres. Pero ahora, es visible y aplica un patrón claro para escribir código concurrente en toda la base de código y, lo que es más importante, en todo su equipo de desarrollo.

Este es un beneficio importante de usar un marco de concurrencia en lugar de las clases básicas antiguas que tratan con la concurrencia sin procesar. Sin embargo, no me malentiendan. Creo firmemente en limitar las dependencias de la biblioteca externa, pero en este caso específico, tendrá que crear un marco personalizado para su base de código, que es una tarea que requiere mucho tiempo y que solo se puede hacer después de una experiencia avanzada. Por lo tanto, se prefieren los marcos de concurrencia sobre el uso de clases simples como devoluciones de llamada, etc.

TL'DR

Si ya está utilizando RxJava para la codificación concurrente en toda la base de código, simplemente use RxJava Observable / Flowable. Una pregunta sabia es si debo usar Observables para Flowables. Si no, entonces, adelante y usa un invocable.