RxAndroid y Retrofit: No se puede crear el adaptador de llamada para io.reactivex.Observable <retrofit2.Response <okhttp3.ResponseBody >>

Estoy intentando usar rxJava, rxAndroid, Retrofit2, y OkHTTP3 para descargar un archivo desde un punto final de URL. Mi código no puede crear el adaptador de llamada para un "Observable <retrofit2.Response <okhttp3.ResponseBody >>". Estos métodos son nuevos para mí, así que creo que estoy perdiendo un concepto importante aquí. Cualquier dirección o puntos es muy apreciada.

FATAL EXCEPTION: main Proceso: com.example.khe11e.rxdownloadfile, PID: 14130 java.lang.IllegalArgumentException: No se puede crear el adaptador de llamada para io.reactivex.Observable> para el método RetrofitInterface.downloadFileByUrlRx en retrofit2.ServiceMethod $ Builder.methodError (ServiceMethod .java: 720) en retrofit2.ServiceMethod $ Builder.createCallAdapter (ServiceMethod.java:234) en retrofit2.ServiceMethod $ Builder.build (ServiceMethod.java:160) en retrofit2.Retrofit.loadServiceMethod (Retrofit.java:166) en retrofit2 .Retrofit $ 1.invoke (Retrofit.java:145) en java.lang.reflect.Proxy.invoke (Proxy.java:393) en $ Proxy0.downloadFileByUrlRx (fuente desconocida) en com.example.khe11e.rxdownloadfile.MainActivity.downloadImage (MainActivity.java:46) en com.example.khe11e.rxdownloadfile.MainActivity $ 1.onClick (MainActivity.java:39) en android.view.View.performClick (View.java:5207) en android.view.View $ PerformClick .run (View.java:21168) en android.os.Handler.handleCallback (Handler.java:746) en android.os.Handler.dispatchMe Ssage (Handler.java:95) en android.os.Looper.loop (Looper.java:148) en android.app.ActivityThread.main (ActivityThread.java:5491) en java.lang.reflect.Method.invoke (Nativo Método) en com.android.internal.os.ZygoteInit $ MethodAndArgsCaller.run (ZygoteInit.java:728) en com.android.internal.os.ZygoteInit.main (ZygoteInit.java:618) Causado por: java.lang.IllegalArgumentException : No se pudo localizar el adaptador de llamadas para io.reactivex.Observable>. Se ha intentado: * retrofit2.adapter.rxjava.RxJavaCallAdapterFactory * retrofit2.ExecutorCallAdapterFactory en retrofit2.Retrofit.nextCallAdapter (Retrofit.java:237) en retrofit2.Retrofit.callAdapter (Retrofit.java:201) en retrofit2.ServiceMethod $ Builder.createCallAdapter (ServiceMethod .java: 232) … 16 más

• Maneje la respuesta vacía con retrofit y rxjava 2.x
• Observable.combineInferencia de tipo más reciente en kotlin
• RxJava2 función de rebote no funciona correctamente en RecyclerView - Android
• Interrumpir el subproceso JavaRx2 de actualización
• ¿Cuál es la diferencia entre RxJava 2 Cancellable y Desechable?

Build.gradle:

compile 'io.reactivex.rxjava2:rxandroid:2.0.1' compile 'io.reactivex.rxjava2:rxjava:2.0.4' compile 'com.squareup.retrofit2:retrofit:2.1.0' compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0' 

