test.cpp

#include <iostream>
#include <memory>

#include <rxcpp/rx.hpp>


struct Abcd {
    Abcd() {
        printf("Abcd\n");
    }

    ~Abcd() {
        printf("~Abcd\n");
    }
};

template<typename A>
struct Context1 {
    typedef ::rxcpp::observable<A> observable_type;

    Context1(int a, int x) : a(a), x(x) {

    }


    int a;
    int x;
};

template<typename A>
struct Context2 {
    typedef ::rxcpp::observable<A> observable_type;

    Context2(observable_type &&observable): observable(::std::move(observable)) {

    }

    Context2(const observable_type &observable): observable(observable) {

    }

    observable_type observable;

    const observable_type &get() const {
        return observable;
    }
};

template<typename A>
::rxcpp::observable<A> flat_map(const ::rxcpp::observable<Context2<A>> &a) {
    return a.flat_map(
            [](const auto &a) { return a.get(); },
            [](const auto &a, const auto &b) { return b; }
    );
};

template<typename Type>
::rxcpp::observable<Context2<Type>> abcd() {

    auto hook = ::std::make_shared<Abcd>();
    return ::rxcpp::observable<>::create<Context1<Type>>(
            [](::rxcpp::subscriber<Context1<Type>> s) {
                s.on_next(Context1<Type>(1, 0));
                s.on_next(Context1<Type>(2, 20));
                s.on_completed(); // comment for triggering the issue
            })
            .group_by(
            [](const auto &a) {
                return a.a;
            },
            [](const auto &a) {
                return a.x;
            }
    ).map([hook](const auto &a) {
        return Context2<Type>(a);
    });
}

int main() {
    std::cout << "Hello, World!" << std::endl;

    auto sub = flat_map<int>(abcd<int>()).subscribe([](const auto &a) {
        printf("Value: %d\n", a);
    });

    std::cout << "Unsubscribe" << std::endl;
    sub.unsubscribe();

    std::cout << "End of World!" << std::endl;

    return 0;
}