ReactivePlusPlus (RPP)

ReactivePlusPlus (RPP) is a reactive programming library for C++20 designed for building asynchronous, event-driven data streams. It uses a declarative approach where streams are constructed by composing primitive operators using the pipe (|) operator.

Key characteristics:

• C++20 Native: Built using modern C++20 features, specifically concepts, which provide clearer error messages and better type safety during compilation.
• Zero-Overhead Principle: Designed for high performance, aiming to be faster than RxCpp by minimizing heap usage and unnecessary copies.
• Declarative Syntax: Uses a sequence of operators to transform and manage data flows.

RPP also provides specialized modules for integration with:

• Qt (rppqt)
• gRPC (rppgrpc)
• boost::asio (rppasio)

rpp::source::from_callable(&::getchar)
   | rpp::operators::repeat()
   | rpp::operators::take_while([](char v) { return v != '0'; })
   | rpp::operators::filter(std::not_fn(&::isdigit))
   | rpp::operators::map(&::toupper)
   | rpp::operators::subscribe([](char v) { std::cout << v; });

Reactive programming in RPP is based on the relationship between an Observable (the data source) and an Observer (the consumer).

An Observer subscribes to an Observable. The Observable then notifies the Observer of events via a specific contract:

• on_next(T): Notifies the observer about a new event or emission.
• on_error(std::exception_ptr): Notifies the observer about an error. This is a termination event; no further calls will be made from this observable.
• on_completed(): Notifies the observer of successful completion. This is also a termination event.
• is_disposed(): Allows the observer to check if it is still interested in receiving data from the source.

During subscription, an Observable may provide a Disposable to the Observer, allowing the Observer to terminate the observable's internal actions or state early.

#include <rpp/rpp.hpp>
#include <iostream>

int main()
{
    rpp::source::create<int>([](const auto& observer)
    {
        while (!observer.is_disposed())
        {
            char ch = ::getchar();
            if (!::isdigit(ch))
            {
              observer.on_error(std::make_exception_ptr(std::runtime_error{"Invalid symbol"}));
              return;
            }
            int digit = ch - '0';
            if (digit == 0)
            {
                observer.on_completed();
                return;
            }
            observer.on_next(digit);
        }
    })
    .subscribe([](int val)
               {
                   std::cout << "obtained val " << val << std::endl;
               },
                [](std::exception_ptr err)
                {
                   std::cout << "obtained error " << std::endl;
                },
               []()
               {
                   std::cout << "Completed" << std::endl;
               });

    return 0;
}

By default, an Observable performs its work in the thread where the subscribe operation occurs. To enable asynchronous or non-blocking operations across multiple threads, use Schedulers via two primary operators:

1. subscribe_on(scheduler): Forces the subscription process and the subsequent internal work (like iterating and emitting values) to occur on the specified scheduler.
2. observe_on(scheduler): Specifies the scheduler used for emissions for all subsequent operators in the chain. The subscription flow happens in the original thread, but the delivery of values is transferred to the new scheduler.

A Scheduler controls the multithreading behavior, such as using a new thread, a thread pool, or a raw queue.

// Example: Using subscribe_on to move the subscription work to a new thread
rpp::source::just(1,2,3,4,5)                                  
| rpp::operators::subscribe_on(rpp::schedulers::new_thread{}) 
| rpp::operators::subscribe([](auto){});                      

// Example: Using observe_on to move emissions to a new thread
rpp::source::just(1,2,3,4,5)                                  
| rpp::operators::observe_on(rpp::schedulers::new_thread{})
| rpp::operators::subscribe([](auto){});

Most RPP classes like observable and observer are heavily templated. While you should generally use auto to deduce types, you may need type-erased versions when you cannot use auto (e.g., when storing an observable as a member variable in a struct or class).

dynamic_observable<T> and dynamic_observer<T>

These are type-erased wrappers that hide complex template parameters.

