RxJava2: Using Flowable with zipWith

Question

I'm trying to make the following code work me but something is wrong, here is a snippet:

private void myMethod() {
Flowable.fromIterable(cache)
.zipWith(this::doesExist, (record, exist) -> {
    // do stuff
    return true;
}).subscrib();
}

private Flowable<Boolean> doesExist(CacheRecord record) {
    // Do something
    return Flowable.just(true);
}

This doesn't compile, any idea?

UPDATE: Any thoughts about the following snippet:

Flowable.fromIterable(m_cache) // 
.flatMapCompletable(cachedStation -> { 
    return Single.zip(Single.just(cachedStation), doesIssueExist(cachedStation), (record, exist) -> { 
    System.out.println(cachedStation + ", " + exist); 
    return true; 
}).toCompletable(); 
}).subscribe();

Sarath Kn Sarath Kn · Accepted Answer · 2018-11-30T05:53:00

Your doesExist method requires a CacheRecord as a parameter. But the method reference you have given this::doesExist sends an instance of Subscriber<? super Object> that's why the incompatible type error is showing.

The expanded form of your method is given below.

  private void myMethod() {
    Flowable.fromIterable(cache)
            .zipWith(new Publisher<Object>() {
                @Override
                public void subscribe(Subscriber<? super Object> s) {
                    doesExist(s);
                }
            }, (record, exist) -> {
                // do stuff
                return true;
            }).subscribe();
 }

Here, the first parameter to zipWith

  new Publisher<Object>() {
        @Override
         public void subscribe(Subscriber<? super Object> s) {
                doesExist(s);
            }
   }

is what you have shortened as this::doesExist

As you can see the zipWith requires the first parameter a Publisher, and you have created an anonymous Publisher, and in the subscribe method you are calling doesExist(s) by sending the Subscriber<? super Object> s, which is not the required type. Your method reference statement this::doesExist does exactly the above operation and that's why the incompatible type error is shown by the compiler.

If you are trying to zip the Flowable with the flowable returned by doesExist method, you can directly call it, without method reference, by passing a valid CacheRecord object as follows

Flowable.fromIterable(cache)
        .zipWith(doesExist(anotherCache), (record, exist) -> {
            // do stuff
            return true;
        }).subscribe();

Note: See method reference for more information

Update: If you are trying to pass the items emitted by fromIterable to doesExist method and get combined result boolean and cacheRecord, then

create a holder class as follows

class CacheRecordResult {
    CacheRecord cacheRecord;
    boolean isExist;

    public CacheRecordResult(CacheRecord cacheRecord, boolean isExist) {
        this.cacheRecord = cacheRecord;
        this.isExist = isExist;
    }
   }

Then subscribe to CacheRecordResult as follows

  private void myMethod() {
    Flowable.fromIterable(cache)
          .flatMap(cacheRecord -> doesExist(cacheRecord)
                  .map(exist -> new CacheRecordResult(cacheRecord, exist)))
          .subscribe(cacheRecordResult -> {
                CacheRecord cacheRecord = cacheRecordResult.cacheRecord;
                boolean isExist = cacheRecordResult.isExist;
            });
  }

RxJava2: Using Flowable with zipWith

1 Answers