1 / 6
Tango Users Meeting · C++ Edition
Reactive Programming
for Tango Controls
C++17 — same ReactiveX vocabulary, zero-overhead native pipelines
RxTango/cpp · observable<T> streams with RxCpp + cppTango
Igor Khokhriakov  ·  Principal Software Engineer
github.com/scientific-software-hub/rx-controls-suite
2 / 6
Pattern 1 Correlated Multi-Attribute Reads Beyond sequential DeviceProxy::read_attribute()
The scenario
"Every 500 ms, read double_scalar and float_scalar from TangoTest — only process the pair if both arrive in the same polling window. Discard if either fails."
With today's tools…
  • Two sequential read_attribute() calls — a timing gap opens between them
  • Manual std::thread + std::condition_variable to run them in parallel
  • Shared state to pair results; try/catch in each thread; no built-in "atomic multi-attribute read"
Most C++ Tango clients end up hand-rolling the same parallelism and pairing scaffolding — each slightly different, each with its own edge cases around failure and timing.
With Rx — rxcpp::observable<>::zip()
C++17 · observable<>::zip()
rxcpp::observable<>::interval(500ms)
  .flat_map([](long) {
    // Both reads fire on background threads in parallel
    return rxcpp::observable<>::zip(
      [](double a, float b) { return process(a, b); },
      rxtango::read_attribute<double>(
          device, "double_scalar"),
      rxtango::read_attribute<float>(
          device,  "float_scalar")
    );
  })
  .subscribe([](auto result) { publish(result); });
✓  If either read fails the pair is dropped — never half-processed.
✓  Zero std::condition_variable. Zero shared state.
✓  Template parameter selects DeviceAttribute extraction type.
3 / 6
Pattern 2 Real-Time Event Stream Processing Beyond manual CallBack::push_event() bookkeeping
The scenario
"Monitor double_scalar at PERIODIC events, compute a 5-sample sliding average, and write back only if it deviates more than 10% from the last written value."
With today's tools…
  • Inherit Tango::CallBack, override push_event()
  • Own std::deque<double> + index + last-written — same scaffolding every project
  • Thread-safe access from the cppTango event callback thread via mutex
Every Tango C++ project implementing stream processing reimplements the same ring buffer, write-guard, and lock — in slightly different ways each time.
With Rx — monitor_attribute + buffer + distinctUntilChanged
C++17 · buffer(5,1) + distinct_until_changed
rxtango::monitor_attribute<double>(
    device, "double_scalar", "periodic")

  // sliding window — no deque, no index arithmetic
  .buffer(5, 1)
  .map([](const auto& w) {
    return std::accumulate(
        w.begin(), w.end(), 0.0) / w.size();
  })

  // skip write if within 10% of last value
  .distinct_until_changed(
      [](double p, double c) {
        return std::abs(c-p)/p < 0.1; })

  .flat_map([](double avg) {
    return rxtango::write_attribute<double>(
        device, "double_scalar_w", avg); });
✓  No deque. No index. No write-guard variable.
✓  Dispose the subscription — no manual unsubscribe_event().
4 / 6
Pattern 3 Alarm Fan-In Beyond per-device subscribe_event registrations
The scenario
"Watch three TangoTest devices. When any one reports a fault, notify the operator immediately. The other two must continue monitoring unaffected."
With today's tools…
  • One DeviceProxy::subscribe_event() per device — three separate callback registrations
  • Manual fan-in: shared queue, std::mutex, or a dedicated aggregator thread
  • Error isolation requires extra shared state — one device failing can affect others
Each additional device is another registration, another edge case, and another potential threading bug.
With Rx — rxcpp::observable<>::merge()
C++17 · observable<>::merge()
rxcpp::observable<>::merge(
  rxtango::monitor_attribute<double>(
      dev1, "double_scalar"),
  rxtango::monitor_attribute<double>(
      dev2, "double_scalar"),
  rxtango::monitor_attribute<double>(
      dev3, "double_scalar")
)
.filter([](double v) {
  return std::abs(v) > THRESHOLD; })
.subscribe([](double v) {
  notify_operator(v); });
✓  Any one failing device propagates — others continue.
✓  Add a fourth device: one line.
✓  Callbacks serialised by monitor's internal mutex — no data races.
5 / 6
Library RxTango/cpp — Data Flow reference implementation

cppTango · DeviceProxy

Single-shot reads via read_attribute(); writes via write_attribute(); commands via command_inout(); push events via subscribe_event() + Tango::CallBack.

Reactive Observable (single-shot)

read_attribute<T> · write_attribute<T> · execute_command<R,A>
Dispatched on std::thread; emits one value then completes.

Reactive Observable (push)

monitor_attribute<T> — EventCallback bridges the cppTango event thread to rxcpp::subscriber<T> via std::mutex. Dispose calls unsubscribe_event().

RxCpp Operators

flat_map · zip · merge · buffer · filter · scan · sample_with_time · distinct_until_changed

Application / Downstream

Write-back, alarm, HMI update, file sink — pure functions, no shared state, ReactiveX contract verified.

Showstopper pipeline
6-step fluent chain
Read → calibrate → write → read-back → assert → log
All steps in one TangoClient chain; no intermediate variables.
TangoClient — fluent builder
rxtango::TangoClient()
  .read(device, "double_scalar")
  .map([](std::any v) -> std::any {
    return std::any_cast<double>(v) * 1.0328;
  })
  .write(device, "double_scalar_w")
  .subscribe(
    [](std::any v) {
      std::cout << std::any_cast<double>(v);
    });
Header-only · CMake + FetchContent · C++17
ReactiveX contract verified: ./build/tests/verify_contract
6 / 6
Thank you
Same ReactiveX idioms.
Every language.
poll  ·  zip  ·  sliding average  ·  backpressure  ·  fluent pipelines
AGPL-3.0 · C++17 · CMake + FetchContent · RxCpp · cppTango
github.com/scientific-software-hub/rx-controls-suite
Java edition RxTango/java · RxJava3 · jbang Python edition RxTango/python · RxPY v4 · uv EPICS C++ edition RxEpics/cpp · RxCpp · PVXS