• Conversion: You can convert a regular observable/observer to a dynamic version by calling the .as_dynamic() member function or by passing the object directly to the dynamic_* constructor.
• Performance: Because they use type-erasure (involving std::shared_ptr and indirect calls), they incur a minor performance penalty.
• Best Practice: Avoid using dynamic_* in hot paths like operators (e.g., inside flat_map). Instead, use auto or specific constraints to avoid unnecessary heap allocations and conversions.

Example: Storing in a struct

#include <rpp/rpp.hpp>
#include <iostream>

struct some_data
{
  rpp::dynamic_observable<int> observable;
  rpp::dynamic_observer<int> observer;
};

int main() {
    some_data v{rpp::source::just(1,2,3),
                rpp::make_lambda_observer([](int value){
                    std::cout << value << std::endl;
                })};

    v.observable.subscribe(v.observer);
}

ReactivePlusPlus uses Disposable objects to manage the lifecycle and cleanup of subscriptions.

Disposable Types

• interface_disposable: The base interface for all disposables. It provides dispose() and is_disposed() methods. This is the type typically passed to an observer.
• callback_disposable: A simple disposable that executes a noexcept callback when dispose() is called.
• interface_composite_disposable: A base interface for disposables that can aggregate other disposables. It includes an add() method. When the main composite disposable is disposed, it automatically calls dispose() on all its dependents.
• composite_disposable: A concrete implementation of interface_composite_disposable.
• refcount_disposable: A variant of composite_disposable that uses reference counting. You can increment the counter using add_ref(), which returns a new dependent composite_disposable. The original refcount_disposable is only disposed once all dependents created via add_ref() have called their dispose() methods.

Usage Note

All disposables in RPP should be managed via the following wrappers:

• rpp::disposable_wrapper: A wrapper over interface_disposable.
• rpp::composite_disposable_wrapper: A wrapper over interface_composite_disposable.

auto d = rpp::make_callback_disposable([]() noexcept { std::cout << "DISPOSED! " << std::endl; });

ReactivePlusPlus is a header-only library. While you can build it to generate artifacts, by default it compiles "nothing" unless you enable specific modules via CMake options.

Build Steps

For single-configuration generators (e.g., Unix Makefiles):

cmake -S . -B build -D CMAKE_BUILD_TYPE=Release
cmake --build build

For multi-configuration generators (e.g., Visual Studio):

cmake -S . -B build
cmake --build build --config Release

Build Options

You can enable or disable specific parts of the project using the following CMake flags:

• RPP_BUILD_TESTS: Build unit tests (default: OFF)
• RPP_BUILD_EXAMPLES: Build usage examples (default: OFF)
• RPP_BUILD_SFML_CODE: Build SFML-related code (requires SFML installed, default: OFF)
• RPP_BUILD_QT_CODE: Build QT-related code (requires QT5/6 installed, default: OFF)
• RPP_BUILD_GRPC_CODE: Build gRPC-related code (requires gRPC++/protobuf installed, default: OFF)
• RPP_BUILD_ASIO_CODE: Build RPPASIO related code (requires asio installed, default: OFF)

The project uses CMake presets to simplify configuration. Developers should create a CMakeUserPresets.json file at the project root to define custom presets. This file allows you to inherit from base presets defined in CMakePresets.json while adding your own configuration, such as setting RPP_BUILD_EXAMPLES to ON or specifying build types.

It is highly recommended to use user-* presets and replace <some_preset> with a valid base preset name from the main CMakePresets.json file.

{
  "version": 2,
  "cmakeMinimumRequired": {
    "major": 3,
    "minor": 14,
    "patch": 0
  },
  "configurePresets": [
    {
      "name": "dev",
      "binaryDir": "${sourceDir}/build",
      "inherits": ["<some_preset>"],
      "cacheVariables": {
        "CMAKE_BUILD_TYPE" : "Release",
        "RPP_BUILD_EXAMPLES" : "ON"
      }
    }
  ],
  "buildPresets": [
    {
      "name": "dev",
      "configurePreset": "dev",
      "configuration": "Release",
      "jobs" : 2
    }
  ],
  "testPresets": [
    {
      "name": "dev",
      "configurePreset": "dev",
      "configuration": "Release",
      "output": {
        "outputOnFailure": true
      }
    }
  ]
}