### Include ReactivePlusPlus Library in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Includes the entire ReactivePlusPlus library functionality in your C++ code. This is the simplest way to get started with the library.

```cpp
#include <rpp/rpp.hpp>
```

--------------------------------

### Install ReactivePlusPlus with CMake (Single-Config)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Installs the built ReactivePlusPlus artifacts using a single-configuration CMake generator. This command assumes the project has already been built successfully.

```sh
cmake --install build
```

--------------------------------

### Fetch ReactivePlusPlus using CMake FetchContent

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Includes and makes the ReactivePlusPlus library available in your project using CMake's FetchContent module. This downloads the library from a Git repository and tag.

```cmake
Include(FetchContent)

FetchContent_Declare(
    RPP
    GIT_REPOSITORY https://github.com/victimsnino/ReactivePlusPlus.git
    GIT_TAG        origin/v2
)

FetchContent_MakeAvailable(RPP)
```

--------------------------------

### Install ReactivePlusPlus with CMake (Multi-Config)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Installs the built ReactivePlusPlus artifacts using a multi-configuration CMake generator. This command requires the build to have been completed with a specified configuration.

```sh
cmake --install build --config Release
```

--------------------------------

### Complete Reactive Pipeline Example in C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Illustrates building a comprehensive reactive data processing pipeline by chaining multiple operators. This example reads characters, filters out digits, converts remaining characters to uppercase, and prints them until '0' is encountered.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <functional>

int main() {
    // Read characters from input until '0', filter out digits,
    // convert to uppercase, and print
    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; });

    return 0;
}

```

--------------------------------

### Build ReactivePlusPlus with CMake (Multi-Config)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Builds the ReactivePlusPlus library in release mode using a multi-configuration CMake generator such as Visual Studio. This process configures the build and then initiates the build with a specified configuration.

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

--------------------------------

### Include ReactivePlusPlus Forward Declarations in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Includes forward declarations for ReactivePlusPlus, which can be useful for reducing compile times when the full library functionality is not immediately needed.

```cpp
#include <rpp/fwd.hpp>
```

--------------------------------

### Void Function Example in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

A basic C++ function that takes no arguments and returns nothing. This demonstrates a function with an empty input and output, serving as a foundational example in programming.

```cpp
void function() { }
```

--------------------------------

### Build ReactivePlusPlus with CMake (Single-Config)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Builds the ReactivePlusPlus library in release mode using a single-configuration CMake generator like Unix Makefiles. This involves configuring the build with CMake and then executing the build command.

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

--------------------------------

### Find and Link ReactivePlusPlus CMake Package

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Finds the ReactivePlusPlus CMake package and links its targets to your project. This allows you to use the library's features by declaring it as a build requirement.

```cmake
find_package(RPP REQUIRED)
# Declare the imported target as a build requirement using PRIVATE, where
# project_target is a target created in the consuming project
target_link_libraries(project_target
    PRIVATE
        RPP::rpp
        RPP::rppqt
        RPP::rppasio
        RPP::rppgrpc
)
```

--------------------------------

### CMake Build Configuration for ReactivePlusPlus

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/CMakeLists.txt

This is a CMakeLists.txt file that configures the build process for the ReactivePlusPlus library. It uses the add_subdirectory command to include different parts of the project, such as the core library, extensions, benchmarks, tests, and examples. The inclusion of benchmarks, tests, and examples can be controlled by CMake variables like RPP_BUILD_BENCHMARKS, RPP_BUILD_TESTS, and RPP_BUILD_EXAMPLES.

```cmake
add_subdirectory(rpp)
add_subdirectory(extensions)

if (RPP_BUILD_BENCHMARKS)
    add_subdirectory(benchmarks)
endif()

if(RPP_BUILD_TESTS)
    add_subdirectory(tests)
endif()

if(RPP_BUILD_EXAMPLES)
    add_subdirectory(examples)
endif()
```

--------------------------------

### CMake Project Setup for ReactivePlusPlus

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CMakeLists.txt

Initializes the CMake project, setting the minimum version, project name, version, description, homepage URL, and supported languages. It also includes external CMake scripts for prelude and top-level project checks.

```cmake
cmake_minimum_required(VERSION 3.14)

include(cmake/prelude.cmake)

project(
    ReactivePlusPlus
    VERSION 2.2.1
    DESCRIPTION "ReactivePlusPlus is library for building asynchronous event-driven streams of data with help of sequences of primitive operators in the declarative form"
    HOMEPAGE_URL "https://github.com/victimsnino/ReactivePlusPlus"
    LANGUAGES CXX
)

