Enforce Extended Reactive

Enforce Extended Reactive
Enforce Extended Reactive is a lightweight synchronous reactive programming library for Arma Reforger, built on top of Enforce Extended.
It introduces observable streams and composable operators inspired by reactive programming libraries such as Reactive Extensions, while remaining designed around the capabilities and execution model of Enforce Script.
Create observable pipelines using strongly typed Observable<T>, Subject<T>, Observer<T>, subscriptions, generic extension methods, and lambda expressions.
Features
Observable streams
Publish values through subjects and subscribe with strongly typed observers.
Subject<int> source = new Subject<int>();

Observer<int> observer = new Observer<int>();
observer.Initialize(
    value => Print(value.ToString()),
    error => Print(error),
    () => Print("Completed")
);

source.Subscribe(observer);

source.OnNext(42);
source.OnCompleted();

Composable operators
Operators are implemented as extension methods, allowing pipelines to be written naturally:
source
    .Where(value => value > 0)
    .Select(value => value * 2)
    .Take(5);

Available operators include:
Where — filter values using a predicate
Select — transform values
Do — perform side effects while forwarding values
Skip — ignore the first N values
Take — emit only the first N values
SkipWhile — skip values while a predicate remains true
TakeWhile — emit values until a predicate fails
DistinctUntilChanged — suppress consecutive equal values
Scan — emit accumulated values
Switch — follow only the latest inner observable
SelectMany — merge values from multiple inner observables
Stream flattening
Nested observable streams can be flattened using Switch or SelectMany.
Switch subscribes only to the latest inner observable:
Observable<Observable<int>> sources = ...;

Subject<int> latest = sources.Switch();

When a new inner observable arrives, the previous one is detached.
SelectMany merges all active inner streams:
Subject<int> merged = sources.SelectMany();

The resulting stream completes after the outer observable and all active inner observables have completed.
Accumulation
Scan maintains state across emitted values and publishes every accumulated result:
Subject<int> totals = source.Scan(
    0,
    (sum, value) => sum + value
);

The seed initializes the accumulator but is not emitted by itself.
Automatic subscription lifetime management
Operators maintain their upstream subscriptions and propagate:
values
errors
completion
cancellation and detachment
Operators such as Take, TakeWhile, and Switch detach from upstream sources when they no longer need additional values.
Strongly typed generic operators
The library takes advantage of Enforce Extended generics and constraints:
static Subject<TResult> Select<TSource, TValue, TResult>(
    this TSource source,
    Func<TValue, TResult> selector
)
    where TSource implements Observable<TValue>

This allows operators to work with concrete observable implementations while preserving compile-time value types throughout a pipeline.
Example
Subject<int> source = new Subject<int>();

Subject<int> result = source
    .Where(value => value > 0)
    .DistinctUntilChanged()
    .TakeWhile(value => value < 10)
    .Scan(0, (sum, value) => sum + value);

Observer<int> observer = new Observer<int>();

observer.Initialize(
    value => Print("Value: " + value.ToString()),
    error => Print("Error: " + error),
    () => Print("Completed")
);

result.Subscribe(observer);

source.OnNext(1);
source.OnNext(1);
source.OnNext(2);
source.OnNext(3);
source.OnNext(10);

This pipeline:
removes non-positive values,
suppresses consecutive duplicates,
accepts values while they remain below 10,
produces a running total,
completes when the predicate fails.
Requirements
Enforce Extended Reactive is written using Enforce Extended language features such as:
generic classes and methods
generic interface constraints
extension methods
lambda expressions
Func and Action
interfaces
It is intended to be used together with the Enforce Extended compiler/lowering toolchain.
Scope
The library currently focuses on synchronous observable composition.
It provides filtering, projection, side effects, accumulation, stream limiting, distinct-value filtering, and nested-stream flattening without introducing a separate asynchronous scheduler or threading model.
Enforce Extended Reactive is intended to make event-driven and state-driven Reforger code easier to compose, reuse, and reason about while retaining the strongly typed structure of Enforce Extended.