title: RxJava/Retrofit2/Java - generic NetworkBoundResource not fired
body: <p>I'm new in RxJava so my deep apology if this subject has already been covered or seems pretty easy for some of you. I've thoroughly been searching for a solution without any luck.</p>
  
  <p>Many of my API calls are are based on the original Google LiveData NetworkBoundResource class. Because of the complexity of my business logic I decided to move from LiveData to RxJava. I converted a Kotlin version <a href="https://android.jlelse.eu/networkboundresource-with-rxjava-and-kotlin-sealed-classes-1574bc516f82" rel="nofollow noreferrer">https://android.jlelse.eu/networkboundresource-with-rxjava-and-kotlin-sealed-classes-1574bc516f82</a> into a Java one and adapted the rest of the implementation accordingly. </p>
  
  <pre><code>public abstract class NetworkBoundResource&lt;ResultType, RequestType&gt; {
  
  private final String TAG = NetworkBoundResource.class.getSimpleName();
  
  private Flowable&lt;Resource&lt;ResultType&gt;&gt; result;
  
  protected NetworkBoundResource() {
  
      // Lazy db observable.
      Flowable&lt;ResultType&gt; dbObservable = Flowable.defer(() -&gt;
          loadFromDb()
              // Read from disk on Computation Scheduler
              .subscribeOn(Schedulers.computation())
      );
  
      // Lazy network observable.
      Flowable&lt;ResultType&gt; networkObservable = Flowable.defer(() -&gt;
              createCall()
              // Request API on IO Scheduler
              .subscribeOn(Schedulers.io())
              // Read/Write to disk on Computation Scheduler
              .observeOn(Schedulers.computation())
              .doOnNext (request -&gt; {
                      if (request.isSuccessful()) {
                          saveCallResult(processResponse(request));
                      } else {
                          processInternalError(request);
                      }
                  })
              .onErrorReturn (throwable -&gt; {
                  throw Exceptions.propagate(throwable);
              })
              .flatMap(__ -&gt; loadFromDb())
      );
  
      result = NetworkUtils.isNetworkStatusAvailable()
          ? networkObservable
                  .map(Resource::success)
                  .onErrorReturn (t -&gt; Resource.error(t.getMessage(), null))
                  .observeOn(AndroidSchedulers.mainThread())
                  .startWith(Resource::loading)
          : dbObservable
                  .map(Resource::success)
                  .onErrorReturn (t -&gt; Resource.error(t.getMessage(), null))
                  .observeOn(AndroidSchedulers.mainThread())
                  .startWith(Resource::loading);
  }
  
  public Flowable&lt;Resource&lt;ResultType&gt;&gt; asFlowable() {
      return result;
  }
  
  private RequestType processResponse(Response&lt;RequestType&gt; response) {
      return response.body();
  }
  
  private void processInternalError(Response&lt;RequestType&gt; response){
      String error = null;
      if (response.errorBody() != null) {
          try {
              error = response.errorBody().string();
          } catch (java.io.IOException ignored) {
              android.util.Log.e(TAG, "Error while parsing response: " + ignored.getMessage());
          }
          throw Exceptions.propagate(new Throwable(response.code() + ": " + error));
      }
  }
  
  protected abstract void saveCallResult(@NonNull RequestType item);
  
  protected abstract Flowable&lt;ResultType&gt; loadFromDb();
  
  protected abstract Flowable&lt;Response&lt;RequestType&gt;&gt; createCall();
  
  }
  </code></pre>
  
  <p>My View model which subscribes to the Steam</p>
  
  <pre><code>public abstract class ItemListViewModel&lt;T extends IListItem&gt; extends AndroidViewModel {
  
  private MutableLiveData&lt;Resource&lt;List&lt;T&gt;&gt;&gt; itemListObservable;
  private Disposable disposable;
  
  protected ItemListViewModel(Application application) {
      super(application);
      itemListObservable = new MutableLiveData&lt;&gt;();
      subscribe();
  }
  
  public LiveData&lt;Resource&lt;List&lt;T&gt;&gt;&gt; getItemListObservable() {
      return itemListObservable;
  }
  
  private void subscribe() {
      disposable = loadListItem()
              .subscribe(
                      list -&gt; itemListObservable.setValue(list),
                      (Throwable t) -&gt; itemListObservable.setValue(Resource.error(t.getMessage(), null))
              );
      DisposableManager.add(disposable);
  }
  
  @Override
  protected void onCleared() {
      disposable.dispose();
  }
  
  protected abstract void init();
  
  protected abstract Flowable&lt;Resource&lt;List&lt;T&gt;&gt;&gt; loadListItem();
  }
  </code></pre>
  
  <p>My generic fragment is handling the resource status</p>
  
  <pre><code>public abstract class ListItemFragment&lt;T extends IListItem&gt; extends Fragment {
  
  protected ItemListAdapter&lt;T&gt; adapter;
  protected RecyclerViewBinding binding;
  
  @Override
  public View onCreateView(@NonNull LayoutInflater inflater, ViewGroup container,
                           Bundle savedInstanceState) {
  
      binding = DataBindingUtil.inflate(inflater, R.layout.recycler_view, container, false);
  
      binding.recyclerView.setAdapter(adapter);
      binding.recyclerView.setHasFixedSize(true);
      binding.recyclerView.setLayoutManager(new LinearLayoutManager(getActivity()));
      binding.setIsLoading(true);
  
      return binding.getRoot();
  }
  
  protected void observeViewModel(ItemListViewModel&lt;T&gt; viewModel) {
      // Observe and update the list when the data changes
      viewModel.getItemListObservable().observe(this, (@Nullable Resource&lt;List&lt;T&gt;&gt; iObservables) -&gt; {
          if (iObservables != null) {
              processResponse(iObservables);
          }
      });
  }
  
  private void processResponse(Resource&lt;List&lt;T&gt;&gt; resource) {
      switch (resource.status) {
          case LOADING:
              renderLoadingState();
              break;
  
          case SUCCESS:
              renderDataState(resource.data);
              break;
  
          case ERROR:
              renderErrorState(resource.message);
              break;
      }
  }
  
  private void renderLoadingState(){
      binding.setIsLoading(true);
      adapter.replace(Collections.emptyList());
  }
  
  private void renderDataState(List&lt;T&gt; data){
      binding.setIsLoading(false);
      adapter.replace(data);
  }
  
  private void renderErrorState(String error){
      Toast.makeText(this.getContext(),error, Toast.LENGTH_SHORT).show();
  }
  }
  </code></pre>
  
  <p>The createCall() is never fired though I subscribed to it as explained. Can't see it in my logging interceptor as it was before</p>
  
  <p>My guess was Flowable> was not correctly recognized by Retrofit. I made sure to add the CallAdapterFactory from the squareup repo to my HttpClient</p>
  
  <pre><code>BDRepository(String URL) {
  
  GsonBuilder gsonBuilder = new GsonBuilder();
  gsonBuilder.enableComplexMapKeySerialization();
  Gson gson = gsonBuilder.create();
  
  retrofit = new Retrofit
              .Builder()
              .baseUrl(URL)
              .client(CustomTrust.getHttpClient())
              .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
              .addConverterFactory(GsonConverterFactory.create(gson))
              .build();
  }
  </code></pre>
  
  <p>My gradle conf</p>
  
  <pre><code>//API request
  implementation "com.squareup.retrofit2:retrofit:$rootProject.retrofit"
  implementation "com.squareup.retrofit2:converter-gson:$rootProject.retrofit"
  implementation "com.squareup.retrofit2:adapter-rxjava2:$rootProject.retrofit"
  </code></pre>
  
  <p>The API call returns data when trying it in a browser.
  When breaking down the code my network is always available so networkObservable is supposed to be the actual stream.</p>
  
  <p>My retrofit service</p>
  
  <pre><code>@GET("items")
  Flowable&lt;Response&lt;List&lt;Items&gt;&gt;&gt; getItemList(@Query("lang") String 
  language);
  </code></pre>
  
  <p>Any help would be much appreciated</p>