RxJava Observable doesn't emit item

Question

I am writing a client app for the kaltura video platform, and now, I am developing a playlist built from a video list. You can drag video from one list and drop to another.

First of all, I want to get all videos (without filter) and put it to

List<KalturaVideo> sourceList;

variable. When you added video to

List<KalturaVideo> targetList;

I want to remove this item from sourceList. So, I have a class KalturaVideoRetriver, which has one public method:

public static Observable<KalturaVideo> getVideoList(Context context, String kalturaPlaylistId)

It returns Observable<KalturaVideo>, and I want to get KalturaVideo items from this observable.

This is the code of KalturaVideoRetriever class:

public class KalturaVideoRetriever {

    public static final String KALTURA_NEW_PLAYLIST_ID = "NEW_PLAYLIST";

    public static Observable<KalturaVideo> getVideoList(Context context, String kalturaPlaylistId){
        return Observable.create(new Observable.OnSubscribe<KalturaVideo>() {
            @Override
            public void call(Subscriber<? super KalturaVideo> subscriber) {
                getKalturaPlaylistContentObservable(kalturaPlaylistId, context)
                        .flatMap(new Func1<String[], Observable<?>>() {
                            @Override
                            public Observable<?> call(String[] videoIdList) {
                                return getKalturaVideoListObservable(context, videoIdList);
                            }
                        });
            }
        });
    }

    private static Observable<KalturaVideo> getKalturaVideoListObservable(Context context, String[] kalturaVideoIdArray){
        return Observable.create(new Observable.OnSubscribe<KalturaVideo>() {
            @Override
            public void call(Subscriber<? super KalturaVideo> subscriber) {
                Cursor query;
                List<KalturaVideo> result = new ArrayList<>();
                if(kalturaVideoIdArray == null || kalturaVideoIdArray.length == 0) {
                    query = context.getContentResolver().query(KalturaVideoColumns.CONTENT_URI, null, null, null, null);
                }else{
                    KalturaVideoSelection where = new KalturaVideoSelection();
                    where.kalturaIdLike(kalturaVideoIdArray);
                    query = context.getContentResolver().query(
                            KalturaVideoColumns.CONTENT_URI,
                            null,
                            where.sel(),
                            where.args(),
                            null);
                    if(!query.moveToFirst()){
                        subscriber.onCompleted();
                    }
                    KalturaVideoCursor cursor = new KalturaVideoCursor(query);
                    do{
                        KalturaVideo video = new KalturaVideo();
                        video.setId(cursor.getKalturaId());
                        video.setName(cursor.getName());
                        video.setDescription(cursor.getDescription());
                        video.setCategories(cursor.getCategories());
                        video.setCategoriesIds(cursor.getCategoriesIds());
                        video.setDownloadUrl(cursor.getDownloadUrl());
                        video.setThumbnailUrl(cursor.getThumbnailUrl());
                        video.setDataUrl(cursor.getDataUrl());
                        video.setDuration(cursor.getDuration());
                        subscriber.onNext(video);
                    }while (query.moveToNext());
                    subscriber.onCompleted();
                }
            }
        });
    }

    private static Observable<String[]> getKalturaPlaylistContentObservable(String kalturaPlaylistId, Context context){
        return Observable.create(new Observable.OnSubscribe<String[]>() {
            @Override
            public void call(Subscriber<? super String[]> subscriber) {
                KalturaPlaylistContentSelection where = new KalturaPlaylistContentSelection();
                where.playlistId(kalturaPlaylistId);
                Cursor query = context.getContentResolver().query(
                        KalturaPlaylistContentColumns.CONTENT_URI,
                        null,
                        where.sel(),
                        where.args(),
                        null
                );
                if(!query.moveToFirst()){
                    subscriber.onNext(new String[]{});
                    subscriber.onCompleted();
                }
                KalturaPlaylistContentCursor cursor = new KalturaPlaylistContentCursor(query);
                String[] result = new String[query.getCount()];
                int index = 0;
                do{
                    result[index] = cursor.getKalturaVideoId();
                    index++;
                }while (query.moveToNext());
                subscriber.onNext(result);
                subscriber.onCompleted();
            }
        });

    }

}