RetrofitInterface.java:

 package com.example.khe11e.rxdownloadfile; import io.reactivex.Observable; import okhttp3.ResponseBody; import retrofit2.Call; import retrofit2.Response; import retrofit2.http.GET; import retrofit2.http.Streaming; import retrofit2.http.Url; public interface RetrofitInterface { // Retrofit 2 GET request for rxjava @Streaming @GET Observable<Response<ResponseBody>> downloadFileByUrlRx(@Url String fileUrl); } 

MainActivity.java:

 package com.example.khe11e.rxdownloadfile; import android.support.v7.app.AppCompatActivity; import android.os.Bundle; import android.util.Log; import android.view.View; import android.widget.Button; import java.io.File; import io.reactivex.Observable; import io.reactivex.ObservableEmitter; import io.reactivex.ObservableOnSubscribe; import io.reactivex.Observer; import io.reactivex.android.schedulers.AndroidSchedulers; import io.reactivex.disposables.Disposable; import io.reactivex.functions.Function; import io.reactivex.schedulers.Schedulers; import okhttp3.OkHttpClient; import okhttp3.ResponseBody; import okio.BufferedSink; import okio.Okio; import retrofit2.Response; import retrofit2.Retrofit; import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory; public class MainActivity extends AppCompatActivity { Button downloadImgBtn; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); downloadImgBtn = (Button) findViewById(R.id.downloadImgBtn); downloadImgBtn.setOnClickListener(new View.OnClickListener() { @Override public void onClick(View v) { downloadImage(); } }); } public void downloadImage(){ RetrofitInterface downloadService = createService(RetrofitInterface.class, "https://www.nasa.gov/"); downloadService.downloadFileByUrlRx("sites/default/files/iss_1.jpg") .flatMap(processResponse()) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(handleResult()); } public <T> T createService(Class<T> serviceClass, String baseUrl){ Retrofit retrofit = new Retrofit.Builder() .baseUrl(baseUrl) .client(new OkHttpClient.Builder().build()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()).build(); return retrofit.create(serviceClass); } public Function<Response<ResponseBody>, Observable<File>> processResponse(){ return new Function<Response<ResponseBody>, Observable<File>>() { @Override public Observable<File> apply(Response<ResponseBody> responseBodyResponse) throws Exception { return saveToDiskRx(responseBodyResponse); } }; } private Observable<File> saveToDiskRx(final Response<ResponseBody> response){ return Observable.create(new ObservableOnSubscribe<File>() { @Override public void subscribe(ObservableEmitter<File> subscriber) throws Exception { String header = response.headers().get("Content-Disposition"); String filename = header.replace("attachment; filename=", ""); new File("/data/data/" + getPackageName() + "/images").mkdir(); File destinationFile = new File("/data/data/" + getPackageName() + "/images/" + filename); BufferedSink bufferedSink = Okio.buffer(Okio.sink(destinationFile)); bufferedSink.writeAll(response.body().source()); bufferedSink.close(); subscriber.onNext(destinationFile); subscriber.onComplete(); } }); } private Observer<File> handleResult(){ return new Observer<File>() { @Override public void onSubscribe(Disposable d) { Log.d("OnSubscribe", "OnSubscribe"); } @Override public void onNext(File file) { Log.d("OnNext", "File downloaded to " + file.getAbsolutePath()); } @Override public void onError(Throwable e) { e.printStackTrace(); Log.d("Error", "Error " + e.getMessage()); } @Override public void onComplete() { Log.d("OnComplete", "onCompleted"); } }; } } 

He intentado agregar Call como se mencionó aquí por lo que parece:

 Call<Observable<Response<ResponseBody>>> downloadFileByUrlRx(@Url String fileUrl); 

Sin embargo, esto causa problemas con la función flatMap, ya que no puede encontrar el método de símbolo flatMap (Function <Response <ResponseBody>, Observable <File >>).

 //compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0' compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0' 

 compile 'com.squareup.retrofit2:retrofit:2.2.0' compile 'com.squareup.retrofit2:adapter-rxjava2:2.2.0' 

 compile 'io.reactivex.rxjava2:rxandroid:2.0.1' compile 'io.reactivex.rxjava2:rxjava:2.1.0' compile 'com.squareup.retrofit2:retrofit:2.3.0' compile 'com.squareup.retrofit2:adapter-rxjava2:2.3.0' 

 compile 'com.squareup.retrofit2:adapter-rxjava2:2.3.0' 

 compile 'com.squareup.retrofit2:adapter-rxjava:2.3.0'