include(cmake/project-is-top-level.cmake)
```

--------------------------------

### Include Multiple ReactivePlusPlus Operators in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Includes multiple specific operators from the ReactivePlusPlus library, such as 'map' and 'filter'. Note that for partial inclusion, headers for each used operator must be explicitly included.

```cpp
#include <rpp/operators/map.hpp>
#include <rpp/operators/filter.hpp>
```

--------------------------------

### CMake Install Rules Configuration

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CMakeLists.txt

Conditionally includes CMake scripts for defining installation rules if CMAKE_SKIP_INSTALL_RULES is not set. This allows the project to be installed to the system or a specified location.

```cmake
# ---- Install rules ----

if(NOT CMAKE_SKIP_INSTALL_RULES)
  include(cmake/install-rules.cmake)
endif()
```

--------------------------------

### Use dynamic observable and observer in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Illustrates how to use `dynamic_observable` and `dynamic_observer` in C++ to manage observables and observers with potentially complex template types. This example shows storing them as member variables and subscribing.

```cpp
#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);
}

```

--------------------------------

### Just Operator with Stack Memory Model

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Example of the rpp::source::just operator using the default 'use_stack' memory model, where variables are copied or moved as needed.

```cpp
rpp::source::just(my_custom_variable);
```

--------------------------------

### Include Specific ReactivePlusPlus Operator in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/BUILDING.md

Includes a specific operator from the ReactivePlusPlus library, such as 'map'. This approach avoids including the entire library, potentially improving compile times.

```cpp
#include <rpp/operators/map.hpp>
#include <rpp/observables/fwd.hpp>
```

--------------------------------

### CMake Build Configuration for ReactivePlusPlus Modules

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/tests/CMakeLists.txt

This section shows the conditional registration of tests for different ReactivePlusPlus modules (core, Qt, gRPC, Asio) using the `rpp_register_tests` macro. It also includes setup for gRPC proto targets. The modules are built only if their corresponding CMake variables (e.g., `RPP_BUILD_QT_CODE`) are enabled.

```cmake
rpp_register_tests(rpp)

if (RPP_BUILD_QT_CODE)
  rpp_register_tests(rppqt)
endif()

if (RPP_BUILD_GRPC_CODE)
  rpp_add_proto_target(rppgrpc_tests_proto rppgrpc/proto.proto)
  rpp_register_tests(rppgrpc)
endif()

if (RPP_BUILD_ASIO_CODE)
  rpp_register_tests(rppasio)
endif()
```

--------------------------------

### C++ Operator Constraint Example

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CONTRIBUTING.md

Illustrates correct and incorrect C++ template constraint usage for operators. The preferred method allows IDEs like VS Code to correctly deduce observable types.

```cpp
void operator(auto&& value, const constraint::subscriber auto& subscribier)

template<constraint::subscriber TSub>
void operator(auto&& value, const TSub& subscribier)
```

--------------------------------

### CMake Build and Test Commands (Shell)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/HACKING.md

Provides the essential shell commands to configure, build, and test the project using CMake presets. These commands are designed to be run from the project root and are OS-agnostic.

```sh
cmake --preset=dev
cmake --build --preset=dev
ctest --preset=dev
```

--------------------------------

### CMake User Presets Configuration (JSON)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/HACKING.md

Defines CMake presets for development, including configuration, build, and test settings. It allows customization of build directories, inheritance from base presets, and specific build configurations like Release with parallel jobs.

```json
{
  "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
      }
    }
  ]
}
```

--------------------------------

### C++20 Reactive Stream Example

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/Readme.md

This C++20 code snippet demonstrates building an asynchronous event-driven stream using ReactivePlusPlus. It reads characters from stdin, repeats them, filters out digits, converts letters to uppercase, and prints them to the console until '0' is entered. Dependencies include the ReactivePlusPlus library and standard C++ headers.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <functional>

int main()
{
    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; });
    return 0;
}
```

--------------------------------

### Create Observables with 'just' in C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Creates an observable that emits a fixed sequence of values. Supports basic usage, shared memory models for expensive objects, and specifying emission schedulers.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <array>

int main() {
    // Basic usage - emit multiple values
    rpp::source::just(42, 53, 10, 1)
        .subscribe([](int v) { std::cout << v << std::endl; });
    // Output: 42 53 10 1

    // Using shared memory model for expensive-to-copy objects
    std::array<int, 100> expensive_to_copy_1{};
    std::array<int, 100> expensive_to_copy_2{};
    rpp::source::just<rpp::memory_model::use_shared>(expensive_to_copy_1, expensive_to_copy_2)
        .subscribe([](const auto&) {});

    // Specify scheduler for emission
    rpp::source::just(rpp::schedulers::immediate{}, 42, 53)
        .subscribe([](const auto&) {});
    rpp::source::just(rpp::schedulers::current_thread{}, 42, 53)
        .subscribe([](const auto&) {});

    return 0;
}
```

--------------------------------

### Infinite Character Input Loop in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

This C++ code snippet demonstrates handling input distributed in time by continuously reading characters from standard input and printing them. It illustrates a basic event-driven approach where the program reacts to incoming data asynchronously.

```cpp
#include <iostream>

