Quando alguém deve usar o RxJava Observable e quando simples retorno de chamada no Android?

Estou trabalhando em rede para meu aplicativo. Então eu decidi experimentar o Retrofit da Square . Eu vejo que eles suportam simplesCallback

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

e RxJava's Observable

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

Ambos parecem bastante semelhantes à primeira vista, mas quando se trata de implementação, fica interessante ...

Embora a implementação simples de retorno de chamada seja semelhante a esta:

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

o que é bastante simples e direto. E com Observableisso rapidamente fica detalhado e 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());
}

E não é isso. Você ainda precisa fazer algo assim:

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?
            }
        });

Estou faltando alguma coisa aqui? Ou esse é um caso errado para usar Observables? Quando / deve-se preferir Observableo simples retorno de chamada?

Atualizar

Usar retrofit é muito mais simples que o exemplo acima, como o @Niels mostrou em sua resposta ou no projeto de exemplo de Jake Wharton U2020 . Mas, essencialmente, a questão permanece a mesma - quando alguém deve usar de uma maneira ou de outra?

Para coisas simples de rede, as vantagens do RxJava sobre o retorno de chamada são muito limitadas. O exemplo simples 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 
            }
     });

Ligue de volta:

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

A variante RxJava não é muito melhor que a variante de retorno de chamada. Por enquanto, vamos ignorar o tratamento de erros. Vamos tirar uma 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)
        }
    });

Ligue de volta:

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);
            }
        }
    }
});

Agora, a variante RxJava ainda não é menor, embora, com o Lambdas, ele fique mais próximo da variante de retorno de chamada. Além disso, se você tiver acesso ao feed JSON, seria meio estranho recuperar todas as fotos quando estiver exibindo apenas os PNGs. Basta ajustar o feed para exibir apenas PNGs.

Primeira conclusão

Isso não diminui sua base de código quando você está carregando um JSON simples que você preparou para estar no formato certo.

Agora, vamos tornar as coisas um pouco mais interessantes. Digamos que você não apenas deseja recuperar a foto do usuário, mas também possui um clone do Instagram e deseja recuperar 2 JSONs: 1. getUserDetails () 2. getUserPhotos ()

Você deseja carregar esses dois JSONs em paralelo e, quando ambos são carregados, a página deve ser exibida. A variante de retorno de chamada se tornará um pouco mais difícil: você precisará criar dois retornos de chamada, armazenar os dados na atividade e, se todos os dados estiverem carregados, exibir a página:

Ligue de volta:

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 chegando a algum lugar! O código do RxJava agora é tão grande quanto a opção de retorno de chamada. O código RxJava é mais robusto; Pense no que aconteceria se precisássemos carregar um terceiro JSON (como os vídeos mais recentes)? O RxJava precisaria apenas de um pequeno ajuste, enquanto a variante de retorno de chamada precisa ser ajustada em vários locais (em cada retorno de chamada, precisamos verificar se todos os dados são recuperados).

Outro exemplo; queremos criar um campo de preenchimento automático, que carrega dados usando o Retrofit. Não queremos fazer uma chamada da Web toda vez que um EditText tiver um TextChangedEvent. Ao digitar rápido, apenas o último elemento deve acionar a chamada. No RxJava, podemos usar o operador debounce:

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

Não vou criar a variante de retorno de chamada, mas você entenderá que isso é muito mais trabalhoso.

Conclusão: RxJava é excepcionalmente bom quando os dados são enviados como um fluxo. O Retrofit Observable empurra todos os elementos no fluxo ao mesmo tempo. Isso não é particularmente útil em comparação com o retorno de chamada. Mas quando existem vários elementos enviados no fluxo e em horários diferentes, e você precisa fazer coisas relacionadas ao tempo, o RxJava torna o código muito mais sustentável.

O material Observable já está pronto no Retrofit, portanto o código pode ser o seguinte:

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

No caso de getUserPhoto (), as vantagens do RxJava não são grandes. Mas vamos dar outro exemplo quando você obter todas as fotos para um usuário, mas apenas quando a imagem for PNG e você não tiver acesso ao JSON para fazer a filtragem no 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.
        }
    });

Agora, o JSON retorna uma lista de fotos. Nós os mapeamos para itens individuais. Ao fazer isso, poderemos usar o método de filtro para ignorar fotos que não são PNG. Depois disso, assinaremos e obteremos um retorno de chamada para cada foto individual, um errorHandler e um retorno de chamada quando todas as linhas forem concluídas.

TLDR Point aqui sendo; o retorno de chamada apenas retorna um retorno de sucesso e falha; o RxJava Observable permite mapear, reduzir, filtrar e muito mais.

