test_eventstream_nasty.cpp

#include "eventstream.h"

#include <cassert>
#include <cstddef>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <limits>
#include <optional>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

#define CHECK(cond) do { \
  if (!(cond)) { \
    std::cerr << "CHECK failed: " #cond "\n" \
              << "  at " << __FILE__ << ":" << __LINE__ << "\n"; \
    std::abort(); \
  } \
} while (0)

namespace {

using eventstream::control;

// Stan bez konstruktora domyślnego (częsty „wredny” przypadek).
struct NoDefaultState {
  int v;
  explicit NoDefaultState(int x) : v(x) {}
};

// Obserwator zliczający wywołania i (opcjonalnie) zatrzymujący po N eventach.
template <class Event>
auto make_collecting_observer(std::vector<Event>& out,
                              std::size_t stop_after = (std::numeric_limits<std::size_t>::max)())
{
  return [&out, stop_after, seen = std::size_t{0}](Event e, auto state) mutable
      -> std::pair<decltype(state), control>
  {
    out.push_back(e);
    ++seen;
    const control c = (seen >= stop_after) ? control::stop : control::continue_;
    return {state, c};
  };
}

// Obserwator, który modyfikuje stan, i pozwala testować poprawne „threading” stanu.
template <class Event>
auto make_stateful_observer(std::vector<Event>& out,
                            std::size_t stop_after,
                            int delta_per_event)
{
  return [&out, stop_after, delta_per_event, seen = std::size_t{0}](Event e, int state) mutable
      -> std::pair<int, control>
  {
    out.push_back(e);
    ++seen;
    state += delta_per_event;
    const control c = (seen >= stop_after) ? control::stop : control::continue_;
    return {state, c};
  };
}

// Strumień „pusty” (emituje 0 zdarzeń).
auto empty_stream()
{
  return [](auto /*observer*/, auto state) { return state; };
}

template <class State>
using BoxObs = std::function<std::pair<State, control>(int, State)>;

template <class State>
using BoxStream = std::function<State(BoxObs<State>, State)>;

template <class State, class Stream>
BoxStream<State> box_stream(Stream s)
{
  return BoxStream<State>(
    [s](BoxObs<State> obs, State st) mutable -> State {
      return s(obs, st);
    }
  );
}

template <class State>
BoxStream<State> box_empty_stream()
{
  return BoxStream<State>(
    [](BoxObs<State>, State st) -> State { return st; }
  );
}


// Pomocniczo: uruchom strumień i zbierz eventy do vectora, stan ignorowany.
template <class Stream, class State, class Event = int>
std::vector<Event> run_collect(Stream s, State init_state)
{
  std::vector<Event> out;
  auto obs = make_collecting_observer<Event>(out);
  (void)s(obs, init_state);
  return out;
}

} // namespace

// ------------------------------
// emit
// ------------------------------
static void test_emit_basic()
{
  auto s = eventstream::emit(7);
  std::vector<int> out;
  auto obs = make_stateful_observer<int>(out, 10, +1);

  int st = s(obs, 100);
  CHECK((out == std::vector<int>{7}));
  CHECK(st == 101);
}

static void test_emit_stop_is_respected()
{
  auto s = eventstream::emit(42);
  std::vector<int> out;

  auto obs = [&out](int e, int state) -> std::pair<int, control> {
    out.push_back(e);
    return {state + 1, control::stop};
  };

  int st = s(obs, 5);
  CHECK((out == std::vector<int>{42}));
  CHECK(st == 6);
}

// ------------------------------
// generate
// ------------------------------
static void test_generate_emits_init_even_if_step_nullopt()
{
  int step_calls = 0;
  auto s = eventstream::generate(
    123,
    [&step_calls](int) -> std::optional<int> {
      ++step_calls;
      return std::nullopt;
    }
  );

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  int st = s(obs, 0);
  CHECK((out == std::vector<int>{123}));
  CHECK(st == 0);
  // init jest emitowane bez wołania step przed nim; step może zostać wywołany dopiero „po init”.
  CHECK(step_calls == 1); // po init próbujemy ustalić kolejny element
}