int main()
{
  while(true)
  {
     auto ch = ::getchar();
     std::cout << "Obtained char " << ch << std::endl;
  }
}
```

--------------------------------

### CMake Build Configuration for Snake Game

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/examples/rpp/sfml/snake/CMakeLists.txt

Configures the build for the Snake game executable. It defines the target name, lists all source and header files, links necessary libraries (ReactivePlusPlus and SFML), and sets target properties. Conditional post-build commands are included for Windows to copy SFML DLLs to the executable directory.

```cmake
set(TARGET snake)

add_executable(${TARGET}
    main.cpp

    snake.cpp
    snake.hpp

    utils.hpp

    canvas.hpp
    canvas.cpp
)

target_link_libraries(${TARGET} PRIVATE rpp sfml-graphics)
set_target_properties(${TARGET} PROPERTIES FOLDER Examples/rpp/SFML)

# if (WIN32)
#   add_custom_command (TARGET ${TARGET} POST_BUILD  COMMAND ${CMAKE_COMMAND} -E copy_if_different $<TARGET_FILE:sfml-graphics> $<TARGET_FILE_DIR:${TARGET}>)
#   add_custom_command (TARGET ${TARGET} POST_BUILD  COMMAND ${CMAKE_COMMAND} -E copy_if_different $<TARGET_FILE:sfml-system> $<TARGET_FILE_DIR:${TARGET}>)
#   add_custom_command (TARGET ${TARGET} POST_BUILD  COMMAND ${CMAKE_COMMAND} -E copy_if_different $<TARGET_FILE:sfml-window> $<TARGET_FILE_DIR:${TARGET}>)
# endif()
```

--------------------------------

### Define Executable Target and Link Libraries (CMake)

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/benchmarks/CMakeLists.txt

This CMake code defines an executable target named 'benchmarks' and links it with the 'rpp' and 'nanobench' libraries. It also includes conditional linking of 'rxcpp' and sets compile definitions based on build flags like RPP_BUILD_RXCPP and RPP_DISABLE_DISPOSABLES_OPTIMIZATION.

```cmake
set(TARGET benchmarks)

add_executable(${TARGET} benchmarks.cpp)

target_link_libraries(${TARGET} PRIVATE rpp nanobench::nanobench)
if (RPP_BUILD_RXCPP)
    target_link_libraries(${TARGET} PRIVATE rxcpp)
    target_compile_definitions(${TARGET} PRIVATE RPP_BUILD_RXCPP)

    get_target_property(DEP_DIR rxcpp INTERFACE_INCLUDE_DIRECTORIES)
    target_include_directories(${TARGET} SYSTEM PRIVATE ${DEP_DIR})
endif()
if(RPP_DISABLE_DISPOSABLES_OPTIMIZATION)
    target_compile_definitions(${TARGET} PRIVATE "RPP_DISABLE_DISPOSABLES_OPTIMIZATION=${RPP_DISABLE_DISPOSABLES_OPTIMIZATION}")
endif()

set_target_properties(${TARGET} PROPERTIES FOLDER Tests)
set_target_properties(${TARGET} PROPERTIES CXX_CLANG_TIDY "")


add_test(NAME ${TARGET} COMMAND  $<TARGET_FILE:${TARGET}> --dump=${RPP_TEST_RESULTS_DIR}/benchmarks_results.json)
```

--------------------------------

### Subscribe on New Thread Scheduler

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Demonstrates using the subscribe_on operator to force observable subscription and work onto a new thread scheduler. This is useful for offloading work from the main thread.

```cpp
rpp::source::just(1,2,3,4,5)                                  \
| rpp::operators::subscribe_on(rpp::schedulers::new_thread{}) \
| rpp::operators::subscribe([](auto){});
```

--------------------------------

### C++ Function and Class Formatting

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CONTRIBUTING.md

Demonstrates C++ code style for function and class formatting, including bracket placement and inline function exceptions. Follows project-specific naming conventions.

```cpp
void my_long_function()
{
  int v;
  int b = v+2;
}

# Option 1
int my_short_function() { return 2; }

