Skip to content

Commit 286ae1c

Browse files
terukazu-inoueterukazu-inoue
authored
0.3.10: Merge pull request #13 from CODIANZ/development
adopted `stream_controller` for `inflow_restriction`.
2 parents 71caa0d + 79700b6 commit 286ae1c

3 files changed

Lines changed: 195 additions & 80 deletions

File tree

‎include/another-rxcpp/utils/inflow_restriction.h‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
#define __another_rxcpp_h_inflow_restriction__
33

44
#include "../observable.h"
5+
#include "../internal/tools/stream_controller.h"
56
#include "sem.h"
67

78
namespace another_rxcpp {
@@ -24,20 +25,28 @@ class inflow_restriction {
2425
{
2526
auto sem = sem_;
2627
return observable<>::create<T>([sem, o](subscriber<T> s){
28+
auto sctl = internal::stream_controller<T>(s);
29+
sctl.set_on_finalize([sem](){
30+
sem->unlock();
31+
});
2732
sem->lock();
28-
o.subscribe({
29-
[s](const T& v){
30-
s.on_next(v);
33+
if(!sctl.is_subscribed()) {
34+
sem->unlock();
35+
return;
36+
}
37+
o.subscribe(sctl.template new_observer<T>(
38+
[sctl](auto, const T& v){
39+
sctl.sink_next(v);
3140
},
32-
[s, sem](std::exception_ptr e){
33-
s.on_error(e);
41+
[sctl, sem](auto, std::exception_ptr e){
42+
sctl.sink_error(e);
3443
sem->unlock();
3544
},
36-
[s, sem](){
37-
s.on_completed();
45+
[sctl, sem](auto serial){
46+
sctl.sink_completed(serial);
3847
sem->unlock();
3948
}
40-
});
49+
));
4150
});
4251
}
4352
};

‎test/case_7.cpp‎

Lines changed: 118 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -8,81 +8,146 @@
88
using namespace another_rxcpp;
99
using namespace another_rxcpp::operators;
1010

11-
void test_case_7() {
12-
log() << "test_case_7 -- begin" << std::endl;
11+
struct watier : public std::enable_shared_from_this<watier> {
12+
subjects::subject<int> breaker_subject;
1313

14-
{
15-
log() << "#1 normal observer -- begin" << std::endl;
16-
auto sbsc = observable<>::create<int>([](subscriber<int> s){
17-
subjects::subject<int> sbj;
18-
observables::range(0, 10)
19-
.observe_on(schedulers::observe_on_new_thread())
20-
.take_until(sbj.as_observable())
14+
virtual ~watier() {
15+
log() << "~watier" << std::endl;
16+
}
17+
18+
auto wait_subscribe(std::string memo){
19+
auto THIS = shared_from_this();
20+
return observable<>::create<int>([THIS, memo](subscriber<int> s){
21+
observables::interval(std::chrono::milliseconds(700), schedulers::new_thread_scheduler())
22+
.take_until(THIS->breaker_subject.as_observable())
2123
.subscribe(
22-
[s, sbj](auto x) {
23-
log() << "#1 inner next: " << x << std::endl;
24-
if(x == 5){
25-
sbj.as_subscriber().on_next(1);
26-
sbj.as_subscriber().on_completed();
27-
}
24+
[s, memo](auto x) {
25+
log() << memo << " next: " << x << std::endl;
2826
},
29-
[s](auto err) {
30-
log() << "#1 inner error" << std::endl;
27+
[s, memo](auto err) {
28+
log() << memo << " error" << std::endl;
3129
s.on_error(err);
3230
},
33-
[s]{
34-
log() << "#1 inner completed" << std::endl;
31+
[s, memo]{
32+
log() << memo << " completed" << std::endl;
3533
s.on_next(999);
3634
s.on_completed();
3735
}
3836
);
39-
})
40-
.subscribe(
41-
[](auto x) { log() << "#1 outer next: " << x << std::endl; },
42-
[](auto) { log() << "#1 outer error" << std::endl; },
43-
[] { log() << "#1 outer completed" << std::endl; }
44-
);
45-
while(sbsc.is_subscribed()) {}
46-
log() << "#1 normal observer -- end" << std::endl;
37+
});
4738
}
4839

49-
{
50-
log() << "#2 stream_controller -- begin" << std::endl;
51-
52-
auto sbsc = observable<>::create<int>([](subscriber<int> s){
40+
auto wait_stream_controller(std::string memo){
41+
auto THIS = shared_from_this();
42+
return observable<>::create<int>([THIS, memo](subscriber<int> s){
5343
internal::stream_controller<int> sctl(s);
54-
subjects::subject<int> sbj;
55-
observables::range(0, 10)
56-
.observe_on(schedulers::observe_on_new_thread())
57-
.take_until(sbj.as_observable())
44+
observables::interval(std::chrono::milliseconds(700), schedulers::new_thread_scheduler())
45+
.take_until(THIS->breaker_subject.as_observable())
5846
.subscribe(sctl.new_observer<int>(
59-
[sctl, sbj](auto, auto x) {
60-
log() << "#2 inner next: " << x << std::endl;
61-
if(x == 5){
62-
sbj.as_subscriber().on_next(1);
63-
sbj.as_subscriber().on_completed();
64-
}
47+
[sctl, memo](auto, auto x) {
48+
log() << memo << " next: " << x << std::endl;
6549
},
66-
[sctl](auto, auto err) {
67-
log() << "#2 inner error" << std::endl;
50+
[sctl, memo](auto, auto err) {
51+
log() << memo << " error" << std::endl;
6852
sctl.sink_error(err);
6953
},
70-
[sctl](auto serial){
71-
log() << "#2 inner completed" << std::endl;
54+
[sctl, memo](auto serial){
55+
log() << memo << " completed" << std::endl;
7256
sctl.sink_next(999);
7357
sctl.sink_completed(serial);
7458
}
7559
));
76-
})
77-
.subscribe(
78-
[](auto x) { log() << "#2 outer next: " << x << std::endl; },
79-
[](auto) { log() << "#2 outer error" << std::endl; },
80-
[] { log() << "#2 outer completed" << std::endl; }
81-
);
82-
while(sbsc.is_subscribed()) {}
60+
});
61+
}
62+
};
63+
64+
auto emit_error(int ms){
65+
return observables::just(1)
66+
.observe_on(schedulers::new_thread_scheduler())
67+
.flat_map([ms](auto){
68+
return observable<>::create<int>([ms](subscriber<int> s){
69+
s.on_next(1);
70+
std::this_thread::sleep_for(std::chrono::milliseconds(ms));
71+
log() << "emit_error" << std::endl;
72+
s.on_error(std::make_exception_ptr(std::runtime_error("error")));
73+
});
74+
});
75+
}
76+
77+
void test_case_7() {
78+
log() << "test_case_7 -- begin" << std::endl;
79+
80+
{
81+
log() << "#1 -- begin" << std::endl;
82+
{
83+
auto w = std::make_shared<watier>();
84+
auto sbsc = emit_error(1000)
85+
.flat_map([w](auto){
86+
return w->wait_subscribe("#1 watier1")
87+
.amb(schedulers::new_thread_scheduler(), w->wait_subscribe("#1 watier2"));
88+
})
89+
.on_error_resume_next([w](auto){
90+
log() << "#1 on_error_resume_next" << std::endl;
91+
std::thread([w]{
92+
wait(100);
93+
log() << "#1 break!" << std::endl;
94+
w->breaker_subject.as_subscriber().on_next(1);
95+
}).detach();
96+
return observables::just(123456);
97+
})
98+
.subscribe(
99+
[](auto x) {
100+
log() << "#1 next: " << x << std::endl;
101+
},
102+
[](auto err) {
103+
log() << "#1 error" << std::endl;
104+
},
105+
[]{
106+
log() << "#1 completed" << std::endl;
107+
}
108+
);
109+
while(sbsc.is_subscribed()) {}
110+
}
111+
log() << "#1-- end" << std::endl;
112+
}
113+
114+
wait(1000);
83115

84-
log() << "#2 stream_controller -- end" << std::endl;
116+
{
117+
log() << "#2 -- begin" << std::endl;
118+
{
119+
auto w = std::make_shared<watier>();
120+
auto sbsc = emit_error(1000)
121+
.flat_map([w](auto){
122+
return w->wait_stream_controller("#2 watier1")
123+
.amb(schedulers::new_thread_scheduler(), w->wait_stream_controller("#2 watier2"));
124+
})
125+
.on_error_resume_next([w](auto){
126+
log() << "#2 on_error_resume_next" << std::endl;
127+
std::thread([w]{
128+
wait(100);
129+
log() << "#2 break!" << std::endl;
130+
w->breaker_subject.as_subscriber().on_next(1);
131+
}).detach();
132+
return observables::just(123456);
133+
})
134+
.subscribe(
135+
[](auto x) {
136+
log() << "#2 next: " << x << std::endl;
137+
},
138+
[](auto err) {
139+
log() << "#2 error" << std::endl;
140+
},
141+
[]{
142+
log() << "#2 completed" << std::endl;
143+
}
144+
);
145+
while(sbsc.is_subscribed()) {}
146+
}
147+
log() << "#2-- end" << std::endl;
85148
}
86149

150+
wait(1000);
151+
87152
log() << "test_case_7 -- end" << std::endl << std::endl;
88153
}

‎test/inflow_restriction.cpp‎

Lines changed: 60 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -21,22 +21,25 @@ void test_inflow_restriction() {
2121

2222
enum class result { success, failure };
2323
struct long_api {
24-
std::mutex mtx_;
25-
int count_ = 0;
24+
std::shared_ptr<std::mutex> mtx_ = std::make_shared<std::mutex>();
25+
std::shared_ptr<int> count_ = std::make_shared<int>(0);
2626
observable<result> call() {
27+
auto mtx = mtx_;
28+
auto count = count_;
2729
return observables::just(unit{}, new_thread_scheduler())
28-
| tap([=](unit){
29-
std::lock_guard<std::mutex> lock(mtx_);
30-
const int x = count_++;
30+
| tap([mtx, count](unit){
31+
std::lock_guard<std::mutex> lock(*mtx);
32+
(*count)++;
33+
const int x = (*count)++;
3134
std::cout << std::this_thread::get_id() << " : enter #" << x << std::endl;
3235
})
3336
| delay(std::chrono::seconds(1), new_thread_scheduler())
34-
| map([=](unit){
37+
| map([](unit){
3538
return result::success;
3639
})
37-
| tap([=](result){
38-
std::lock_guard<std::mutex> lock(mtx_);
39-
const int x = count_--;
40+
| tap([mtx, count](result){
41+
std::lock_guard<std::mutex> lock(*mtx);
42+
const int x = (*count)--;
4043
std::cout << std::this_thread::get_id() << " : leave #" << x << std::endl;
4144
});
4245
}
@@ -50,23 +53,23 @@ void test_inflow_restriction() {
5053

5154
auto sbsc = (
5255
observables::iterate(list)
53-
| flat_map([=](int n){
56+
| flat_map([api](int n){
5457
return api->call()
55-
| map([=](result){
58+
| map([n](result){
5659
return n;
5760
});
5861
})
5962
)
6063
.subscribe({
61-
[=](const int& x){
64+
[mtx](const int& x){
6265
std::lock_guard<std::mutex> lock(*mtx);
6366
log() << "next " << x << std::endl;
6467
},
65-
[=](std::exception_ptr){
68+
[mtx](std::exception_ptr){
6669
std::lock_guard<std::mutex> lock(*mtx);
6770
log() << "error " << std::endl;
6871
},
69-
[=](){
72+
[mtx](){
7073
std::lock_guard<std::mutex> lock(*mtx);
7174
log() << "completed " << std::endl;
7275
}
@@ -84,23 +87,61 @@ void test_inflow_restriction() {
8487

8588
auto sbsc = (
8689
observables::iterate(list)
87-
| flat_map([=](int n){
90+
| flat_map([ifr, api](int n){
91+
return ifr->enter(api->call())
92+
| map([n](result){
93+
return n;
94+
});
95+
})
96+
)
97+
.subscribe({
98+
[mtx](const int& x){
99+
std::lock_guard<std::mutex> lock(*mtx);
100+
log() << "next " << x << std::endl;
101+
},
102+
[mtx](std::exception_ptr){
103+
std::lock_guard<std::mutex> lock(*mtx);
104+
log() << "error " << std::endl;
105+
},
106+
[mtx](){
107+
std::lock_guard<std::mutex> lock(*mtx);
108+
log() << "completed " << std::endl;
109+
}
110+
});
111+
112+
while(sbsc.is_subscribed()) {}
113+
}
114+
115+
{
116+
log() << "use inflow_restriction + unsubscribe" << std::endl;
117+
auto mtx = std::make_shared<std::mutex>();
118+
auto list = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15};
119+
auto api = std::make_shared<long_api>();
120+
auto ifr = std::make_shared<inflow_restriction<4>>();
121+
122+
auto sbsc = (
123+
observables::iterate(list)
124+
| flat_map([ifr, api](int n){
88125
return ifr->enter(api->call())
89-
| map([=](result){
126+
| tap([n](auto){
127+
log() << "tap: " << n << std::endl;
128+
})
129+
| map([n](result){
90130
return n;
91131
});
92132
})
93133
)
134+
.take(5)
94135
.subscribe({
95-
[=](const int& x){
136+
[mtx](const int& x){
96137
std::lock_guard<std::mutex> lock(*mtx);
97138
log() << "next " << x << std::endl;
98139
},
99-
[=](std::exception_ptr){
140+
[mtx](std::exception_ptr){
100141
std::lock_guard<std::mutex> lock(*mtx);
101142
log() << "error " << std::endl;
102143
},
103-
[=](){
144+
[mtx](){
104145
std::lock_guard<std::mutex> lock(*mtx);
105146
log() << "completed " << std::endl;
106147
}

0 commit comments

Comments
 (0)