Creare un osservabile per un ciclo di eventi

Question

Recentemente ho letto un po 'di rx-java. Mi chiedo se la struttura si adatta al sistema di comunicazione tra i thread. Sto lavorando su un server REST scritto in java. Ogni volta che qualche risorsa è PUT/POSTed voglio fare un po 'di calcolo usando un pool di thread worker. Tuttavia, mi piacerebbe ancora poter monitorare le richieste, magari per stampare alcune statistiche. In sostanza vorrei un Observable in modo da poter gestire le richieste in modo flessibile con più Observer s.

La mia domanda è, come posso creare uno Observable adatto? La maggior parte delle guide che ho visto si occupano di operazioni su oggetti osservabili, come la mappatura, ecc. Gli oggetti visibili sono per lo più creati da collezioni o intervalli interi. In ogni caso sembra impossibile trasmettere nuovi valori agli Osservabili creati. Apparentemente l'unico modo per mantenere questa flessibilità è usare Observable.create. Tuttavia, questo sembra essere piuttosto di basso livello. Dovrei implementare un elenco di code per ogni nuovo iscritto e fare un push su synchronized a ogni singolo. È davvero necessario o qualcosa di simile è già implementato in rx-java?

Answer 1

Quello che stai cercando è un Subject. Questi agiscono sia come osservatore che come osservabili. Ad esempio un ReplaySubject riprodurrà tutti gli eventi inviati a tutti gli utenti.

Subject<String> replaySubject = ReplaySubject.create(); 
replaySubject.subscribe(s -> System.out.println(s)); 

// elsewhere... 

replaySubject.onNext("First"); 
replaySubject.onNext("Second"); 
replaySubject.onComplete();