1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#pragma once
#if !defined(RXCPP_SOURCES_RX_INTERVAL_HPP)
#define RXCPP_SOURCES_RX_INTERVAL_HPP
#include "../rx-includes.hpp"
namespace rxcpp {
namespace sources {
namespace detail {
template<class Coordination>
struct interval : public source_base<long>
{
typedef interval<Coordination> this_type;
typedef rxu::decay_t<Coordination> coordination_type;
typedef typename coordination_type::coordinator_type coordinator_type;
struct interval_initial_type
{
interval_initial_type(rxsc::scheduler::clock_type::time_point i, rxsc::scheduler::clock_type::duration p, coordination_type cn)
: initial(i)
, period(p)
, coordination(std::move(cn))
{
}
rxsc::scheduler::clock_type::time_point initial;
rxsc::scheduler::clock_type::duration period;
coordination_type coordination;
};
interval_initial_type initial;
interval(rxsc::scheduler::clock_type::time_point i, rxsc::scheduler::clock_type::duration p, coordination_type cn)
: initial(i, p, std::move(cn))
{
}
template<class Subscriber>
void on_subscribe(Subscriber o) const {
static_assert(is_subscriber<Subscriber>::value, "subscribe must be passed a subscriber");
// creates a worker whose lifetime is the same as this subscription
auto coordinator = initial.coordination.create_coordinator(o.get_subscription());
auto controller = coordinator.get_worker();
auto counter = std::make_shared<long>(0);
auto producer = [o, counter](const rxsc::schedulable&) {
// send next value
o.on_next(++(*counter));
};
auto selectedProducer = on_exception(
[&](){return coordinator.act(producer);},
o);
if (selectedProducer.empty()) {
return;
}
controller.schedule_periodically(initial.initial, initial.period, selectedProducer.get());
}
};
template<class Duration, class Coordination>
struct defer_interval : public defer_observable<
rxu::all_true<
std::is_convertible<Duration, rxsc::scheduler::clock_type::duration>::value,
is_coordination<Coordination>::value>,
void,
interval, Coordination>
{
};
}
template<class Duration>
auto interval(Duration period)
-> typename std::enable_if<
detail::defer_interval<Duration, identity_one_worker>::value,
typename detail::defer_interval<Duration, identity_one_worker>::observable_type>::type {
return detail::defer_interval<Duration, identity_one_worker>::make(identity_current_thread().now(), period, identity_current_thread());
}
template<class Coordination>
auto interval(rxsc::scheduler::clock_type::duration period, Coordination cn)
-> typename std::enable_if<
detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::value,
typename detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::observable_type>::type {
return detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::make(cn.now(), period, std::move(cn));
}
template<class Duration>
auto interval(rxsc::scheduler::clock_type::time_point when, Duration period)
-> typename std::enable_if<
detail::defer_interval<Duration, identity_one_worker>::value,
typename detail::defer_interval<Duration, identity_one_worker>::observable_type>::type {
return detail::defer_interval<Duration, identity_one_worker>::make(when, period, identity_current_thread());
}
template<class Coordination>
auto interval(rxsc::scheduler::clock_type::time_point when, rxsc::scheduler::clock_type::duration period, Coordination cn)
-> typename std::enable_if<
detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::value,
typename detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::observable_type>::type {
return detail::defer_interval<rxsc::scheduler::clock_type::duration, Coordination>::make(when, period, std::move(cn));
}
}
}
#endif
|
