How to pass custom pojo as parameter to the interf

Question

Current code

    Observable.from(listMovie)//list of movie
            .flatMap(new Func1<Movie, Observable<FavMovieRes>>() {
                @Override
                public Observable<FavMovieRes> call(Movie movie) {
                    return moviesAPI.makeMovieFav(userId),
                    sessionId, new MakeMovieFav("movie", movie.getId(), movie.isFavList()));
                }
            })
            .subscribe(new Subscriber<FavMovieRes>() {
        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(FavMovieRes favMovieRes) {

        }
    });

In above code I am passing Movie object list to the Observable and perform an operation on each movie instance in list when result gets from the API I want to change some database regarding that movie instance how can I get each Movie instance in OnNext() as well as onError method of subscribe method. what I want is

Observable.from(listMovie)
                .flatMap(new Func1<Movie, Observable<FavMovieRes>>() {
                    @Override
                    public Observable<FavMovieRes> call(Movie movie) {
                        return moviesAPI.makeMovieFav(String.valueOf(SharedPreferenceDataManager.getUserId(SyncFavListPeriodicTask.this)), SharedPreferenceDataManager.getSessionId(SyncFavListPeriodicTask.this), new MakeMovieFav("movie", movie.getId(), movie.isFavList()));
                    }
                }).subscribe(new Subscriber<FavMovieRes>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e,Movie movie) {//or MakeMovieFav makeMovieFav

            }

            @Override
            public void onNext(FavMovieRes favMovieRes,Movie movie) {//or MakeMovieFav makeMovieFav

            }
        });

Answer 1

I guess you have a List and want to process each received item for another single async operation. For this case you can flatmap each result.

flatMapIterable means that it will split each item in the list as Observable to the next Operation in your Stream. flatMap means that it will do an operation on the value received.

If you want to put the results back together you can use toList after flatMap.

You need to create an Observable (Operation) for your flatmap.

Kotlin:

Observable.just(data)
        .flatMapIterable { it }
        .flatMap{ moviesAPI.makeMovieFavObservable(whatEver) }
        .subscribe( ... , ... , ... )

Java (Untested)

Observable.just(data)
            //parse each item in the list and return it as observable
            .flatMapIterable(d -> d) 
            // use each item and to another observable operation
            .flatMap(data -> Observable.just(moviesAPI.movieStuff(data)))                         
            // use each result and put it back into a list
            .toList() // put the result back to a list
            // subscribe it and log the result with Tag data, throw the error and output when completed
            .subscribe( data -> Log.d("Data", "Data received "+ data), 
                        error -> error.printStackTrace(), 
                        () -> Log.d("Completed", "Completed")
            );