static void test_generate_stop_on_init_means_no_step_call()
{
  int step_calls = 0;
  auto s = eventstream::generate(
    1,
    [&step_calls](int x) -> std::optional<int> {
      ++step_calls;
      return x + 1;
    }
  );

  std::vector<int> out;
  auto obs = [&out](int e, int st) -> std::pair<int, control> {
    out.push_back(e);
    return {st, control::stop};
  };

  int final_st = s(obs, 0);
  CHECK((out == std::vector<int>{1}));
  CHECK(final_st == 0);
  CHECK(step_calls == 0); // kluczowe: stop po init => nie wolno liczyć dalej
}

static void test_generate_threads_state_correctly()
{
  auto s = eventstream::generate(
    10,
    [](int x) -> std::optional<int> {
      if (x >= 13) return std::nullopt;
      return x + 1;
    }
  );

  std::vector<int> out;
  auto obs = make_stateful_observer<int>(out, 10, +2);

  int st = s(obs, 1);
  CHECK((out == std::vector<int>{10, 11, 12, 13}));
  // 4 eventy => +8
  CHECK(st == 9);
}

// ------------------------------
// counter
// ------------------------------
static void test_counter_copy_is_independent()
{
  auto c1 = eventstream::counter();
  auto c2 = c1; // kopia musi mieć niezależny licznik

  std::vector<int> a, b;
  auto obs_a = make_collecting_observer<int>(a, 3);
  auto obs_b = make_collecting_observer<int>(b, 2);

  (void)c1(obs_a, 0); // 1,2,3
  (void)c2(obs_b, 0); // 1,2

  CHECK((a == std::vector<int>{1,2,3}));
  CHECK((b == std::vector<int>{1,2}));
}

static void test_counter_same_object_multiple_runs_continue_counting()
{
  auto c = eventstream::counter();

  std::vector<int> first, second;
  auto obs1 = make_collecting_observer<int>(first, 3);
  auto obs2 = make_collecting_observer<int>(second, 2);

  (void)c(obs1, 0); // 1,2,3
  (void)c(obs2, 0); // powinno być 4,5 (kontynuacja)

  CHECK((first == std::vector<int>{1,2,3}));
  CHECK((second == std::vector<int>{4,5}));
}

static void test_counter_in_pipeline_copy_independence()
{
  auto s = eventstream::counter() | eventstream::map([](int x){ return x * 10; }) | eventstream::take(3);
  auto t = s; // kopia całego potoku => niezależne źródło liczb

  std::vector<int> a, b;
  auto obs_a = make_collecting_observer<int>(a);
  auto obs_b = make_collecting_observer<int>(b);

  (void)s(obs_a, 0);
  (void)t(obs_b, 0);

  CHECK((a == std::vector<int>{10,20,30}));
  CHECK((b == std::vector<int>{10,20,30}));
}

// ------------------------------
// map
// ------------------------------
static void test_map_transforms_events_and_threads_state()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 3) ? std::nullopt : std::optional<int>(x + 1); }
  );

  auto m = eventstream::map([](int x){ return x * x; }, s);

  std::vector<int> out;
  auto obs = make_stateful_observer<int>(out, 10, +1);

  int st = m(obs, 0);
  CHECK((out == std::vector<int>{1,4,9}));
  CHECK(st == 3);
}

static void test_map_must_not_call_f_after_downstream_stop()
{
  int f_calls = 0;

  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 10) ? std::nullopt : std::optional<int>(x + 1); }
  );

  auto m = eventstream::map([&](int x){ ++f_calls; return x; }, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out, 2); // stop po 2 eventach

  (void)m(obs, 0);
  CHECK((out == std::vector<int>{1,2}));
  CHECK(f_calls == 2); // kluczowe: brak „nadmiarowych” wywołań f
}

// ------------------------------
// filter
// ------------------------------
static void test_filter_passes_only_matching_events()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 6) ? std::nullopt : std::optional<int>(x + 1); }
  );

  auto f = eventstream::filter([](int x){ return x % 2 == 0; }, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)f(obs, 0);
  CHECK((out == std::vector<int>{2,4,6}));
}

static void test_filter_propagates_stop_from_downstream()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 100) ? std::nullopt : std::optional<int>(x + 1); }
  );

  int pred_calls = 0;
  auto f = eventstream::filter([&](int x){ ++pred_calls; return x % 2 == 1; }, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out, 3); // stop po 3 przekazanych nieparzystych

  (void)f(obs, 0);
  CHECK((out == std::vector<int>{1,3,5}));
  // Żeby uzyskać 1,3,5 trzeba przetworzyć 1..5 => pred wywołany 5 razy, a potem stop.
  CHECK(pred_calls == 5);
}

