Skip to content

simplify disposables in subjects #511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AlexInLog merged 1 commit into from
Jan 23, 2024
Merged

simplify disposables in subjects #511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions src/rpp/rpp/observables/observable.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ class observable
[[nodiscard("Use returned disposable or use subscribe(observer) instead")]] composite_disposable_wrapper subscribe_with_disposable(observer<Type, ObserverStrategy>&& observer) const
{
if (!observer.is_disposed())
return m_strategy.subscribe(rpp::composite_disposable_wrapper{std::make_shared<rpp::composite_disposable_impl<typename expected_disposable_strategy::disposable_container>>()}, std::move(observer));
return subscribe(rpp::composite_disposable_wrapper{std::make_shared<rpp::composite_disposable_impl<typename expected_disposable_strategy::disposable_container>>()}, std::move(observer));
return {};
}

Expand Down Expand Up @@ -264,10 +264,10 @@ class observable
composite_disposable_wrapper subscribe(const composite_disposable_wrapper& d, OnNext&& on_next, OnError&& on_error = {}, OnCompleted&& on_completed = {}) const
{
if (!d.is_disposed())
m_strategy.subscribe(make_lambda_observer<Type>(d,
std::forward<OnNext>(on_next),
std::forward<OnError>(on_error),
std::forward<OnCompleted>(on_completed)));
subscribe(make_lambda_observer<Type>(d,
std::forward<OnNext>(on_next),
std::forward<OnError>(on_error),
std::forward<OnCompleted>(on_completed)));
return d;
}

Expand Down
4 changes: 3 additions & 1 deletion src/rpp/rpp/operators/details/forwarding_subject.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ class forwarding_strategy
{
struct observer_strategy
{
using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;

std::shared_ptr<subjects::details::subject_state<Type>> state{};

void set_upstream(const disposable_wrapper& d) const noexcept { state->add(d); }
Expand All @@ -48,7 +50,7 @@ class forwarding_strategy

auto get_observer() const
{
return rpp::observer<Type, rpp::details::with_external_disposable<observer_strategy>>{composite_disposable_wrapper::from_shared(m_state), observer_strategy{m_state}};
return rpp::observer<Type, observer_strategy>{observer_strategy{m_state}};
}

template<rpp::constraint::observer_of_type<Type> TObs>
Expand Down
2 changes: 2 additions & 0 deletions src/rpp/rpp/subjects/details/subject_state.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,12 +79,14 @@ class subject_state : public std::enable_shared_from_this<subject_state<Type>>
{
if (const auto observers = exchange_observers_under_lock_if_there(err))
rpp::utils::for_each(*observers, [&](const auto& sub) { sub.on_error(err); });
dispose();
}

void on_completed()
{
if (const auto observers = exchange_observers_under_lock_if_there(completed{}))
rpp::utils::for_each(*observers, rpp::utils::static_mem_fn<&dynamic_observer<Type>::on_completed>{});
dispose();
}

private:
Expand Down
4 changes: 3 additions & 1 deletion src/rpp/rpp/subjects/publish_subject.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ class publish_strategy
{
struct observer_strategy
{
using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;

std::shared_ptr<subject_state<Type>> state{};

void set_upstream(const disposable_wrapper& d) const noexcept { state->add(d); }
Expand All @@ -44,7 +46,7 @@ class publish_strategy

auto get_observer() const
{
return rpp::observer<Type, rpp::details::with_external_disposable<observer_strategy>>{composite_disposable_wrapper{m_state}, observer_strategy{m_state}};
return rpp::observer<Type, observer_strategy>{observer_strategy{m_state}};
}

template<rpp::constraint::observer_of_type<Type> TObs>
Expand Down
4 changes: 3 additions & 1 deletion src/rpp/rpp/subjects/serialized_subject.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ class serialized_strategy

struct observer_strategy
{
using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;

std::shared_ptr<serialized_state> state{};

void set_upstream(const disposable_wrapper& d) const noexcept { state->add(d); }
Expand Down Expand Up @@ -64,7 +66,7 @@ class serialized_strategy

auto get_observer() const
{
return rpp::observer<Type, rpp::details::with_external_disposable<observer_strategy>>{get_disposable(), observer_strategy{m_state}};
return rpp::observer<Type, observer_strategy>{observer_strategy{m_state}};
}

template<rpp::constraint::observer_of_type<Type> TObs>
Expand Down