# Option 2
int my_short_function()
{
    return 2;
}
```

--------------------------------

### Reactive C++ Observable from Console Input

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Demonstrates creating a Reactive++ observable that emits integer digits from console input. It handles 'on_next' for valid digits, 'on_error' for non-digit input, and 'on_completed' when '0' is entered. This showcases asynchronous event handling and error management in Reactive++.

```cpp
#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;
               });
    // input: 123456d
    // output:  obtained val 1
    //          obtained val 2
    //          obtained val 3
    //          obtained val 4
    //          obtained val 5
    //          obtained val 6
    //          obtained error

    // input: 1230
    // output: obtained val 1
    //         obtained val 2
    //         obtained val 3
    //         Completed

    return 0;
}
```

--------------------------------

### CMake Macro for Adding Test Targets

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/tests/CMakeLists.txt

Defines a CMake macro `add_test_target` to create executable test targets. It handles linking necessary libraries (like doctest, trompeloeil, and module-specific libraries), setting target properties, and conditionally adding support for Qt, gRPC, or Asio based on the module name. It also sets C++20 as a required feature and applies compiler warnings and error flags.

```cmake
macro(add_test_target target_name module files)
  set(TARGET ${target_name})

  add_executable(${TARGET} ${files})
  target_link_libraries(${TARGET} PRIVATE rpp_doctest_main trompeloeil::trompeloeil rpp_tests_utils ${module})
  set_target_properties(${TARGET} PROPERTIES FOLDER Tests/Suites/${module})

  add_test_with_coverage(${TARGET})

  if (${module} STREQUAL rppqt)
    rpp_add_qt_support_to_executable(${TARGET})
  endif()

  if (${module} STREQUAL rppgrpc)
    target_link_libraries(${TARGET} PRIVATE rppgrpc_tests_proto)
  endif()

  if (${module} STREQUAL rppasio)
    rpp_add_asio_support_to_executable(${TARGET})
  endif()

  target_compile_features(${TARGET} PRIVATE cxx_std_20)

  if(MSVC)
    target_compile_options(${TARGET} PRIVATE /W4 /WX /wd4702)
  else()
    target_compile_options(${TARGET} PRIVATE -Wall -Wextra -Wpedantic -Werror -Wconversion -Wno-gnu-zero-variadic-macro-arguments)
  endif()
endmacro()
```

--------------------------------

### Custom Observable Subscription on New Thread

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Provides an internal view of how subscribe_on schedules work by creating a custom observable that uses a new thread scheduler for its subscription logic.

```cpp
rpp::source::create<int>([](const auto& observer)
{
  rpp::schedulers::new_thread{}.create_worker([](...) { // Lambda for worker execution
    rpp::source::just(1,2,3,4,5).subscribe(observer);
  });
}).subscribe(...);
```

--------------------------------

### Multicasting Operator - publish and connect in C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Demonstrates how to convert a cold observable into a hot connectable observable using the 'publish' operator and manage its subscription lifecycle with 'connect' and 'dispose'. This is useful for sharing a single subscription among multiple observers.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <thread>
#include <chrono>

int main() {
    // Create connectable observable
    const auto observable = rpp::source::interval(std::chrono::milliseconds{50},
                                                  rpp::schedulers::new_thread{})
                          | rpp::ops::map([](int v) {
                                std::cout << "value in map " << v << std::endl;
                                return v;
                            })
                          | rpp::ops::publish();

    std::cout << "CONNECT" << std::endl;
    auto d = observable.connect(); // Subscription happens now

    std::this_thread::sleep_for(std::chrono::milliseconds{150});

    std::cout << "SUBSCRIBE" << std::endl;
    observable.subscribe([](int v) {
        std::cout << "observer value " << v << std::endl;
    });

    std::this_thread::sleep_for(std::chrono::milliseconds{150});

    d.dispose();
    std::cout << "DISPOSE" << std::endl;

    std::this_thread::sleep_for(std::chrono::milliseconds{150});

    return 0;
}

```

--------------------------------

### CMake Include and Subdirectory Directives

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CMakeLists.txt

Includes common CMake configuration files for variables and dependencies. It also adds the 'src' subdirectory, which likely contains the main source code for the library.

```cmake
include(cmake/variables.cmake)
include(cmake/dependencies.cmake)


add_subdirectory(src)
```

--------------------------------

### Just Operator with Shared Memory Model

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Demonstrates using rpp::source::just with the 'use_shared' memory model, which copies the variable once to a shared_ptr for efficient reuse.

```cpp
rpp::source::just<rpp::memory_model::use_shared>(my_custom_variable);
```

--------------------------------

### CMake Macro for Registering Tests

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/src/tests/CMakeLists.txt

Defines a CMake macro `rpp_register_tests` that finds all test files (`test_*.cpp`) within a specified module directory. It conditionally adds test targets either individually or together based on the `RPP_BUILD_TESTS_TOGETHER` variable, utilizing the `add_test_target` macro.

```cmake
macro(rpp_register_tests module)
  file(GLOB_RECURSE RPP_FILES "${module}/test_*.cpp")
  if (RPP_BUILD_TESTS_TOGETHER)
    add_test_target(test_${module} ${module} "${RPP_FILES}")
  else()
    foreach(SOURCE ${RPP_FILES})
      get_filename_component(BASE_NAME ${SOURCE} NAME_WE)
      add_test_target(${BASE_NAME} ${module} ${SOURCE})
    endforeach()
  endif()
endmacro()
```

