pipes-rt

A few pipes to yield values at a steady rate, as a poisson process, or according to the values themselves

For example:

λ: import Pipes
λ: import qualified Pipes.Prelude as PP
λ: import Pipes.RealTime
λ: import Data.Time
λ: import Data.Time.Clock 

λ: -- Pass values at 2 Hz
λ: runEffect $ each [1..4] >-> steadyCat 2 >-> PP.print
  [1/2 second pause...]
1 [1/2 second pause...]
2 [1/2 second pause...]
3 [1/2 second pause...]
4 [1/2 second pause...]

λ: -- Pass Values at 100 Hz, printing arrival times
λ: runEffect $ for (each [1..10] >-> steadyCat 100) (const $ lift (getCurrentTime >>= print))
2013-10-10 19:55:53.944484 UTC  
2013-10-10 19:55:53.954939 UTC      [          ]
2013-10-10 19:55:53.964623 UTC    [              ]
2013-10-10 19:55:53.975125 UTC  [   .. pauses ..    ]
2013-10-10 19:55:53.984759 UTC    [              ]
2013-10-10 19:55:53.994345 UTC      [          ]
2013-10-10 19:55:54.004886 UTC
2013-10-10 19:55:54.01449 UTC
2013-10-10 19:55:54.025124 UTC
2013-10-10 19:55:54.034661 UTC

λ: -- Pass values with Poisson timing, average 4 Hz, print data and arrival time
λ: runEffect $ for (each "Testing" >-> poissonCat 4) (\c -> lift (getCurrentTime >>= \t -> print (c,t)))
('T',2013-10-10 19:57:29.707621 UTC)
('e',2013-10-10 19:57:29.710815 UTC)
('s',2013-10-10 19:57:29.71766 UTC)
('t',2013-10-10 19:57:29.726371 UTC)
('i',2013-10-10 19:57:29.74401 UTC)
('n',2013-10-10 19:57:29.744338 UTC)
('g',2013-10-10 19:57:29.759882 UTC)

λ: -- Get timestamps from the data being piped
λ: import Data.Char
λ: let timeOfChar = (/ 10) . fromIntegral . (\c -> ord c - ord 'a')
λ: runEffect $ for (each "abcdwxyz" >-> relativeTimeCat timeOfChar) (\v -> lift (getCurrentTime >>= \t -> print (v,t)))
('a',2013-11-07 15:54:05.645025 UTC)   [ .. short pause .. ]
('b',2013-11-07 15:54:05.745847 UTC)   [ .. short pause .. ]
('c',2013-11-07 15:54:05.845771 UTC)   [ .. short pause .. ]
('d',2013-11-07 15:54:05.945533 UTC)   [ .. long  pause .. ]
('w',2013-11-07 15:54:07.847302 UTC)   [ .. short pause .. ]
('x',2013-11-07 15:54:07.946071 UTC)   [ .. short pause .. ]
('y',2013-11-07 15:54:08.045846 UTC)   [ .. short pause .. ]
('z',2013-11-07 15:54:08.145573 UTC)   [ .. short pause .. ]

λ: -- "Delay" the output by -2 seconds, which means
λ: -- skip ahead by 2 seconds
λ: let myPrint = (\v -> (getCurrentTime >>= \t -> print (v,t))) :: Char -> IO ()
λ: runEffect $ for (each "abcdwxyz" >-> relativeTimeCatDelayedBy timeOfChar (-0.2)) (lift . myPrint)
('c',2013-11-08 02:55:37.131626 UTC)
('d',2013-11-08 02:55:37.232347 UTC)   [ .. we discarded the early data .. ]
('w',2013-11-08 02:55:39.134008 UTC)   [ .. and jumped in immediately   .. ]  
('x',2013-11-08 02:55:39.232772 UTC)   [ .. at 'c'                      .. ]
('y',2013-11-08 02:55:39.332545 UTC)
('z',2013-11-08 02:55:39.432303 UTC)