A toy C++ FRP framework

toy_frp.cpp

// A toy C++ FRP framework.
// Author: Bruce Li (github.com/kkspeed)
// Licensed under BSD.
//
// Note:
// 1. Compiles with C++ 17
// 2. Signal values are copy-constructable. Non-copyconstructable values
//    could be wrapped in shared_ptr.
// 3. Signal graph is statically connected and unsubscription is not
//    an option.

#include <cassert>
#include <functional>
#include <iostream>
#include <memory>
#include <vector>

namespace naive {

namespace {

template <typename T>
class SignalCore {
 public:
  void add_observer(const std::function<void()> &func) {
    observers_.push_back(func);
  }

  void notify() const {
    if (!has_value()) return;
    for (const auto &f : observers_) {
      f();
    }
  }

  bool has_value() const { return value_.has_value(); }
  const T &value() const { return value_.value(); }
  void set_value(const T &value) { value_.emplace(value); }
  void drop_value() { value_.reset(); }

 private:
  std::optional<T> value_;
  std::vector<std::function<void()>> observers_;
};

}  // namespace

// Represents a stream of event of type |T|. See main function for
// detailed usage example.
template <typename T>
class Signal {
 public:
  Signal() : core_(std::make_shared<SignalCore<T>>()) {}

  // Send |value| into signal and activate downstream listeners.
  void Activate(const T &value) {
    core_->set_value(value);
    core_->notify();
    core_->drop_value();
  }

  // Subscribe |f| to current signal.
  template <typename F>
  void Sink(F &&f) {
    auto this_core = core_;
    core_->add_observer(
        [this_core, func = std::move(f)]() { func(this_core->value()); });
  }

  // Creates a new signal with by applying |f| to each element of current
  // signal.
  template <typename F>
  auto Map(F &&f) -> Signal<std::invoke_result_t<F, T>> {
    Signal<std::invoke_result_t<F, T>> signal;
    auto this_core = core_;
    auto new_core = signal.core();
    core_->add_observer([this_core, new_core, func = std::move(f)]() {
      new_core->set_value(func(this_core->value()));
      new_core->notify();
      new_core->drop_value();
    });
    return signal;
  }

  // Reduces over current signal and returns a new signal with accumulated
  // value.
  template <typename F, typename B>
  auto Fold(F &&f, B init) -> Signal<std::invoke_result_t<F, B, T>> {
    Signal<std::invoke_result_t<F, B, T>> signal;
    auto state = std::make_shared<std::optional<B>>();
    auto this_core = core_;
    auto new_core = signal.core();
    core_->add_observer(
        [state, this_core, new_core, func = std::move(f), init]() {
          if (!state->has_value()) {
            state->emplace(init);
          }
          state->emplace(func(state->value(), this_core->value()));
          new_core->set_value(state->value());
          new_core->notify();
          new_core->drop_value();
        });
    return signal;
  }

  // Filters current signal and fires only when predicate |f| is true.
  template <typename F>
  Signal<T> Filter(F &&f) {
    Signal<T> signal;
    auto this_core = core_;
    auto new_core = signal.core_;
    core_->add_observer([this_core, new_core, func = std::move(f)]() {
      if (!func(this_core->value())) {
        new_core->drop_value();
        return;
      }
      new_core->set_value(this_core->value());
      new_core->notify();
      new_core->drop_value();
    });
    return signal;
  }

  // Merge current signal and |other|. Fires when each of them fires.
  Signal<T> Merge(Signal<T> &other) {
    Signal<T> signal;
    auto new_core = signal.core_;
    auto this_core = core_;
    this_core->add_observer([this_core, new_core]() {
      new_core->set_value(this_core->value());
      new_core->notify();
      new_core->drop_value();
    });

    auto other_core = other.core_;
    other_core->add_observer([other_core, new_core]() {
      new_core->set_value(other_core->value());
      new_core->notify();
      new_core->drop_value();
    });
    return signal;
  }

  // Sync current signal and |other|. Fires when both of them fires and returns
  // a signal with a pair.
  template <typename B>
  Signal<std::pair<T, B>> Sync(Signal<B> &other) {
    Signal<std::pair<T, B>> signal;
    auto state =
        std::make_shared<std::pair<std::optional<T>, std::optional<B>>>();
    auto new_core = signal.core();
    auto this_core = core();
    this_core->add_observer([state, this_core, new_core]() {
      state->first.emplace(this_core->value());
      if (state->second.has_value()) {
        new_core->set_value(
            std::make_pair(state->first.value(), state->second.value()));
        new_core->notify();
        new_core->drop_value();
        state->first.reset();
        state->second.reset();
      }
    });
    auto other_core = other.core();
    other_core->add_observer([state, other_core, new_core]() {
      state->second.emplace(other_core->value());
      if (state->first.has_value()) {
        new_core->set_value(
            std::make_pair(state->first.value(), state->second.value()));
        new_core->notify();
        new_core->drop_value();
        state->first.reset();
        state->second.reset();
      }
    });
    return signal;
  }