--------------------------------

### CMake Developer Mode and Testing Configuration

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/CMakeLists.txt

Configures developer mode if RPP_DEVELOPER_MODE is enabled and includes development-specific CMake scripts. It also sets up testing infrastructure, including enabling tests, defining a results directory, and providing a macro for adding tests with optional coverage.

```cmake
if(RPP_DEVELOPER_MODE)
  if(NOT PROJECT_IS_TOP_LEVEL)
    message(AUTHOR_WARNING "RPP_DEVELOPER_MODE is intended for developers of ReactivePlusPlus")
  endif()

  include(cmake/dev-mode.cmake)
endif()


if (RPP_BUILD_TESTS OR RPP_BUILD_BENCHMARKS)
  enable_testing()
  set(RPP_TEST_RESULTS_DIR  ${CMAKE_BINARY_DIR}/test_results)
  file(MAKE_DIRECTORY ${RPP_TEST_RESULTS_DIR})

  macro(add_test_with_coverage TARGET)
    if (RPP_ENABLE_COVERAGE)
      add_test(NAME ${TARGET} COMMAND ${CMAKE_COMMAND} -E env LLVM_PROFILE_FILE=${RPP_TEST_RESULTS_DIR}/${TARGET}.profraw $<TARGET_FILE:${TARGET}>)
    else()
      add_test(NAME ${TARGET} COMMAND $<TARGET_FILE:${TARGET}>)
    endif()
  endmacro()
endif()
```

--------------------------------

### Create Custom Observables with 'create' in C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Allows for the creation of custom observables by providing a lambda function that defines the emission logic. This offers full control over `on_next`, `on_error`, and `on_completed` events, and can capture external variables.

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

int main() {
    // Basic custom observable
    rpp::source::create<int>([](const auto& sub) {
        sub.on_next(42);
        sub.on_completed();
    }).subscribe([](int v) { std::cout << v << std::endl; });
    // Output: 42

    // With captured variables
    int val = 42;
    rpp::source::create<int>([val](const auto& sub) {
        sub.on_next(val);
        sub.on_next(val * 2);
        sub.on_completed();
    }).subscribe([](int v) { std::cout << v << std::endl; });
    // Output: 42 84

    return 0;
}
```

--------------------------------

### Create Interval Observable with Reactive++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Generates sequential integers at specified time intervals using Reactive++. Supports defining interval periods and initial delays. Emits values and completes.

```cpp
#include <rpp/rpp.hpp>
#include <chrono>
#include <iostream>

int main() {
    // Emit every 10ms
    rpp::source::interval(std::chrono::milliseconds(10), rpp::schedulers::immediate{}) 
        | rpp::operators::take(3)
        | rpp::operators::subscribe(
            [start = rpp::schedulers::clock_type::now()](size_t v) {
                std::cout << "emit " << v
                         << " duration since start "
                         << std::chrono::duration_cast<std::chrono::milliseconds>(
                                rpp::schedulers::clock_type::now() - start).count()
                         << "ms\n";
            },
            rpp::utils::rethrow_error_t{},
            []() { std::cout << "On complete\n"; });
    // Output: emit 0 duration since start 0ms
    //         emit 1 duration since start 10ms
    //         emit 2 duration since start 20ms
    //         On complete

    // With initial delay and period
    rpp::source::interval(std::chrono::milliseconds(5),
                         std::chrono::milliseconds(10),
                         rpp::schedulers::immediate{})
        | rpp::operators::take(3)
        | rpp::operators::subscribe(
            [start = rpp::schedulers::clock_type::now()](size_t v) {
                std::cout << "emit " << v
                         << " duration since start "
                         << std::chrono::duration_cast<std::chrono::milliseconds>(
                                rpp::schedulers::clock_type::now() - start).count()
                         << "ms\n";
            },
            rpp::utils::rethrow_error_t{},
            []() { std::cout << "On complete\n"; });
    // Output: emit 0 duration since start 5ms
    //         emit 1 duration since start 15ms
    //         emit 2 duration since start 25ms
    //         On complete

    return 0;
}

```

--------------------------------

### Create and use a callback disposable in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Demonstrates the creation of a `callback_disposable` in C++. This disposable executes a noexcept lambda function when its `dispose()` method is called. It's useful for simple cleanup actions.

```cpp
#include <rpp/disposable.hpp>
#include <iostream>

rpp::callback_disposable d = rpp::make_callback_disposable([]() noexcept {
    std::cout << "DISPOSED! " << std::endl;
});
// Later, when the resource is no longer needed:
d.dispose();