static void test_filter_never_calls_downstream_for_rejected_events()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 5) ? std::nullopt : std::optional<int>(x + 1); }
  );

  int downstream_calls = 0;
  auto f = eventstream::filter([](int){ return false; }, s);

  auto obs = [&](int /*e*/, int st) -> std::pair<int, control> {
    ++downstream_calls;
    return {st, control::continue_};
  };

  (void)f(obs, 0);
  CHECK(downstream_calls == 0);
}

// ------------------------------
// take
// ------------------------------
static void test_take_basic()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 10) ? std::nullopt : std::optional<int>(x + 1); }
  );

  auto t = eventstream::take(3, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)t(obs, 0);
  CHECK((out == std::vector<int>{1,2,3}));
}

static void test_take_must_stop_upstream_even_if_downstream_continue()
{
  int step_calls = 0;
  auto s = eventstream::generate(
    1,
    [&](int x) -> std::optional<int> {
      ++step_calls;
      return x + 1;
    }
  );

  auto t = eventstream::take(1, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)t(obs, 0);
  CHECK((out == std::vector<int>{1}));
  CHECK(step_calls == 0); // po 1. evencie take powinien zakończyć pobieranie => step nie może się wykonać
}

static void test_take_zero_must_not_touch_source()
{
  int touched = 0;
  auto s = eventstream::generate(
    1,
    [&](int x) -> std::optional<int> {
      ++touched;
      return x + 1;
    }
  );

  auto t = eventstream::take(0, s);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)t(obs, 123);
  CHECK(out.empty());
  CHECK(touched == 0); // wredny warunek: take(0) nie powinno w ogóle uruchamiać źródła
}

static void test_take_propagates_downstream_stop_earlier_than_n()
{
  auto s = eventstream::generate(
    1,
    [](int x) -> std::optional<int> { return (x >= 10) ? std::nullopt : std::optional<int>(x + 1); }
  );

  auto t = eventstream::take(10, s); // duże n

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out, 2); // stop po 2

  (void)t(obs, 0);
  CHECK((out == std::vector<int>{1,2}));
}

// ------------------------------
// operator| i tap
// ------------------------------
static void test_pipe_map_filter_take()
{
  auto s =
      eventstream::generate(1, [](int x) -> std::optional<int> {
        if (x >= 10) return std::nullopt;
        return x + 1;
      })
      | eventstream::map([](int x){ return x * 2; })
      | eventstream::filter([](int x){ return x % 4 == 0; })
      | eventstream::take(3);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)s(obs, 0);
  CHECK((out == std::vector<int>{4,8,12}));
}

static void test_tap_side_effect_and_control_transparency()
{
  int side = 0;
  auto s =
      eventstream::generate(1, [](int x) -> std::optional<int> {
        if (x >= 100) return std::nullopt;
        return x + 1;
      })
      | eventstream::tap([&](int){ ++side; })
      | eventstream::take(5);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)s(obs, 0);
  CHECK((out == std::vector<int>{1,2,3,4,5}));
  CHECK(side == 5);
}

// ------------------------------
// flatten
// ------------------------------
static void test_flatten_basic_order_and_state_threading()
{
  using St = NoDefaultState;
  using Inner = BoxStream<St>;

  // Inner strumienie: [1,2], [10], []
  auto s1 = eventstream::generate(1, [](int x) -> std::optional<int> {
    if (x >= 2) return std::nullopt;
    return x + 1;
  });
  auto s2 = eventstream::emit(10);
  auto s3 = empty_stream();

  // Zbijamy wszystkie do jednego typu (Inner = std::function<St(BoxObs<St>,St)>)
  std::vector<Inner> inners = {
    box_stream<St>(s1),
    box_stream<St>(s2),
    box_stream<St>(s3) // albo box_empty_stream<St>()
  };

  // Strumień strumieni: emituj kolejno inners[0], inners[1], inners[2]
  auto ss = eventstream::generate(
    inners[0],
    [inners, idx = std::size_t{0}](Inner) mutable -> std::optional<Inner> {
      ++idx;
      if (idx < inners.size()) return inners[idx];
      return std::nullopt;
    }
  );

  auto flat = eventstream::flatten(ss);

  std::vector<int> out;

  BoxObs<St> obs = [&](int e, St st) -> std::pair<St, control> {
    out.push_back(e);
    st.v += 1;
    return {st, control::continue_};
  };

  St init_state{7};
  auto final_state = flat(obs, init_state);

  CHECK((out == std::vector<int>{1, 2, 10})); // s3 puste => nic
  CHECK(final_state.v == 10);                 // 3 eventy => +3
}