Com o rxjava, você pode fazer mais coisas com menos código.

Vamos supor que você deseja implementar a pesquisa instantânea no seu aplicativo. Com os retornos de chamada, você se preocupou em cancelar a solicitação anterior e se inscrever na nova, lida com a mudança de orientação ... Eu acho que é muito código e muito detalhado.

Com o rxjava é muito simples.

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);
  }
}

se você deseja implementar a pesquisa instantânea, basta escutar TextChangeListener e chamar photoModel.setUserId(EditText.getText());

No método onCreate de Fragmento ou atividade que você assina no Observable que retorna photoModel.subscribeToPhoto (), ele retorna um Observable que sempre emite os itens emitidos pelo último Observable (solicitação).

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

Além disso, se o PhotoModel for um Singleton, por exemplo, você não precisa se preocupar com mudanças de orientação, porque o BehaviorSubject emite a última resposta do servidor, independentemente de quando você se inscrever.

Com essas linhas de código, implementamos uma pesquisa instantânea e lidamos com mudanças de orientação. Você acha que pode implementar isso com retornos de chamada com menos código? Eu duvido.

Geralmente seguimos a seguinte lógica:

1. Se for uma chamada de uma resposta simples, o retorno de chamada ou o futuro é melhor.
2. Se for uma chamada com várias respostas (fluxo) ou quando houver uma interação complexa entre chamadas diferentes (consulte a resposta da @Niels ), os Observáveis ​​serão melhores.

Pelas amostras e conclusões de outras respostas, acho que não há grande diferença para tarefas simples de uma ou duas etapas. No entanto, o retorno de chamada é simples e direto. RxJava é mais complicado e grande demais para tarefas simples. Existe a terceira solução: AbacusUtil . Deixe-me implementar os casos de uso acima com todas as três soluções: Callback, RxJava, CompletableFuture (AbacusUtil) com Retrolambda :

Obter foto da rede e salvar / exibir no 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.
            }
        });

Carregar detalhes do usuário e foto em 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
});

Pessoalmente, prefiro usar o Rx para obter respostas da API nos casos em que tenho que filtrar, mapear ou algo assim nos dados ou nos casos em que tenho que fazer outras chamadas da API com base nas respostas de chamadas anteriores

Parece que você está reinventando a roda, o que você está fazendo já está implementado no retrofit.

Por exemplo, você pode olhar para o RestAdapterTest.java do retrofit , onde eles definem uma interface com o Observable como tipo de retorno e depois usá-lo .

Ao criar um aplicativo para diversão, um projeto para animais de estimação ou um POC ou o 1º protótipo, você usa classes android / java básicas simples, como retornos de chamada, tarefas assíncronas, loopers, threads etc. Eles são simples de usar e não requerem nenhum Integração de lib de terceiros. Uma grande integração de bibliotecas apenas para criar um projeto imutável em um período pequeno é ilógica quando algo semelhante pode ser feito da maneira certa.

No entanto, estes são como uma faca muito afiada. Usá-los em um ambiente de produção é sempre legal, mas eles também terão consequências. É difícil escrever código simultâneo seguro se você não conhece bem os princípios de codificação limpa e SOLID. Você precisará manter uma arquitetura adequada para facilitar mudanças futuras e melhorar a produtividade da equipe.

Por outro lado, bibliotecas de simultaneidade como RxJava, Co-rotinas etc. são testadas e testadas mais de um bilhão de vezes para ajudar a escrever código simultâneo pronto para produção. Agora, novamente, não é que usando essas bibliotecas você não esteja escrevendo código simultâneo ou abstraindo toda a lógica de concorrência. Você ainda é. Mas agora, é visível e aplica um padrão claro para escrever código simultâneo em toda a base de código e, mais importante, em toda a equipe de desenvolvimento.

Esse é um grande benefício do uso de uma estrutura de simultaneidade, em vez das antigas classes básicas simples que lidam com a simultaneidade bruta. No entanto, não me entenda mal. Acredito muito na limitação das dependências da biblioteca externa, mas nesse caso específico, você precisará criar uma estrutura personalizada para sua base de código, que é uma tarefa demorada e só pode ser feita após uma experiência prévia. Portanto, as estruturas de simultaneidade são preferidas ao usar classes simples, como retornos de chamada, etc.

TL'DR

Se você já está usando o RxJava para codificação simultânea em toda a base de código, basta usar o RxJava Observable / Flowable. Uma pergunta sensata a ser feita é se devo usar Observables para Flowables. Caso contrário, vá em frente e use um indicativo.