```

--------------------------------

### Custom Observable Emission on New Thread

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Shows an internal implementation detail of observe_on, where a custom observable uses a new thread scheduler to emit values to the observer.

```cpp
rpp::source::create<int>([](const auto& observer)
{
  auto worker = rpp::schedulers::new_thread{}.create_worker();
  rpp::source::just(1,2,3,4,5).subscribe([](int v) {
    worker.scheduler([](...) { // Lambda for scheduling emission
      observer.on_next(v);
    });
  });
}).subscribe(...);
```

--------------------------------

### Observe on New Thread Scheduler

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Illustrates the observe_on operator, which specifies the scheduler for emissions after the observe_on operator. The subscription flow remains in the current thread, but emissions are moved to a new thread.

```cpp
rpp::source::just(1,2,3,4,5)
| rpp::operators::observe_on(rpp::schedulers::new_thread{})
| rpp::operators::subscribe([](auto){});
```

--------------------------------

### Sum Function with Static Input in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

A C++ function that calculates the sum of two integers. This exemplifies handling static input through function arguments, a common pattern in synchronous programming.

```cpp
int sum(int a, int b) { return a + b; }
```

--------------------------------

### Create Observables from Iterables and Callables in C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Generates observables from standard containers (like std::vector) or from callable functions. Supports shared memory models and specific schedulers for emission.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <vector>

int main() {
    // Create from container
    std::vector<int> vals{1, 2, 3};
    rpp::source::from_iterable(vals)
        .subscribe([](int v) { std::cout << v << " "; });
    // Output: 1 2 3

    // Using shared memory model
    rpp::source::from_iterable<rpp::memory_model::use_shared>(vals)
        .subscribe([](int v) { std::cout << v << " "; });

    // With scheduler
    rpp::source::from_iterable(vals, rpp::schedulers::immediate{})
        .subscribe([](int v) { std::cout << v << " "; });
    rpp::source::from_iterable(vals, rpp::schedulers::current_thread{})
        .subscribe([](int v) { std::cout << v << " "; });

    // Create from callable function
    rpp::source::from_callable([]() { return 49; })
        .subscribe([](int v) { std::cout << v << " "; });
    // Output: 49

    return 0;
}
```

--------------------------------

### Create Defer Observable with Reactive++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

Defers the creation of an observable until subscription time, ensuring a new observable is generated for each subscriber. This is useful for creating fresh state or ensuring observable logic executes upon subscription.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <array>
#include <memory>

int main() {
    // Defer creation until subscription
    rpp::source::defer([] {
        std::cout << "Observable factory called\n";
        return rpp::source::from_iterable(std::array{1, 2, 3});
    }).subscribe([](int v) { std::cout << v << "\n"; },
                 rpp::utils::rethrow_error_t{},
                 []() { std::cout << "On complete\n"; });
    // Output: Observable factory called
    //         1
    //         2
    //         3
    //         On complete

    // Create fresh state for each subscription
    auto obs = rpp::source::defer([] {
        std::cout << "Observable factory called\n";
        const auto state = std::make_shared<int>(0);
        auto inner_obs = rpp::source::create<int>([state](const auto& obs) {
            obs.on_next((*state)++);
            obs.on_completed();
        });
        return inner_obs;
    });

    obs.subscribe([](int v) { std::cout << v << "\n"; },
                  rpp::utils::rethrow_error_t{},
                  []() { std::cout << "On complete\n"; });
    obs.subscribe([](int v) { std::cout << v << "\n"; },
                  rpp::utils::rethrow_error_t{},
                  []() { std::cout << "On complete\n"; });
    // Output: Observable factory called
    //         0
    //         On complete
    //         Observable factory called
    //         0
    //         On complete

    return 0;
}

```

--------------------------------

### Avoid dynamic observable in hot paths with auto in C++

Source: https://github.com/victimsnino/reactiveplusplus/blob/v2/docs/readme.md

Shows a reactive stream pipeline in C++ where using `auto` or `constraint::observable_of_type` is preferred over `dynamic_observable` within `flat_map` to avoid performance penalties associated with type erasure on hot paths.

```cpp
#include <rpp/rpp.hpp>

rpp::source::just(1, 2, 3)
| rpp::ops::map([](int v) { return rpp::source::just(v); })
| rpp::ops::flat_map([](const rpp::constraint::observable_of_type<int> auto& observable) { // or just `const auto& observable`
    return observable | rpp::ops::filter([](int v){ return v % 2 == 0;
    });
});