static void test_flatten_must_stop_outer_and_inner_on_downstream_stop()
{
  int outer_step_calls = 0;
  int inner_step_calls = 0;

  // Inner: 1,2,3,... (długi)
  auto inner = eventstream::generate(
    1,
    [&](int x) -> std::optional<int> {
      ++inner_step_calls;
      return x + 1;
    }
  );

  // Outer: emituje w nieskończoność kolejne inner-y (tu: generujemy indeksy 0,1,2,...)
  auto outer = eventstream::generate(
    0,
    [&](int i) -> std::optional<int> {
      ++outer_step_calls;
      return i + 1;
    }
  );

  auto ss = eventstream::map([&](int){ return inner; }, outer);
  auto flat = eventstream::flatten(ss);

  // Zatrzymujemy po 1 evencie globalnie.
  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out, 1);

  (void)eventstream::take(1, flat)(obs, 0);

  CHECK((out == std::vector<int>{1}));
  CHECK(outer_step_calls == 0); // w ogóle nie wolno przejść do kolejnego inner-a
  CHECK(inner_step_calls == 0); // po pierwszym evencie inner nie powinien próbować produkować kolejnego
}

static void test_flatten_stop_mid_inner_preserves_state_returned()
{
  // Inner: 5,6,7
  auto inner = eventstream::generate(
    5,
    [](int x) -> std::optional<int> {
      if (x >= 7) return std::nullopt;
      return x + 1;
    }
  );

  // Outer emituje dwa identyczne inner-y.
  auto ss = eventstream::generate(
    inner,
    [inner, emitted = false](auto) mutable -> std::optional<decltype(inner)> {
      if (emitted) return std::nullopt;
      emitted = true;
      return inner;
    }
  );

  auto flat = eventstream::flatten(ss);

  // Obserwator zatrzymuje się po 2 eventach łącznie: ma przerwać w środku pierwszego inner-a
  int seen = 0;
  int st = flat(
    [&](int e, int state) -> std::pair<int, control> {
      (void)e;
      ++seen;
      state += 10;
      return {state, (seen >= 2) ? control::stop : control::continue_};
    },
    0
  );

  CHECK(seen == 2);
  CHECK(st == 20);
}

// ------------------------------
// memoize
// ------------------------------
static int g_calls_str = 0;
static int f_str_len_by_value(std::string s)
{
  ++g_calls_str;
  return static_cast<int>(s.size());
}

static int g_calls_two = 0;
static std::string f_join(std::string a, const std::string& b)
{
  ++g_calls_two;
  return a + "|" + b;
}

static int g_calls_ref = 0;
static std::vector<int> g_store(10, 0);
static int& f_ref_at(int i)
{
  ++g_calls_ref;
  return g_store.at(static_cast<std::size_t>(i));
}

static void test_memoize_basic_caches_result()
{
  g_calls_str = 0;
  auto m = eventstream::memoize(&f_str_len_by_value);

  int a = m(std::string("abcd"));
  int b = m(std::string("abcd"));
  int c = m("abcd"); // niejawna konwersja const char* -> std::string

  CHECK(a == 4 && b == 4 && c == 4);
  CHECK(g_calls_str == 1); // kluczowe: jedno realne wywołanie
}

static void test_memoize_multiple_args_and_conversions()
{
  g_calls_two = 0;
  auto m = eventstream::memoize(&f_join);

  auto r1 = m(std::string("A"), std::string("B"));
  auto r2 = m("A", "B"); // obie konwersje
  auto r3 = m("A", std::string("B"));

  CHECK(r1 == "A|B");
  CHECK(r2 == "A|B");
  CHECK(r3 == "A|B");
  CHECK(g_calls_two == 1);
}

