Create an observable for an event loop

Question

I have been reading a little about rx-java recently. I am wondering whether the framework fits the bill for a communication system between threads. I am working on a REST server written in java. Each time some resource is PUT/POSTed I want to do some computation using a pool of worker threads. However, I would still like to be able to monitor the requests, maybe to print out some statistics. Essentially I would like an Observable so I can deal with the requests in a flexible way with multiple Observers.

My question is, how can I create a suitable Observable? Most guides I have seen deal with operations on Observables, such as mapping etc. Obervables are mostly created from collections or integer ranges. In any case it seems to be impossible to push new values to the created Observables. Apparently the only way to retain this flexibility is to use Observable.create. However, this seems to be rather low-level. I would have to implement a list of queues for each new subscriber and do a synchronized push to every single one. Is this really necessary or is something like this implemented already in rx-java?

Answer 1

What you're looking for is a Subject. These act as both Observer and Observables. For example a ReplaySubject will replay all events sent to it to all subscribers.

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

// elsewhere...

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