```

--------------------------------

### Control observable emission thread with subscribe_on - C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The 'subscribe_on' operator dictates which scheduler the source observable uses for its emissions. It requires 'iostream' and 'thread' headers, along with 'rpp/rpp.hpp'. This allows for offloading the work of an observable to a different thread, preventing blocking of the main thread.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <thread>

int main() {
    std::cout << std::this_thread::get_id() << std::endl;

    rpp::source::create<int>([](const auto& sub) {
        std::cout << "on_subscribe thread " << std::this_thread::get_id() << std::endl;
        sub.on_next(1);
        sub.on_completed();
    })
        | rpp::operators::subscribe_on(rpp::schedulers::new_thread{})
        | rpp::operators::as_blocking()
        | rpp::operators::subscribe([](int v) {
              std::cout << "[" << std::this_thread::get_id() << "] : " << v << "\n";
          });

    std::cout << std::this_thread::get_id() << std::endl;
    // Output shows subscription happens on new thread while main thread continues

    return 0;
}

```

--------------------------------

### Buffer Emitted Items into Batches with buffer (C++)

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The buffer operator collects emitted items from an observable into batches or buffers of a specified size. Once a buffer is full, it's emitted, and a new buffer is created. Requires rpp/rpp.hpp.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <vector>

std::ostream& operator<<(std::ostream& out, const std::vector<int>& list) {
    out << "{";
    auto size = list.size();
    for (size_t i = 0; i < size; ++i) {
        out << list.at(i);
        if (i < size - 1) {
            out << ",";
        }
    }
    out << "}";
    return out;
}

int main() {
    // Buffer items in groups of 2
    rpp::source::just(1, 2, 3, 4, 5)
        | rpp::ops::buffer(2)
        | rpp::ops::subscribe([](const std::vector<int>& v) { std::cout << v << "-"; },
                              [](const std::exception_ptr&) {},
                              []() { std::cout << "|" << std::endl; });
    // Output: {1,2}-{3,4}-{5}-|

    return 0;
}
```

--------------------------------

### Concatenate Observables Sequentially with concat (C++)

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The concat operator subscribes to observables sequentially, emitting all items from the first observable before subscribing to the next. It can be used as a source operator or an intermediate operator. Requires rpp/rpp.hpp.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <vector>

int main() {
    // Concat as source
    rpp::source::concat(rpp::source::just(1),
                       rpp::source::just(2),
                       rpp::source::just(1, 2, 3))
        .subscribe([](int v) { std::cout << v << ", "; },
                   [](const std::exception_ptr&) {},
                   []() { std::cout << "completed\n"; });
    // Output: 1, 2, 1, 2, 3, completed

    // Concat from vector of observables
    auto observables = std::vector{rpp::source::just(1), rpp::source::just(2)};
    rpp::source::concat<rpp::memory_model::use_shared>(observables)
        .subscribe([](int v) { std::cout << v << ", "; },
                   [](const std::exception_ptr&) {},
                   []() { std::cout << "completed\n"; });
    // Output: 1, 2, completed

    // Concat as operator
    rpp::source::just(rpp::source::just(1).as_dynamic(),
                      rpp::source::just(2).as_dynamic(),
                      rpp::source::just(1, 2, 3).as_dynamic())
        | rpp::operators::concat()
        | rpp::operators::subscribe([](int v) { std::cout << v << ", "; },
                                     [](const std::exception_ptr&) {},
                                     []() { std::cout << "completed\n"; });
    // Output: 1, 2, 1, 2, 3, completed

    return 0;
}
```

--------------------------------

### Delay emissions of an observable by a specified duration - C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The 'delay' operator postpones the emission of items from an observable by a given time duration. It requires the 'chrono' and 'iostream' headers, along with the 'rpp/rpp.hpp' library. The operator takes a duration and a scheduler as arguments, affecting when downstream subscribers receive the emitted values.

```cpp
#include <rpp/rpp.hpp>
#include <chrono>
#include <iostream>

int main() {
    auto start = rpp::schedulers::clock_type::now();

    rpp::source::create<int>([&start](const auto& obs) {
        for (int i = 0; i < 3; ++i) {
            auto emitting_time = rpp::schedulers::clock_type::now();
            std::cout << "emit " << i << " in thread{" << std::this_thread::get_id()
                     << "} duration since start "
                     << std::chrono::duration_cast<std::chrono::seconds>(emitting_time - start).count()
                     << "s" << std::endl;

            obs.on_next(i);
            std::this_thread::sleep_for(std::chrono::seconds{1});
        }
        auto emitting_time = rpp::schedulers::clock_type::now();
        std::cout << "emit error in thread{" << std::this_thread::get_id()
                 << "} duration since start "
                 << std::chrono::duration_cast<std::chrono::seconds>(emitting_time - start).count()
                 << "s" << std::endl;
        obs.on_error({});
    })
        | rpp::operators::delay(std::chrono::seconds{3}, rpp::schedulers::new_thread{})
        | rpp::operators::as_blocking()
        | rpp::operators::subscribe(
            [&](int v) {
                auto observing_time = rpp::schedulers::clock_type::now();
                std::cout << "observe " << v << " in thread{" << std::this_thread::get_id()
                         << "} duration since start "
                         << std::chrono::duration_cast<std::chrono::seconds>(observing_time - start).count()
                         << "s" << std::endl;
            },
            [&](const std::exception_ptr&) {
                auto observing_time = rpp::schedulers::clock_type::now();
                std::cout << "observe error in thread{" << std::this_thread::get_id()
                         << "} duration since start "
                         << std::chrono::duration_cast<std::chrono::seconds>(observing_time - start).count()
                         << "s" << std::endl;
            });
    // Output: emit 0 in thread{...} duration since start 0s
    //         emit 1 in thread{...} duration since start 1s
    //         emit 2 in thread{...} duration since start 2s
    //         observe 0 in thread{...} duration since start 3s
    //         emit error in thread{...} duration since start 3s
    //         observe 1 in thread{...} duration since start 4s
    //         observe 2 in thread{...} duration since start 5s
    //         observe error in thread{...} duration since start 6s

    return 0;
}

```

