@@ -6,51 +6,66 @@ PushPullInputAdapter::PushPullInputAdapter( Engine *engine, CspTypePtr &type, Pu
66 PushGroup *group, bool adjustOutOfOrderTime )
77 : PushInputAdapter(engine, type, pushMode, group),
88 m_nextPullEvent (nullptr ),
9+ m_tailEvent(nullptr ),
910 m_notifiedEndOfPull(false ),
1011 m_adjustOutOfOrderTime(adjustOutOfOrderTime)
12+ {
13+ }
14+
15+ PushPullInputAdapter::~PushPullInputAdapter ()
1116{
1217 // free up any unused events
13- while ( m_nextPullEvent )
18+ PushPullEvent * event = nextPullEvent ();
19+ while ( event )
1420 {
15- delete m_nextPullEvent ;
16- m_nextPullEvent = nextPullEvent ();
21+ delete event ;
22+ event = nextPullEvent ();
1723 }
1824}
1925
2026void PushPullInputAdapter::start ( DateTime start, DateTime end )
2127{
22- m_nextPullEvent = nextPullEvent ();
23- if ( m_nextPullEvent )
24- {
25- m_timerHandle = rootEngine () -> scheduleCallback ( m_nextPullEvent -> time,
26- [this ]() { return processNextPullEvent () ? nullptr : this ; } );
27- }
28+ auto * nextEvent = nextPullEvent ();
29+ if ( nextEvent )
30+ scheduleNextPullEvent ( nextEvent );
2831}
2932
3033void PushPullInputAdapter::stop ()
3134{
3235 rootEngine () -> cancelCallback ( m_timerHandle );
3336 // shouldnt need to lock at this point
34- m_threadQueue.emplace ( nullptr );
37+ auto * replayCompleteEvent = new PushPullEvent ( this , DateTime::NONE () );
38+ rootEngine () -> pushPullEventQueue ().push ( replayCompleteEvent );
39+ }
40+
41+ void PushPullInputAdapter::scheduleNextPullEvent ( PushPullEvent * nextEvent )
42+ {
43+ // Note that we make nextEvent mutable in the lambda since we need to be able to update it in processNextPullEvent
44+ // which can return false to force a rescheduled re-attempt with a new event pointer
45+ m_timerHandle = rootEngine () -> scheduleCallback ( nextEvent -> time,
46+ [this , nextEvent]() mutable
47+ {
48+ return processNextPullEvent ( nextEvent ) ? nullptr : this ;
49+ } );
3550}
3651
37- bool PushPullInputAdapter::processNextPullEvent ()
52+ bool PushPullInputAdapter::processNextPullEvent ( PushPullEvent *& nextEvent )
3853{
3954 bool consumed = switchCspType ( dataType (),
40- [ this ]( auto tag )
55+ [ this , &nextEvent ]( auto tag )
4156 {
4257 using T = typename decltype (tag)::type;
43- TypedPullDataEvent <T> *tevent = static_cast <TypedPullDataEvent <T> *>( m_nextPullEvent );
58+ TypedPushPullEvent <T> *tevent = static_cast <TypedPushPullEvent <T> *>( nextEvent );
4459
4560 bool consumed = consumeTick ( tevent -> data );
4661 assert ( consumed );
4762
4863 delete tevent;
4964
50- while ( ( m_nextPullEvent = nextPullEvent () ) &&
51- m_nextPullEvent -> time == rootEngine () -> now () )
65+ while ( ( nextEvent = nextPullEvent () ) &&
66+ nextEvent -> time == rootEngine () -> now () )
5267 {
53- tevent = static_cast <TypedPullDataEvent <T> *>( m_nextPullEvent );
68+ tevent = static_cast <TypedPushPullEvent <T> *>( nextEvent );
5469 consumed = consumeTick ( tevent -> data );
5570 if ( !consumed )
5671 return false ;
@@ -60,38 +75,41 @@ bool PushPullInputAdapter::processNextPullEvent()
6075 return true ;
6176 } );
6277
63- if ( consumed && m_nextPullEvent )
64- {
65- m_timerHandle = rootEngine () -> scheduleCallback ( m_nextPullEvent->time ,
66- [this ]() { return processNextPullEvent () ? nullptr : this ; } );
67- }
78+ if ( consumed && nextEvent )
79+ scheduleNextPullEvent ( nextEvent );
6880
6981 return consumed;
7082}
7183
72- PushPullInputAdapter::PullDataEvent * PushPullInputAdapter::nextPullEvent ()
84+ PushPullEvent * PushPullInputAdapter::nextPullEvent ()
7385{
74- // spin while we wait for data
75- while ( m_poppedPullEvents.empty () )
86+ while ( m_nextPullEvent == nullptr )
7687 {
77- std::lock_guard<std::mutex> g ( m_queueMutex );
78- m_threadQueue.swap ( m_poppedPullEvents );
88+ // Any PushPullInputAdapter instance can update events on any other adapter
89+ PushPullEvent * event = rootEngine () -> pushPullEventQueue ().popAll ();
90+ while ( event )
91+ {
92+ PushPullEvent * next = event -> next;
93+ event -> adapter -> setNextPushPullEvent ( event );
94+ event = next;
95+ }
7996 }
8097
81- auto * event = m_poppedPullEvents.front ();
82- m_poppedPullEvents.pop ();
98+ // DateTime of None is the sentinel value for replay complete
99+ if ( m_nextPullEvent -> time.isNone () )
100+ return nullptr ;
101+
102+ auto * event = m_nextPullEvent;
103+ m_nextPullEvent = m_nextPullEvent -> next;
83104
84- if ( event )
85- {
86- // Always force time before start to start. There are two possibilities:
87- // - User asked to replay from EARLIEST, so they should get all ticks replayed and we cant replay before starttime
88- // - User asked to replay from STARTTIME in which case, if the adapter is written correctly, we shouldnt get ticks before starttime
89- if ( unlikely ( event -> time < rootEngine () -> startTime () ) )
90- event -> time = rootEngine () -> startTime ();
91-
92- if ( m_adjustOutOfOrderTime )
93- event -> time = std::max ( event -> time, rootEngine () -> now () );
94- }
105+ // Always force time before start to start. There are two possibilities:
106+ // - User asked to replay from EARLIEST, so they should get all ticks replayed and we cant replay before starttime
107+ // - User asked to replay from STARTTIME in which case, if the adapter is written correctly, we shouldnt get ticks before starttime
108+ if ( unlikely ( event -> time < rootEngine () -> startTime () ) )
109+ event -> time = rootEngine () -> startTime ();
110+
111+ if ( m_adjustOutOfOrderTime )
112+ event -> time = std::max ( event -> time, rootEngine () -> now () );
95113
96114 return event;
97115}
0 commit comments