1+
2+ // Copyright Sebastian Jeckel 2014.
3+ // Distributed under the Boost Software License, Version 1.0.
4+ // (See accompanying file LICENSE_1_0.txt or copy at
5+ // http://www.boost.org/LICENSE_1_0.txt)
6+
7+ #pragma once
8+
9+ #ifndef REACT_DISABLE_REACTORS
10+
11+ #include " react/Defs.h"
12+
13+ #include < cstdint>
14+ #include < functional>
15+ #include < memory>
16+ #include < unordered_map>
17+ #include < utility>
18+
19+ #include < boost/coroutine/all.hpp>
20+
21+ #include " GraphBase.h"
22+ #include " EventStreamNodes.h"
23+
24+ /* ********************************/ /* ********************************/
25+
26+ // //////////////////////////////////////////////////////////////////////////////////////
27+ // / ReactorNode
28+ // //////////////////////////////////////////////////////////////////////////////////////
29+ template
30+ <
31+ typename D,
32+ typename TContext
33+ >
34+ class ReactorNode :
35+ public ReactiveNode<D,void ,void >
36+ {
37+ public:
38+ using NodeBasePtrT = NodeBase<D>::PtrT;
39+
40+ using CoroutineT = boost::coroutines::coroutine<const NodeBasePtrT*>;
41+ using LoopT = typename CoroutineT::pull_type;
42+ using OutT = typename CoroutineT::push_type;
43+
44+ using TurnT = typename D::Engine::TurnInterface;
45+
46+ template <typename F>
47+ ReactorNode (F&& func, bool registered) :
48+ ReactiveNode<D,void ,void >(true ),
49+ func_{ std::forward<F>(func) }
50+ {
51+ }
52+
53+ void StartLoop ()
54+ {
55+ // Could already have started it in ctor,
56+ // but lets make sure node is fully constructed before calls to
57+ // context in coroutine can happen.
58+ mainLoop_ = LoopT
59+ (
60+ [&] (OutT& out)
61+ {
62+ curOutPtr_ = &out;
63+
64+ TContext ctx{ *this };
65+
66+ while (true )
67+ {
68+ func_ (ctx);
69+ }
70+ }
71+ );
72+
73+ // First blocking event is not initiated by Tick() but after loop creation.
74+ const auto * p = mainLoop_.get ();
75+
76+ REACT_ASSERT (p != nullptr , " StartLoop: first depPtr was null"
77+
78+ Engine::OnNodeAttach (*this , **p);
79+ ++depCount_;
80+ }
81+
82+ virtual const char * GetNodeType () const override { return " ReactorNode"
83+
84+ virtual bool IsDynamicNode () const override { return true ; }
85+
86+ virtual ETickResult Tick (void * turnPtr) override
87+ {
88+ turnPtr_ = static_cast <TurnT*>(turnPtr);
89+ REACT_SCOPE_EXIT{ turnPtr_ = nullptr ; };
90+
91+ mainLoop_ ();
92+
93+ if (mainLoop_.get () != nullptr )
94+ {
95+ const auto & depPtr = *mainLoop_.get ();
96+ Engine::OnDynamicNodeAttach (*this , *depPtr, *turnPtr_);
97+ ++depCount_;
98+
99+ return ETickResult::invalidated;
100+ }
101+
102+ offsets_.clear ();
103+
104+ return ETickResult::none;
105+ }
106+
107+ virtual int DependencyCount () const override
108+ {
109+ return depCount_;
110+ }
111+
112+ template <typename E>
113+ E& Take (const EventStreamNodePtr<D,E>& events)
114+ {
115+ // First attach to target event node
116+ (*curOutPtr_)(&std::static_pointer_cast<NodeBase<D>>(events));
117+
118+ while (! checkEvent<E>(events))
119+ (*curOutPtr_)(nullptr );
120+
121+ REACT_ASSERT (turnPtr_ != nullptr , " Take: turnPtr_ was null"
122+
123+ Engine::OnDynamicNodeDetach (*this , *events, *turnPtr_);
124+ --depCount_;
125+
126+ return events->Events ()[offsets_[reinterpret_cast <uintptr_t >(&events)]++];
127+ }
128+
129+ template <typename E, typename F>
130+ void RepeatUntil (const EventStreamNodePtr<D,E>& events, F func)
131+ {
132+ // First attach to target event node
133+ if (turnPtr_ != nullptr )
134+ {
135+ (*curOutPtr_)(&std::static_pointer_cast<NodeBase<D>>(events));
136+ }
137+ else
138+ {
139+ // Non-dynamic attach in case first loop until is encountered before the loop was
140+ // suspended for the first time.
141+ Engine::OnNodeAttach (*this , *events);
142+ ++depCount_;
143+ }
144+
145+ // Detach when this function is exited
146+ REACT_SCOPE_EXIT
147+ {
148+ if (turnPtr_ != nullptr )
149+ Engine::OnDynamicNodeDetach (*this , *events, *turnPtr_);
150+ else
151+ Engine::OnNodeDetach (*this , *events);
152+ --depCount_;
153+ };
154+
155+ // Don't enter loop if event already present
156+ if (checkEvent<E>(events))
157+ return ;
158+
159+ auto * parentOutPtr = curOutPtr_;
160+ REACT_SCOPE_EXIT{ curOutPtr_ = parentOutPtr; };
161+
162+ // Create and start loop
163+ LoopT nestedLoop_
164+ {
165+ [&] (OutT& out)
166+ {
167+ curOutPtr_ = &out;
168+ while (true )
169+ func ();
170+ }
171+ };
172+
173+ // First suspend from initial loop run
174+ (*parentOutPtr)(nestedLoop_.get ());
175+
176+ // Further iterations
177+ while (! checkEvent<E>(events))
178+ {
179+ // Advance loop, forward blocking event to parent, and suspend
180+ nestedLoop_ ();
181+ (*parentOutPtr)(nestedLoop_.get ());
182+ }
183+ }
184+
185+ private:
186+ template <typename E>
187+ bool checkEvent (const EventStreamNodePtr<D,E>& events)
188+ {
189+ if (turnPtr_ == nullptr )
190+ {
191+ return false ;
192+ }
193+
194+ events->SetCurrentTurn (*turnPtr_);
195+ return offsets_[reinterpret_cast <uintptr_t >(&events)] < events->Events ().size ();
196+ }
197+
198+ std::function<void (TContext&)> func_;
199+
200+ LoopT mainLoop_;
201+ TurnT* turnPtr_;
202+
203+ OutT* curOutPtr_ = nullptr ;
204+
205+ int depCount_ = 0 ;
206+
207+ std::unordered_map<std::uintptr_t , std::size_t > offsets_;
208+ };
209+
210+ /* *********************************/ /* *********************************/
211+
212+ #endif // REACT_DISABLE_REACTORS
0 commit comments