getKalturaPlaylistContentObservable - gets me an array of KalturaVideo id's by kalturaPlaylistId. If kalturaPlaylistId is KALTURA_NEW_PLAYLIST_ID, then it gets me an empty array, which means, that I want to get all videos in my SQLite db.

getKalturaVideoListObservable - gets me KalturaVideo items by array of video id from getKalturaPlaylistContentObservable. If the array is empty, then it gets me all videos, which I have.

Also, I have an Activity: BuildPlaylistStep2Activity. In this activity I want to display videos if sourceList, and targetList. Here is the code of activity:

public class BuildPlaylistStep2Activity extends BaseActivity implements View.OnTouchListener {

    public ViewHolder viewHolder;
    private View selected_item = null;
    Boolean touchFlag = false;
    boolean dropFlag = false;
    LayoutParams viewParams;

    List<KalturaVideo> sourceList;
    List<KalturaVideo> targetList;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_build_playlist_step2);
        viewHolder = new ViewHolder();
    }

    @Override
    protected void onPostCreate(Bundle savedInstanceState) {
        super.onPostCreate(savedInstanceState);
        getSourceListObservable(KalturaVideoRetriver.KALTURA_NEW_PLAYLIST_ID)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(kalturaVideo -> {
                    sourceList.add(kalturaVideo);
                })
                .doOnCompleted(() -> {
                    viewHolder.sourceListView.setAdapter(new BuildPlaylistContentAdapter(
                            BuildPlaylistStep2Activity.this,
                            -1,
                            sourceList));
                });

    }

    @Override
    public boolean onTouch(View v, MotionEvent motionEvent) {
        switch (motionEvent.getActionMasked()) {
            case MotionEvent.ACTION_DOWN:
                touchFlag = true;
                selected_item = v;
                viewParams = v.getLayoutParams();
                break;
            case MotionEvent.ACTION_UP:
                selected_item = null;
                touchFlag = false;
                break;
            default:
                break;
        }
        return false;
    }

    private Observable<KalturaVideo> getSourceListObservable(String kalturaPlaylistId){
        Observable<String[]> targetIdList = Observable.just(targetList)
                .map((List<KalturaVideo> kalturaVideos) -> {
                    if(targetList == null){
                        return new String[]{};
                    }
                    String[] result = new String[kalturaVideos.size()];
                    int index = 0;
                    for (KalturaVideo item : kalturaVideos) {
                        result[index] = item.getId();
                    }
                    return result;
                });
        return Observable.zip(
                targetIdList,
                KalturaVideoRetriver.getVideoList(BuildPlaylistStep2Activity.this, kalturaPlaylistId),
                (String[] idListOfTarget, KalturaVideo kalturaVideo) -> {
                    for (String item :idListOfTarget){
                        if(item.equals(kalturaVideo.getId())){
                            return null;
                        }
                    }
                    return kalturaVideo;
                })
                .filter(kalturaVideo -> {
                    return kalturaVideo != null;
                });
    }

    public class ViewHolder{
        ListView sourceListView;
        ListView targetListView;

        public ViewHolder() {
            this.sourceListView = (ListView)findViewById(R.id.source_listview);
            this.targetListView = (ListView)findViewById(R.id.target_listview);
        }
    }
}

getSourceListObservable - just removes from videos, which gets me KalturaVideoRetriever.getVideoList, videos which are contained in targetList

As you can see from code, I tried to subscribe on Observable in onPostCreate method and display data on listview, but nothing happens, and I don't know why...

Answer 1

You're calling doOnCompleted and doOnNext methods which are not subscriptions and should be used for side effects - like f.e. logging. To start your observable you need to subscribe via subscribe method.