Создание реактивной очереди или структуры данных с использованием RxJava

Question

Мне нужно создать очередь, где, когда элемент добавляется в очередь, абонент должен быть уведомлен. Единственное решение для моего мозга - использовать цикл while для опроса очереди или использовать просмотр очереди, чтобы сравнить, добавляется ли новый объект в очередь, если новый элемент найден, а затем использовать onNext() для пересылки элемента подписчику. Это был бы бесконечный цикл, и именно это мне нужно наблюдать за очередью.

Answer 1

Вы можете подклассифицировать одну из очередей и использовать методы add/offer/etc ..., которые вызывают Observable.

public class SubscriberQueue<E> extends LinkedList<E> { 

    private Observable<E> mObservable = PublishSubscriber.create(); 

    @Override 
    public boolean add(E e) { 
     if (super.add()) { 
     mObservable.onNext(e); 
     } 
    } 

    public Observable<E> getObservable() {return mObservable;} 
} 

// Test code 
SubscriberQueue<Integer> myQueue = new SubscriberQueue<>(); 
myQueue.getObservable().subscribe(i -> System.out.println("Added: " + i)); 
myQueue.add(1);