@@ -1,5 +1,7 @@
11{-# LANGUAGE CPP #-}
22{-# LANGUAGE DeriveDataTypeable #-}
3+{-# LANGUAGE DeriveGeneric #-}
4+{-# LANGUAGE DeriveTraversable #-}
35{-# LANGUAGE FlexibleContexts #-}
46{-# LANGUAGE MultiParamTypeClasses #-}
57{-# LANGUAGE RecursiveDo #-}
@@ -24,12 +26,15 @@ import Control.Monad
2426import Control.Monad.Fix
2527import Control.Monad.IO.Class
2628import Data.Align
29+import Data.Data (Data)
2730import Data.Fixed
31+import Data.Semigroup
2832import Data.Sequence (Seq, (|>))
2933import qualified Data.Sequence as Seq
3034import Data.These
3135import Data.Time.Clock
3236import Data.Typeable
37+import GHC.Generics (Generic)
3338import System.Random
3439
3540data TickInfo
@@ -266,6 +271,92 @@ throttle t e = do
266271 delayed <- delay t outE
267272 return outE
268273
274+data ThrottleState b
275+ = ThrottleState_Immediate
276+ | ThrottleState_Buffered (ThrottleBuffer b)
277+ deriving (Eq, Ord, Show, Functor, Foldable, Traversable, Generic, Data, Typeable)
278+
279+data ThrottleBuffer b
280+ = ThrottleBuffer_Empty -- Empty conflicts with lens, and hiding it would require turning
281+ -- on PatternSynonyms
282+ | ThrottleBuffer_Full b
283+ deriving (Eq, Ord, Show, Functor, Foldable, Traversable, Generic, Data, Typeable)
284+
285+instance Semigroup b => Semigroup (ThrottleBuffer b) where
286+ x <> y = case x of
287+ ThrottleBuffer_Empty -> y
288+ ThrottleBuffer_Full b1 -> case y of
289+ ThrottleBuffer_Empty -> x
290+ ThrottleBuffer_Full b2 -> ThrottleBuffer_Full $ b1 <> b2
291+ {-# INLINE (<>) #-}
292+
293+instance Semigroup b => Monoid (ThrottleBuffer b) where
294+ mempty = ThrottleBuffer_Empty
295+ {-# INLINE mempty #-}
296+ mappend = (<>)
297+ {-# INLINE mappend #-}
298+
299+-- | Throttle an input event, ensuring that the output event doesn't occur more often than you are ready for it. If the input event occurs too
300+-- frequently, the output event will contain semigroup-based summaries of the input firings that happened since the last output firing.
301+-- If the output event has not occurred recently, occurrences of the input event will cause the output event to fire immediately.
302+-- The first parameter is a function that receives access to the output event, and should construct an event that fires when the receiver is
303+-- ready for more input. For example, using @delay 20@ would give a simple time-based throttle.
304+--
305+-- NB: The provided lag function must *actually* delay the event.
306+throttleBatchWithLag :: (MonadFix m, MonadHold t m, PerformEvent t m, Semigroup a) => (Event t () -> m (Event t ())) -> Event t a -> m (Event t a)
307+-- Invariants:
308+-- * Immediate mode must turn off whenever output is produced.
309+-- * Output must be produced whenever immediate mode turns from on to off.
310+-- * Immediate mode can only go from off to on when the delayed event fires.
311+-- * Every input firing must go into either an immediate output firing or the
312+-- buffer, but not both.
313+-- * An existing full buffer must either stay in the buffer or go to output,
314+-- but not both.
315+throttleBatchWithLag lag e = do
316+ let f state x = case x of -- (Just $ newState, out)
317+ This a -> -- If only the input event fires
318+ case state of
319+ ThrottleState_Immediate -> -- and we're in immediate mode
320+ -- Immediate mode turns off, and the buffer is empty.
321+ -- We fire the output event with the input event value immediately.
322+ (Just $ ThrottleState_Buffered $ ThrottleBuffer_Empty, Just a)
323+ ThrottleState_Buffered b -> -- and we're not in immediate mode
324+ -- Immediate mode remains off, and we accumulate the input value.
325+ -- We don't fire the output event.
326+ (Just $ ThrottleState_Buffered $ b <> ThrottleBuffer_Full a, Nothing)
327+ That _ -> -- If only the delayed output event fires,
328+ case state of
329+ ThrottleState_Immediate -> -- and we're in immediate mode
330+ -- Nothing happens.
331+ (Nothing, Nothing)
332+ ThrottleState_Buffered ThrottleBuffer_Empty -> -- and the buffer is empty:
333+ -- Immediate mode turns back on, and the buffer remains empty.
334+ -- We don't fire.
335+ (Just ThrottleState_Immediate, Nothing)
336+ ThrottleState_Buffered (ThrottleBuffer_Full b) -> -- and the buffer is full:
337+ -- Immediate mode remains off, and the buffer is cleared.
338+ -- We fire with the buffered value.
339+ (Just $ ThrottleState_Buffered ThrottleBuffer_Empty, Just b)
340+ These a _ -> -- If both the input and delayed output event fire simultaneously:
341+ case state of
342+ ThrottleState_Immediate -> -- and we're in immediate mode
343+ -- Immediate mode turns off, and the buffer is empty.
344+ -- We fire with the input event's value, as it is the most recent we have seen at this moment.
345+ (Just $ ThrottleState_Buffered ThrottleBuffer_Empty, Just a)
346+ ThrottleState_Buffered ThrottleBuffer_Empty -> -- and the buffer is empty:
347+ -- Immediate mode stays off, and the buffer remains empty.
348+ -- We fire with the input event's value.
349+ (Just $ ThrottleState_Buffered ThrottleBuffer_Empty, Just a)
350+ ThrottleState_Buffered (ThrottleBuffer_Full b) -> -- and the buffer is full:
351+ -- Immediate mode remains off, and the buffer is cleared.
352+ -- We fire with everything including the buffered value.
353+ (Just $ ThrottleState_Buffered ThrottleBuffer_Empty, Just (b <> a))
354+ rec (_stateDyn, outE) <- mapAccumMaybeDyn f
355+ ThrottleState_Immediate -- We start in immediate mode with an empty buffer.
356+ (align e delayed)
357+ delayed <- lag (void outE)
358+ return outE
359+
269360#ifdef USE_TEMPLATE_HASKELL
270361makeLensesWith (lensRules & simpleLenses .~ True) ''TickInfo
271362#else
@@ -281,4 +372,3 @@ tickInfo_alreadyElapsed :: Lens' TickInfo NominalDiffTime
281372tickInfo_alreadyElapsed f (TickInfo x1 x2 x3) = (\y -> TickInfo x1 x2 y) <$> f x3
282373{-# INLINE tickInfo_alreadyElapsed #-}
283374#endif
284−