@@ -139,9 +139,63 @@ sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
139139 Double -> t m a -> t m a
140140sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
141141
142+data BurstState t x =
143+ BurstNone
144+ | BurstWait !t !x
145+ | BurstDone !x
146+ | BurstDoneNext !x !t !x
147+
148+{-# INLINE sampleBurst #-}
149+sampleBurst :: (IsStream t, MonadAsync m, Functor (t m)) =>
150+ Bool -> Double -> t m a -> t m a
151+sampleBurst sampleAtEnd gap xs =
152+ -- XXX Ideally we should schedule a timer event exactly after gap time,
153+ -- but the tick stream should work well as long as the timer
154+ -- granularity is small enough compared to the gap.
155+ Stream.mapMaybe extract
156+ $ Stream.scanl' step BurstNone
157+ $ Stream.timeIndexed
158+ $ Stream.interjectSuffix 0.01 (return Nothing) (Stream.map Just xs)
159+
160+ where
161+
162+ gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
163+
164+ {-# INLINE step #-}
165+ step BurstNone (t1, Just x1) = BurstWait t1 x1
166+ step BurstNone _ = BurstNone
167+
168+ step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
169+ step (BurstDone _) _ = BurstNone
170+
171+ step old@(BurstWait t0 x0) (t1, Nothing)
172+ | t1 - t0 >= gap1 = BurstDone x0
173+ | otherwise = old
174+ -- This can happen due to scheduling delays, if we received back to
175+ -- back events spaced by more than the timeout without an
176+ -- intervening timeout event then we emit the old event instead of
177+ -- replacing it by the new.
178+ step (BurstWait t0 x0) (t1, Just x1)
179+ | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
180+ | sampleAtEnd = BurstWait t1 x1
181+ | otherwise = BurstWait t1 x0
182+
183+ step (BurstDoneNext _ t0 x0) (t1, Nothing)
184+ | t1 - t0 >= gap1 = BurstDone x0
185+ | otherwise = BurstWait t0 x0
186+ step (BurstDoneNext _ t0 x0) (t1, Just x1)
187+ | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
188+ | sampleAtEnd = BurstWait t1 x1
189+ | otherwise = BurstWait t1 x0
190+
191+ {-# INLINE extract #-}
192+ extract (BurstDoneNext x _ _) = Just x
193+ extract (BurstDone x) = Just x
194+ extract _ = Nothing
195+
142196-- | Sample one event at the end of each burst of events. A burst is a group
143197-- of events close together in time, it ends when an event is spaced by more
144−-- than the specified time interval from the previous event.
198+-- than the specified time interval (in seconds) from the previous event.
145199--
146200-- This is known as @debounce@ in some libraries.
147201--
@@ -152,13 +206,7 @@ sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
152206{-# INLINE sampleBurstEnd #-}
153207sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
154208 Double -> t m a -> t m a
155−sampleBurstEnd gap =
156− let f (t1, _) (t2, _) =
157− t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
158− in Stream.map snd
159− . Stream.catMaybes
160− . Stream.groupsByRolling f Fold.last
161− . Stream.timeIndexed
209+sampleBurstEnd = sampleBurst True
162210
163211-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
164212-- instead of at the end of it.
@@ -168,13 +216,7 @@ sampleBurstEnd gap =
168216{-# INLINE sampleBurstStart #-}
169217sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
170218 Double -> t m a -> t m a
171−sampleBurstStart gap =
172− let f (t1, _) (t2, _) =
173− t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
174− in Stream.map snd
175− . Stream.catMaybes
176− . Stream.groupsByRolling f Fold.head
177− . Stream.timeIndexed
219+sampleBurstStart = sampleBurst False
178220
179221------------------------------------------------------------------------------
180222-- Reordering