# Reactive Programming fundamentals

In this project we utilize reactive programming techniques in order to handle communication between threads while keeping the complexity low.
[Wikipedia](https://en.wikipedia.org/wiki/Reactive_programming) defines reactive programming as a programming paradigm oriented around _data flows_ and the _propagation of change_.
We use the [Reactive Extensions for C++](https://github.com/ReactiveX/RxCpp) library to achieve this functionality.
Since the documentation of the package is sparse we collect major definitions around the subset of the API we utilize in this document from different sources.

### Observable
The observable represents a push-based collection. The observable object represents the object that sends notifications (the provider) [(source)][rxjs].

### Observer
The observer provides support for push-style iteration over an observable sequence. The observer interface provides a generalized mechanism for push-based notification. The observer object represents the class that receives the notifications [(source)][rxjs].

`class rxcpp::observer<T,State,OnNext,OnError, OnCompleted>` : Consumes values from an observable using `State` that may implement `on_next`,`on_error` and `on_completed` with optional overrides of each function [(source)][rxcpp].

After an observer call the observable's `.subscribe()` method the Observable call `Observer.on_next(T)` method to provide notifications. A well-behaved observable will call an observer's `on_completed()` method exactly once or the observer's `on_error()` method exactly once  [(source)][rxjava].

### Subscriber

A subscriber is an observer that can unsubscribe itself.
Binds an observer with a `composite_subscription` that controls lifetime [(source)][rxcpp].


### Subject
Represents an observer and an observable at the same time, allowing multi-casting events from a single source to multiple child observers [(source)][rxjava].

A subject is a sort of a bridge or proxy. It acts both as an observer and an observable. As an observer it can subscribe to one or more observables. As it is also an observable it can pass through the items it observes by re-emitting them. A subject can also emit new items [(source)][reactivex].

Effectively you can expose your subject behind a method that returns an observable but internally you can use `on_next`, `on_error` etc. methods to control the sequence.

#### ReplaySubject
ReplaySubject provides the feature of caching values and then replaying them for any late subscriptions. It can be very handy for eliminating race conditions [(source)][introrx].

ReplaySubject emits to any observer all of the items that were emitted by the source observables regardless of when the observer subscribes.
There are versions of ReplaySubject that discard old items once the replay buffer increases beyond a certain size of after a given time interval has elapsed since the items were originally emitted
[(source)][reactivex].


```cpp
replay(Coordination cn, ...)
replay(std::size_t count, Coordination cn, ...)
replay(..::clock_type::period, Coordination cn, ...)
replay(std::size_t count, ...clock_type::period, Coordination cn, ...)

std::list<T> get_values()
```
In the stopped state in case of a completion or an error on the source observable, ReplaySubject still replays the cached values before sending the notification to new subsequent subscriptions. [(source)][indepth]

#### BehaviorSubject

When an observer subscribes to a BehaviorSubject, it begins by emitting the item mostly recently emitted by the source observable ( or a seed/default value if none has yet been emitted) and then continues to emit any other items emitted later by the source observable [(source)][reactivex].

API in rxcpp:
```cpp
behavior(T value,...);
T get_value();
```

### Coordination
Rxcpp uses the notion of coordination to control synchronization between threads in the observing and the emitting side of an event.
Coordinations come in three "flavors":

- `identity*` assumes that the thread pushing the value is the same with the one consuming so no thread-safety mechanisms are implemented.
- `serialize*` uses a shared mutex to serialize calls to `on_next`, `on_error`, `on_completed` and to the scheduled items.
- `observe_on*` coordinations uses a shared queue to transport work to a single destination thread.

In particular there are the following coordinations available:

- `identity_immediate()` run using the current thread immediately, no thread safety constructs are used.
- `identity_current_thread()` push into internal queue and run later using the current thread (default).
- `observe_on_new_thread()` push into the queue of a newly-created thread
- `observe_on_event_loop()` push into the queue of the event loop. The event loop uses a pool of threads to respond to events.
- `serialize_event_loop()` same as above but calls to `on_next` etc are serialized
- `serialize_new_thread()` similar to `observe_on_new_thread()` but calls to `on_next` etc. are serialized

Additionally, one can use the `rxcpp::schedulers::run_loop` class to simulate scheduling on the (main) thread. If this is used the (main) thread should execute the run loop periodically
```cpp
rxcpp::schedulers::run_loop runLoop;
//... setup subscriptions

while ((!runLoop.empty() && runLoop.peek().when < runLoop.now()))
{
    runLoop.dispatch();
}
```

The same pattern can be used by any thread assuming runLoop is created on that thread.

For a more complete explanation concerning coordinations see https://github.com/ReactiveX/RxCpp/blob/main/DeveloperManual.md

# References
- [rxjs]
- [rxcpp]
- [rxjava]
- [reactivex]
- [introrx]
- [indepth]





[rxjs]: https://rxjs.dev
[rxcpp]: https://reactivex.io/RxCpp/
[rxjava]: https://reactivex.io/RxJava/javadoc/index.html
[reactivex]: https://reactivex.io/
[introrx]: http://introtorx.com/
[indepth]: https://indepth.dev/reference/rxjs/subjects/replay-subject