  std::shared_ptr<SignalCore<T>> core() { return core_; }

 private:
  std::shared_ptr<SignalCore<T>> core_;
};

}  // namespace naive

int main() {
  {
    std::cout << "test basic...";

    std::vector<int> result;
    naive::Signal<int> signal;

    signal.Map([](int x) { return x * 2; }).Sink([&result](int x) {
      result.push_back(x);
    });

    signal.Activate(10);
    assert(result.back() == 20);
    signal.Activate(20);
    assert(result.back() == 40);

    std::cout << "[OK]" << std::endl;
  }

  {
    std::cout << "test filter...";

    std::vector<int> result;
    naive::Signal<int> signal;

    signal.Filter([](int x) { return x > 3; }).Sink([&result](int x) {
      result.push_back(x);
    });

    signal.Activate(3);
    signal.Activate(10);
    signal.Activate(2);
    signal.Activate(6);
    signal.Activate(3);
    assert(result.size() == 2);
    assert(result[0] == 10);
    assert(result[1] == 6);

    std::cout << "[OK]" << std::endl;
  }

  {
    std::cout << "test fold...";

    std::vector<int> result;
    naive::Signal<int> signal;

    signal.Fold([](int x, int y) { return x + y; }, 0).Sink([&result](int x) {
      result.push_back(x);
    });

    signal.Activate(1);
    signal.Activate(2);
    signal.Activate(3);
    signal.Activate(4);
    signal.Activate(5);
    assert(result.size() == 5);
    assert(result[0] == 1);
    assert(result[1] == 3);
    assert(result[2] == 6);
    assert(result[3] == 10);
    assert(result[4] == 15);

    std::cout << "[OK]" << std::endl;
  }

  {
    std::cout << "test merge...";

    std::vector<int> result;
    naive::Signal<int> signal1;
    naive::Signal<int> signal2;
    naive::Signal<int> signal3 = signal1.Merge(signal2);

    signal3.Sink([&result](int x) { result.push_back(x); });

    signal1.Activate(1);
    signal1.Activate(1);
    signal2.Activate(2);
    signal1.Activate(3);
    signal2.Activate(5);
    signal2.Activate(8);
    signal1.Activate(13);

    assert(result.size() == 7);
    assert(result[0] == 1);
    assert(result[1] == 1);
    assert(result[2] == 2);
    assert(result[3] == 3);
    assert(result[4] == 5);
    assert(result[5] == 8);
    assert(result[6] == 13);

    std::cout << "[OK]" << std::endl;
  }

  {
    std::cout << "test sync...";

    std::vector<std::pair<int, bool>> result;
    naive::Signal<int> signal1;
    naive::Signal<bool> signal2;
    naive::Signal<std::pair<int, bool>> signal3 = signal1.Sync(signal2);

    signal3.Sink([&result](auto x) { result.push_back(x); });

    signal1.Activate(1);
    signal1.Activate(1);
    signal2.Activate(true);
    signal2.Activate(false);
    signal1.Activate(3);
    signal2.Activate(true);
    signal2.Activate(true);

    assert(result.size() == 2);
    assert(result[0] == std::make_pair(1, true));
    assert(result[1] == std::make_pair(3, false));

    std::cout << "[OK]" << std::endl;
  }

  {
    std::cout << "test computing fibonacci with fold and map...";
    std::vector<int> result;
    std::vector<int> expected{1, 1, 2, 3, 5, 8, 13};
    naive::Signal<int> signal;
    signal
        .Fold([](auto p,
                 auto) { return std::make_pair(p.second, p.first + p.second); },
              std::make_pair(0, 1))
        .Map([](auto p) { return p.first; })
        .Sink([&result](int i) { result.push_back(i); });

    // The activation value does not care in this case.
    signal.Activate(10);
    signal.Activate(10);
    signal.Activate(10);
    signal.Activate(10);
    signal.Activate(10);
    signal.Activate(10);
    signal.Activate(10);

    assert(result.size() == expected.size());
    for (size_t i = 0; i < result.size(); i++) {
      assert(result[i] == expected[i]);
    }

    std::cout << "[OK]" << std::endl;
  }

  std::cout << "All tests passed!" << std::endl;
  return 0;
}