RxJava Наблюдаемый на основе элементов в списке

Question

Мне нужно Наблюдаемое, которое никогда не заканчивается, и просто обрабатывать некоторые данные и связывать другие наблюдаемые, когда есть элементы в списке. Есть ли способ выполнить это, и какой был бы лучший подход =?

Моя самая близкая идея заключалась в том, чтобы создать наблюдаемый таймер и проверить каждые x секунд, если в списке есть элементы. Эта идея не идеальна, потому что мне нужно обрабатывать данные, как только в этом списке есть значения, которые я изменяю за пределами наблюдаемой цепочки.

return Observable.timer(2, TimeUnit.SECONDS) 
       .flatMap(integer -> captureList.getLatestCaptureCut()) 
       .flatMap(vp::processVideo) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .repeat() 

Answer 1

Я думаю, что вы можете использовать Subject и нажмите ваши следующие пункты там.

PublishSubject<Integer> subject = PublishSubject.<Integer>create(); 

subject.flatMap(integer -> captureList.getLatestCaptureCut()) 
    .flatMap(vp::processVideo) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(); 

//push new items 
subject.onNext(0); 
subject.onNext(1); 

Answer 2

Я хотел бы предложить в вашем CaptureList классе в PublishSubject. Вместо обеспечения тянуть метод getLatestCaptureCut(), вы могли бы обеспечить способ толчок с Subject:

PublishSubject<VP> captured = PublishSubject.create(); 

Вы тогда могли .subscribe() к PublishSubject и обрабатывать данные, когда они приходят в

. В вашей CaptureList вы бы назвали

captured.onNext(vp); 

каждый раз, когда новые данные. Например, в вашем setLatestCaptureCut(). Я предполагаю, что у вас уже есть какая-то рутина, которая генерирует CaptureCut и сохраняет ее, чтобы сделать ее доступной в getLatestCaptureCut().