static void test_memoize_copy_has_independent_cache()
{
  g_calls_str = 0;
  auto m1 = eventstream::memoize(&f_str_len_by_value);
  auto m2 = m1; // kopia wrappera musi mieć własny cache

  (void)m1("xyz");
  CHECK(g_calls_str == 1);

  (void)m1("xyz");
  CHECK(g_calls_str == 1); // cache m1

  (void)m2("xyz");
  CHECK(g_calls_str == 2); // osobny cache => drugie realne wywołanie
}

static void test_memoize_reference_return_same_object()
{
  g_calls_ref = 0;
  g_store.assign(10, 0);

  auto m = eventstream::memoize(&f_ref_at);

  int& a = m(3);
  a = 123;
  int& b = m(3);

  CHECK(&a == &b);         // musi zwracać referencję do tego samego obiektu
  CHECK(b == 123);         // potwierdzenie, że to ten sam element
  CHECK(g_calls_ref == 1); // realnie policzone tylko raz
}

static void test_memoize_reference_different_keys_different_refs()
{
  g_calls_ref = 0;
  g_store.assign(10, 0);

  auto m = eventstream::memoize(&f_ref_at);

  int& a = m(1);
  int& b = m(2);

  CHECK(&a != &b);
  CHECK(g_calls_ref == 2);

  // powtórki nie zwiększają licznika
  (void)m(1);
  (void)m(2);
  CHECK(g_calls_ref == 2);
}

// ------------------------------
// Dodatkowe „wredne” integracyjne
// ------------------------------
static void test_big_state_type_not_default_constructed()
{
  // Sprawdza, czy implementacje nie zakładają, że State da się domyślnie skonstruować.
  auto s =
      eventstream::emit(std::string("x"))
      | eventstream::map([](const std::string& v){ return v + "y"; })
      | eventstream::tap([](const std::string&){})
      | eventstream::take(1);

  NoDefaultState init{99};
  std::vector<std::string> out;

  auto final_state = s(
    [&](std::string e, NoDefaultState st) -> std::pair<NoDefaultState, control> {
      out.push_back(std::move(e));
      st.v += 1;
      return {st, control::continue_};
    },
    init
  );

  CHECK((out == std::vector<std::string>{"xy"}));
  CHECK(final_state.v == 100);
}

static void test_filter_then_take_must_stop_source_early()
{
  // Źródło: 1..100 (ale chcemy wziąć tylko 3 nieparzyste => 1,3,5 => do przetworzenia 1..5)
  int step_calls = 0;
  auto src = eventstream::generate(
    1,
    [&](int x) -> std::optional<int> {
      ++step_calls;
      return x + 1;
    }
  );

  auto s = src
           | eventstream::filter([](int x){ return x % 2 == 1; })
           | eventstream::take(3);

  std::vector<int> out;
  auto obs = make_collecting_observer<int>(out);

  (void)s(obs, 0);

  CHECK((out == std::vector<int>{1,3,5}));
  CHECK(step_calls == 4); // po init=1: step dla 2,3,4,5 => 4 wywołania, i stop
}


int main()
{
  test_emit_basic();
  test_emit_stop_is_respected();

  test_generate_emits_init_even_if_step_nullopt();
  test_generate_stop_on_init_means_no_step_call();
  test_generate_threads_state_correctly();

  test_counter_copy_is_independent();
  test_counter_same_object_multiple_runs_continue_counting();
  test_counter_in_pipeline_copy_independence();

  test_map_transforms_events_and_threads_state();
  test_map_must_not_call_f_after_downstream_stop();

  test_filter_passes_only_matching_events();
  test_filter_propagates_stop_from_downstream();
  test_filter_never_calls_downstream_for_rejected_events();

  test_take_basic();
  test_take_must_stop_upstream_even_if_downstream_continue();
  test_take_zero_must_not_touch_source();
  test_take_propagates_downstream_stop_earlier_than_n();

  test_pipe_map_filter_take();
  test_tap_side_effect_and_control_transparency();

  test_flatten_basic_order_and_state_threading();
  test_flatten_must_stop_outer_and_inner_on_downstream_stop();
  test_flatten_stop_mid_inner_preserves_state_returned();

  test_memoize_basic_caches_result();
  test_memoize_multiple_args_and_conversions();
  test_memoize_copy_has_independent_cache();
  test_memoize_reference_return_same_object();
  test_memoize_reference_different_keys_different_refs();

  test_big_state_type_not_default_constructed();
  test_filter_then_take_must_stop_source_early();

  std::cerr << "All nasty tests passed.\n";
  return 0;
}