Skip to content
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

Jump to bottom

fix blocking_observable::subscribe #453

Merged
merged 1 commit into from
Aug 5, 2018
Merged

fix blocking_observable::subscribe #453

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
45 changes: 7 additions & 38 deletions Rx/v2/src/rxcpp/rx-observable.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,28 +173,9 @@ class blocking_observable
-> void {
std::mutex lock;
std::condition_variable wake;
bool disposed = false;
std::exception_ptr error;

struct tracking
{
~tracking()
{
if (!disposed || !wakened) std::terminate();
}
tracking()
{
disposed = false;
wakened = false;
false_wakes = 0;
true_wakes = 0;
}
std::atomic_bool disposed;
std::atomic_bool wakened;
std::atomic_int false_wakes;
std::atomic_int true_wakes;
};
auto track = std::make_shared<tracking>();

auto dest = make_subscriber<T>(std::forward<ArgN>(an)...);

// keep any error to rethrow at the end.
Expand All @@ -213,31 +194,19 @@ class blocking_observable

auto cs = scbr.get_subscription();
cs.add(
[&, track](){
// OSX geting invalid x86 op if notify_one is after the disposed = true
// presumably because the condition_variable may already have been awakened
// and is now sitting in a while loop on disposed
[&](){
std::unique_lock<std::mutex> guard(lock);
wake.notify_one();
track->disposed = true;
disposed = true;
});

std::unique_lock<std::mutex> guard(lock);
source.subscribe(std::move(scbr));

std::unique_lock<std::mutex> guard(lock);
wake.wait(guard,
[&, track](){
// this is really not good.
// false wakeups were never followed by true wakeups so..

// anyways this gets triggered before disposed is set now so wait.
while (!track->disposed) {
++track->false_wakes;
}
++track->true_wakes;
return true;
[&](){
return disposed;
});
track->wakened = true;
if (!track->disposed || !track->wakened) std::terminate();

if (error) {std::rethrow_exception(error);}
}
Expand Down
6 changes: 3 additions & 3 deletions Rx/v2/test/operators/merge_delay_error.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ const int static_onnextcalls = 1000000;

//merge_delay_error must work the very same way as `merge()` except the error handling

SCENARIO("merge completes", "[merge][join][operators]"){
SCENARIO("merge_delay_error completes", "[merge][join][operators]"){
GIVEN("1 hot observable with 3 cold observables of ints."){
auto sc = rxsc::make_test();
auto w = sc.create_worker();
Expand Down Expand Up @@ -117,7 +117,7 @@ SCENARIO("merge completes", "[merge][join][operators]"){
}
}

SCENARIO("variadic merge completes with error", "[merge][join][operators]"){
SCENARIO("variadic merge_delay_error completes with error", "[merge][join][operators]"){
GIVEN("1 hot observable with 3 cold observables of ints."){
auto sc = rxsc::make_test();
auto w = sc.create_worker();
Expand Down Expand Up @@ -211,7 +211,7 @@ SCENARIO("variadic merge completes with error", "[merge][join][operators]"){
}
}

SCENARIO("variadic merge completes with 2 errors", "[merge][join][operators]"){
SCENARIO("variadic merge_delay_error completes with 2 errors", "[merge][join][operators]"){
GIVEN("1 hot observable with 3 cold observables of ints."){
auto sc = rxsc::make_test();
auto w = sc.create_worker();
Expand Down