Skip to content

Commit 4538dd9

Browse files
schlangsterschlangster
committed
Added event operation: Join.
Joins two streams A,B to tuple<A,B>. Values are buffered for each input slot and emitted once a tuple is complete.
1 parent c29d50e commit 4538dd9

2 files changed

Lines changed: 151 additions & 2 deletions

File tree

‎include/react/Event.h‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,6 @@ auto MakeEventSource()
6060
///////////////////////////////////////////////////////////////////////////////////////////////////
6161
/// Merge
6262
///////////////////////////////////////////////////////////////////////////////////////////////////
63-
// Note: Default template arguments are in forward declaration
6463
template
6564
<
6665
typename D,
@@ -77,7 +76,7 @@ auto Merge(const Events<D,TArg1>& arg1, const Events<D,TArgs>& ... args)
7776
using REACT_IMPL::EventOpNode;
7877

7978
static_assert(sizeof...(TArgs) > 0,
80-
"react::Merge requires at least 2 arguments.");
79+
"Merge: 2+ arguments are required.");
8180

8281
return TempEvents<D,E,TOp>(
8382
std::make_shared<EventOpNode<D,E,TOp>>(
@@ -427,6 +426,27 @@ auto Flatten(const Signal<D,Events<D,TInnerValue>>& outer)
427426
GetNodePtr(outer), GetNodePtr(outer.Value())));
428427
}
429428

429+
///////////////////////////////////////////////////////////////////////////////////////////////////
430+
/// Join
431+
///////////////////////////////////////////////////////////////////////////////////////////////////
432+
template
433+
<
434+
typename D,
435+
typename ... TArgs
436+
>
437+
auto Join(const Events<D,TArgs>& ... args)
438+
-> Events<D, std::tuple<TArgs ...>>
439+
{
440+
using REACT_IMPL::EventJoinNode;
441+
442+
static_assert(sizeof...(TArgs) > 1,
443+
"Join: 2+ arguments are required.");
444+
445+
return Events<D, std::tuple<TArgs ...>>(
446+
std::make_shared<EventJoinNode<D,TArgs...>>(
447+
GetNodePtr(args) ...));
448+
}
449+
430450
///////////////////////////////////////////////////////////////////////////////////////////////////
431451
/// Token
432452
///////////////////////////////////////////////////////////////////////////////////////////////////

‎include/react/detail/graph/EventNodes.h‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -898,6 +898,135 @@ class SyncedEventProcessingNode :
898898
DepHolderT deps_;
899899
};
900900

901+
///////////////////////////////////////////////////////////////////////////////////////////////////
902+
/// EventJoinNode
903+
///////////////////////////////////////////////////////////////////////////////////////////////////
904+
template
905+
<
906+
typename D,
907+
typename ... TValues
908+
>
909+
class EventJoinNode :
910+
public EventStreamNode<D,std::tuple<TValues ...>>
911+
{
912+
using Engine = typename EventJoinNode::Engine;
913+
using TurnT = typename Engine::TurnT;
914+
915+
public:
916+
EventJoinNode(const std::shared_ptr<EventStreamNode<D,TValues>>& ... sources) :
917+
EventJoinNode::EventStreamNode( ),
918+
slots_( sources ... )
919+
{
920+
Engine::OnNodeCreate(*this);
921+
REACT_EXPAND_PACK(Engine::OnNodeAttach(*this, *sources));
922+
}
923+
924+
~EventJoinNode()
925+
{
926+
apply(
927+
[this] (Slot<TValues>& ... slots) {
928+
REACT_EXPAND_PACK(Engine::OnNodeDetach(*this, *slots.Source));
929+
},
930+
slots_);
931+
932+
Engine::OnNodeDestroy(*this);
933+
}
934+
935+
virtual void Tick(void* turnPtr) override
936+
{
937+
TurnT& turn = *reinterpret_cast<TurnT*>(turnPtr);
938+
939+
this->SetCurrentTurn(turn, true);
940+
941+
REACT_LOG(D::Log().template Append<NodeEvaluateBeginEvent>(
942+
GetObjectId(*this), turn.Id()));
943+
944+
// Don't time if there is nothing to do
945+
{// timer
946+
size_t count = 0;
947+
using TimerT = typename EventJoinNode::ScopedUpdateTimer;
948+
TimerT scopedTimer( *this, count );
949+
950+
// Move events into buffers
951+
apply(
952+
[this, &turn] (Slot<TValues>& ... slots) {
953+
REACT_EXPAND_PACK(fetchBuffer(turn, slots));
954+
},
955+
slots_);
956+
957+
while (true)
958+
{
959+
bool isReady = true;
960+
961+
// All slots ready?
962+
apply(
963+
[this,&isReady] (Slot<TValues>& ... slots) {
964+
// Todo: combine return values instead
965+
REACT_EXPAND_PACK(checkSlot(slots, isReady));
966+
},
967+
slots_);
968+
969+
if (!isReady)
970+
break;
971+
972+
// Pop values from buffers and emit tuple
973+
apply(
974+
[this] (Slot<TValues>& ... slots) {
975+
this->events_.emplace_back(slots.Buffer.front() ...);
976+
REACT_EXPAND_PACK(slots.Buffer.pop_front());
977+
},
978+
slots_);
979+
}
980+
981+
count = this->events_.size();
982+
983+
}// ~timer
984+
985+
REACT_LOG(D::Log().template Append<NodeEvaluateEndEvent>(
986+
GetObjectId(*this), turn.Id()));
987+
988+
if (! this->events_.empty())
989+
Engine::OnNodePulse(*this, turn);
990+
else
991+
Engine::OnNodeIdlePulse(*this, turn);
992+
}
993+
994+
virtual const char* GetNodeType() const override { return "EventJoinNode"; }
995+
virtual int DependencyCount() const override { return sizeof...(TValues); }
996+
997+
private:
998+
template <typename T>
999+
struct Slot
1000+
{
1001+
Slot(const std::shared_ptr<EventStreamNode<D,T>>& src) :
1002+
Source( src )
1003+
{}
1004+
1005+
std::shared_ptr<EventStreamNode<D,T>> Source;
1006+
std::deque<T> Buffer;
1007+
};
1008+
1009+
template <typename T>
1010+
static void fetchBuffer(TurnT& turn, Slot<T>& slot)
1011+
{
1012+
slot.Source->SetCurrentTurn(turn);
1013+
1014+
slot.Buffer.insert(
1015+
slot.Buffer.end(),
1016+
slot.Source->Events().begin(),
1017+
slot.Source->Events().end());
1018+
}
1019+
1020+
template <typename T>
1021+
static void checkSlot(Slot<T>& slot, bool& isReady)
1022+
{
1023+
auto t = isReady && !slot.Buffer.empty();
1024+
isReady = t;
1025+
}
1026+
1027+
std::tuple<Slot<TValues>...> slots_;
1028+
};
1029+
9011030
/****************************************/ REACT_IMPL_END /***************************************/
9021031

9031032
#endif // REACT_DETAIL_GRAPH_EVENTNODES_H_INCLUDED

0 commit comments

Comments
 (0)