--------------------------------

### Combine Latest Values from Observables with combine_latest (C++)

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The combine_latest operator merges values from multiple observables. Whenever any source observable emits, it combines the latest values from all sources. It can use a tuple for results or a custom combiner function. Requires the rpp/rpp.hpp header.

```cpp
#include <rpp/rpp.hpp>
#include <iostream>
#include <tuple>

std::ostream& operator<<(std::ostream& out, const std::tuple<int, int>& value) {
    out << "{" << std::get<0>(value) << "," << std::get<1>(value) << "}";
    return out;
}

int main() {
    // Combine with tuple result
    rpp::source::just(rpp::schedulers::current_thread{}, 1, 2, 3)
        | rpp::ops::combine_latest(rpp::source::just(rpp::schedulers::current_thread{}, 4, 5, 6))
        | rpp::ops::subscribe([](const std::tuple<int, int>& v) { std::cout << "-" << v; },
                              [](const std::exception_ptr&) {},
                              []() { std::cout << "-|" << std::endl; });
    // Output: -{1,4}-{1,5}-{2,5}-{2,6}-{3,6}-|

    // With custom combiner function
    rpp::source::just(1, 2, 3)
        | rpp::ops::combine_latest([](int left, int right) { return left + right; },
                                   rpp::source::just(4, 5, 6))
        | rpp::ops::subscribe([](int v) { std::cout << "-" << v; },
                              [](const std::exception_ptr&) {},
                              []() { std::cout << "-|" << std::endl; });
    // Output: -5-6-7-8-9-|

    return 0;
}
```

--------------------------------

### Merge Multiple Observables into One with merge (C++)

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The merge operator combines multiple observables into a single stream, emitting items from all sources concurrently. It can merge an observable of observables or use merge_with to combine two observables. Requires rpp/rpp.hpp.

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

int main() {
    // Merge observable of observables
    rpp::source::just(rpp::source::just(1).as_dynamic(),
                      rpp::source::never<int>().as_dynamic(),
                      rpp::source::just(2).as_dynamic())
        | rpp::operators::merge()
        | rpp::operators::subscribe([](int v) { std::cout << v << " "; });
    // Output: 1 2

    // Merge with another observable
    rpp::source::just(1)
        | rpp::operators::merge_with(rpp::source::just(2))
        | rpp::operators::subscribe([](int v) { std::cout << v << " "; });
    // Output: 1 2

    return 0;
}
```

--------------------------------

### Retry an observable on error up to a specified number of times - C++

Source: https://context7.com/victimsnino/reactiveplusplus/llms.txt

The 'retry' operator resubscribes to the source observable when an error occurs, allowing for retries up to a specified limit. It requires 'exception' and 'iostream' headers, along with 'rpp/rpp.hpp'. This operator is useful for handling transient errors by attempting the operation again.

```cpp
#include <rpp/rpp.hpp>
#include <exception>
#include <iostream>

int main() {
    // Retry up to 2 times (3 subscriptions total)
    rpp::source::concat(rpp::source::just(1, 2, 3), rpp::source::error<int>({}))
        | rpp::operators::retry(2)
        | rpp::operators::subscribe([](int v) { std::cout << v << " "; },
                                    [](const std::exception_ptr&) { std::cout << "error" << std::endl; },
                                    []() { std::cout << "completed" << std::endl; });
    // Output: 1 2 3 1 2 3 1 2 3 error

    // Retry infinitely (use take to limit output)
    rpp::source::concat(rpp::source::just(1, 2, 3), rpp::source::error<int>({}))
        | rpp::operators::retry()
        | rpp::operators::take(10)
        | rpp::operators::subscribe([](int v) { std::cout << v << " "; },
                                    [](const std::exception_ptr&) { std::cout << "error" << std::endl; },
                                    []() { std::cout << "completed" << std::endl; });
    // Output: 1 2 3 1 2 3 1 2 3 1 completed

    return 0;
}

```