pipes-misc 0.2.2.1 → 0.2.3.0

raw patch · 8 files changed

+353/−130 lines, 8 filesdep +Decimaldep +clockdep +mmorphPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: Decimal, clock, mmorph

API changes (from Hackage documentation)

Files

• pipes-misc.cabal +11/−1
• src/Pipes/Misc.hs +10/−129
• src/Pipes/Misc/Concurrent.hs +55/−0
• src/Pipes/Misc/State/Lazy.hs +40/−0
• src/Pipes/Misc/State/Strict.hs +40/−0
• src/Pipes/Misc/Time.hs +71/−0
• src/Pipes/Misc/Util.hs +120/−0
• test/Spec.hs +6/−0

@@ -1,5 +1,5 @@ name:                pipes-misc-version:             0.2.2.1+version:             0.2.3.0 synopsis:            Miscellaneous utilities for pipes, required by glazier-tutorial description:         Please see README.md homepage:            https://github.com/louispan/pipes-misc#readme@@ -16,8 +16,15 @@ library   hs-source-dirs:      src   exposed-modules:     Pipes.Misc+                       Pipes.Misc.Concurrent+                       Pipes.Misc.State.Lazy+                       Pipes.Misc.State.Strict+                       Pipes.Misc.Time+                       Pipes.Misc.Util   build-depends:       base >= 4.7 && < 5+                     , clock >= 0.7 && < 1                      , lens >= 4 && < 5+                     , Decimal >= 0.4 && < 1                      , mtl >= 2 && < 3                      , pipes >= 4 && < 5                      , pipes-category >= 0.2 && < 0.3@@ -34,8 +41,11 @@   main-is:             Spec.hs   build-depends:       base >= 4.7 && < 5                      , lens >=4 &&< 5+                     , mmorph >= 1 && < 2                      , pipes >= 4 && < 5+                     , pipes-concurrency >= 2 && < 3                      , pipes-misc+                     , stm >= 2.4 && < 3                      , transformers >= 0.4 && < 0.6                      , hspec >= 2 && < 3   ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N

@@ -1,130 +1,11 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RankNTypes #-}---- | Miscellaneous utilities for pipes, required by glazier-tutorial-module Pipes.Misc where--import Control.Applicative-import Control.Arrow-import Control.Concurrent-import Control.Concurrent.STM-import Control.Lens-import Control.Monad-import Control.Monad.State.Strict-import Control.Monad.Trans.Maybe-import Control.Monad.Except-import qualified Data.List.NonEmpty as NE-import qualified Pipes as P-import qualified Pipes.Concurrent as PC-import qualified Pipes.Prelude as PP-import qualified Pipes.Shaft as PS---- | Like Pipes.Concurrent.fromInput, but stays in STM.--- Using @hoist atomically@ to convert to IO monad seems to work.--- Do not use @unsafeHoist atomically@.-fromInputSTM :: PC.Input a -> P.Producer' a STM ()-fromInputSTM as = void $ runMaybeT $ forever $ do-    a <- MaybeT $ lift $ PC.recv as-    lift $ P.yield a---- | Like Pipes.Concurrent.toOutput, but stays in STM.--- Using @hoist atomically@ to convert to IO monad seems to work.--- Do not use @unsafeHoist atomically@.-toOutputSTM :: PC.Output a -> P.Consumer' a STM ()-toOutputSTM output = void $ runMaybeT $ forever $ do-    a <- lift P.await-    p <- lift $ lift $ PC.send output a-    guard p---- | Reads as much as possible from an input and return a list of all unblocked values read.--- Blocks if the first value read is blocked.-batch :: PC.Input a -> PC.Input (NE.NonEmpty a)-batch (PC.Input xs) = PC.Input $ do-    x <- xs-    case x of-        Nothing -> pure Nothing-        Just x' -> do-            xs' <- runExceptT . tryNext $ x' NE.:| []-            case xs' of-                Left ys -> pure (Just ys)-                Right ys -> pure (Just ys)-  where-      tryNext ys = do-          ys' <- ExceptT $ (tryCons ys <$> xs) <|> pure (Left ys)-          tryNext ys'-      tryCons ys x = case x of-          Nothing -> Left ys -- return successful reads so far-          Just x' -> Right $ x' NE.<| ys---- | Given a size and a initial tail, create a pipe that--- will buffer the output of a producer.--- This pipe is stateful, and will only buffer until the immediate connecting--- producer is finished.--- @--- forever $ do---   a <- await---   yield a >-> buffer 2 [] -- will only ever result a producer of single 'a : []'.--- @--- @--- (forever $ do---   a <- await---   yield a--- ) >-> buffer 2 [] -- will buffer properly and produce '[latest, old]'--- @-buffer :: Monad m => Int -> [a] -> P.Pipe a [a] m r-buffer n as = do-  a <- P.await-  let as' = take n $ a : as-  case forceSpine as' of -- TODO: can we leave this lazy?-    () -> do-      P.yield as'-      buffer n as'- where-  -- from https://ro-che.info/articles/2015-05-28-force-list-  forceSpine = foldr (const id) ()---- | Store the output of the pipe into a MonadState.-store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r-store v s = forever $ do-  a <- P.await-  s .= view v a-  P.yield a---- | Yields a view into the stored value.-retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r-retrieve v = forever $ do-  a <- P.await-  s <- get-  P.yield (view v s, a)---- | Run a pipe in a larger stream, using view function and modify function--- of the larger stream.-locally ::-  Monad m =>-     (s -> a)-  -> (b -> s -> t)-  -> P.Pipe a b m r-  -> P.Pipe s t m r-locally viewf modifyf p =-  PP.map (\s -> (s, s))-  P.>-> PS.runShaft (first $ PS.Shaft $ PP.map viewf P.>-> p)-  P.>-> PP.map (uncurry modifyf)---- | Do something with the state everytime there is a yield.-onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r-onState f = PP.mapM $ \a -> do-    s <- get-    f s-    pure a+module Pipes.Misc+    ( module Pipes.Misc.Concurrent+    , module Pipes.Misc.State.Strict+    , module Pipes.Misc.Time+    , module Pipes.Misc.Util+    ) where --- | Add a delay after every yield--- To avoid delaying the first yield use:------ @--- Pipes.pull () >> delay d--- @----delay :: MonadIO io => Int -> P.Pipe a a io ()-delay i = PP.mapM $ \a -> do-    liftIO $ threadDelay i-    pure a+import Pipes.Misc.Concurrent+import Pipes.Misc.State.Strict+import Pipes.Misc.Time+import Pipes.Misc.Util

@@ -0,0 +1,55 @@+{-# LANGUAGE RankNTypes #-}++module Pipes.Misc.Concurrent where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Monad+import Control.Monad.Trans.Class+import Control.Monad.Trans.Maybe+import qualified Pipes as P+import qualified Pipes.Concurrent as PC++-- | Like Pipes.Concurrent.fromInput, but stays in STM.+-- Using @hoist atomically@ to convert to IO monad seems to work.+-- Do not use @unsafeHoist atomically@.+-- Each transaction is `atomically` scoped around each yield,+-- so be careful when `Pipes.Prelude.filter` or similar pipes to remove yields+-- as this results in larger transactions and it may cause BlockIndefinitelyOnSTM exceptions.+-- Intead, use Monoids to yield mempty so that the STM state changes.+fromInputSTM :: PC.Input a -> P.Producer' a STM ()+fromInputSTM as = void $ runMaybeT $ forever $ do+    a <- MaybeT $ lift $ PC.recv as+    lift $ P.yield a+{-# INLINABLE fromInputSTM #-}++-- | Like Pipes.Concurrent.toOutput, but stays in STM.+-- Using @hoist atomically@ to convert to IO monad seems to work.+-- Do not use @unsafeHoist atomically@.+toOutputSTM :: PC.Output a -> P.Consumer' a STM ()+toOutputSTM output = void $ runMaybeT $ forever $ do+    a <- lift P.await+    p <- lift $ lift $ PC.send output a+    guard p+{-# INLINABLE toOutputSTM #-}++-- | Convert PC.Output @a -> STM Bool@ to @a -> MaybeT STM ()@+toOutputMaybeT :: PC.Output a -> a -> MaybeT STM ()+toOutputMaybeT output = (MaybeT . fmap guard) <$> PC.send output+{-# INLINABLE toOutputMaybeT #-}++-- | Converts a Producer in IO monad to a producer in STM monad.+mkProducerSTM :: PC.Buffer a -> P.Producer a IO () -> IO (P.Producer a STM ())+mkProducerSTM b xs = do+    (output, input) <- PC.spawn b+    void . forkIO . void . forever . P.runEffect $ xs P.>-> PC.toOutput output+    pure (fromInputSTM input)+{-# INLINABLE mkProducerSTM #-}++-- | Converts a Producer in IO monad to a producer in STM monad. Also returns the seal.+mkProducerSTM' :: PC.Buffer a -> P.Producer a IO () -> IO (STM (), P.Producer a STM ())+mkProducerSTM' b xs = do+    (output, input, seal) <- PC.spawn' b+    void . forkIO . void . forever . P.runEffect $ xs P.>-> PC.toOutput output+    pure (seal, fromInputSTM input)+{-# INLINABLE mkProducerSTM' #-}

@@ -0,0 +1,40 @@+{-# LANGUAGE RankNTypes #-}++module Pipes.Misc.State.Lazy where++import Control.Lens+import Control.Monad.State.Lazy+import qualified Pipes as P+import qualified Pipes.Prelude as PP++-- | Store the output of the pipe into a MonadState.+store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r+store v s = forever $ do+  a <- P.await+  s .= view v a+  P.yield a+{-# INLINABLE store #-}++-- | Yields a view into the stored value.+retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r+retrieve v = forever $ do+  a <- P.await+  s <- get+  P.yield (view v s, a)+{-# INLINABLE retrieve #-}++-- | Yields a view into the stored value+retrieve' :: MonadState s m => Getter s b -> P.Pipe () b m r+retrieve' v = forever $ do+  P.await+  s <- get+  P.yield (view v s)+{-# INLINABLE retrieve' #-}++-- | Do something with the state everytime there is a yield.+onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r+onState f = PP.mapM $ \a -> do+    s <- get+    f s+    pure a+{-# INLINABLE onState #-}

@@ -0,0 +1,40 @@+{-# LANGUAGE RankNTypes #-}++module Pipes.Misc.State.Strict where++import Control.Lens+import Control.Monad.State.Strict+import qualified Pipes as P+import qualified Pipes.Prelude as PP++-- | Store the output of the pipe into a MonadState.+store :: MonadState s m => Getter a b -> Setter' s b -> P.Pipe a a m r+store v s = forever $ do+  a <- P.await+  s .= view v a+  P.yield a+{-# INLINABLE store #-}++-- | Yields a view into the stored value.+retrieve :: MonadState s m => Getter s b -> P.Pipe a (b, a) m r+retrieve v = forever $ do+  a <- P.await+  s <- get+  P.yield (view v s, a)+{-# INLINABLE retrieve #-}++-- | Yields a view into the stored value+retrieve' :: MonadState s m => Getter s b -> P.Pipe () b m r+retrieve' v = forever $ do+  P.await+  s <- get+  P.yield (view v s)+{-# INLINABLE retrieve' #-}++-- | Do something with the state everytime there is a yield.+onState :: (MonadState s m) => (s -> m ()) -> P.Pipe a a m r+onState f = PP.mapM $ \a -> do+    s <- get+    f s+    pure a+{-# INLINABLE onState #-}

@@ -0,0 +1,71 @@+{-# LANGUAGE RankNTypes #-}++-- | You can use the Arrow instance to get different types of ticker time. Eg:+--+-- @+-- diffTimeEvery :: MonadIO io => C.Clock -> Int -> P.Producer' C.TimeSpec io r+-- diffTimeEvery clock micros = always () P.>-> delay micros P.>-> ticker clock P.>-> diffTime+--+-- import Control.Arrow+-- import qualified Control.Category as Cat+-- import qualified Pipes.Shaft as PS+--+-- diffAndTickEvery :: MonadIO io => C.Clock -> Int -> P.Producer' (C.TimeSpec, C.TimeSpec) io r+-- diffAndTickEvery clock micros = always () P.>-> delay micros P.>-> ticker clock P.>->+--    PS.runShaft (PS.Shaft diffTime &&& Cat.id)+--+-- @+--+module Pipes.Misc.Time where++import Control.Concurrent+import Control.Monad.IO.Class+import qualified Pipes as P+import qualified Pipes.Prelude as PP+import qualified System.Clock as C+import qualified Data.Decimal as D+import qualified Pipes.Misc.Util as PM++-- | Add a delay after every await+delay :: MonadIO io => Int -> P.Pipe a a io r+delay i = PP.mapM $ \a -> do+    liftIO $ threadDelay i+    pure a+{-# INLINABLE delay #-}++-- | After the first await, add a delay after every subsequent await.+delay' :: MonadIO io => Int -> P.Pipe a a io r+delay' i = do+    a <- P.await+    P.yield a+    delay i+{-# INLINABLE delay' #-}++-- | obtain the threadDelay given a fps+fps :: D.Decimal -> Int+fps x = truncate (D.roundTo 0 (D.Decimal 0 1000000 / x))+{-# INLINABLE fps #-}++-- | Continuously yield the clock time+-- Use with delay to reduce the yield rate. Eg:+--+-- @+-- tickEvery :: MonadIO io => C.Clock -> Int -> P.Producer' C.TimeSpec io r+-- tickEvery clock micros = always () P.>-> delay micros P.>-> ticker clock+-- @+--+ticker :: MonadIO io => C.Clock -> P.Pipe () C.TimeSpec io r+ticker clock = P.for P.cat $ \() -> do+    t <- liftIO $ C.getTime clock+    P.yield t+{-# INLINABLE ticker #-}++-- | Converts a stream of times, into a stream of delta time. The first yield is zero.+diffTime :: Monad m => P.Pipe C.TimeSpec C.TimeSpec m r+diffTime = PM.compare C.diffTimeSpec (C.TimeSpec 0 0)+{-# INLINABLE diffTime #-}++-- | Converts a stream of epoch times into a stream of epoch time, where zero is the first yielded time.+resetEpoch :: Monad m => P.Pipe C.TimeSpec C.TimeSpec m r+resetEpoch = PM.compare' C.diffTimeSpec+{-# INLINABLE resetEpoch #-}

@@ -0,0 +1,120 @@+module Pipes.Misc.Util where++import Control.Applicative+import Control.Arrow+import Control.Monad.Except+import qualified Data.List.NonEmpty as NE+import qualified Pipes as P+import qualified Pipes.Concurrent as PC+import qualified Pipes.Prelude as PP+import qualified Pipes.Shaft as PS+import Pipes.Internal (Proxy(..))+++-- | Reads as much as possible from an input and return a list of all unblocked values read.+-- Blocks if the first value read is blocked.+batch :: PC.Input a -> PC.Input (NE.NonEmpty a)+batch (PC.Input xs) = PC.Input $ do+    x <- xs+    case x of+        Nothing -> pure Nothing+        Just x' -> do+            xs' <- runExceptT . tryNext $ x' NE.:| []+            case xs' of+                Left ys -> pure (Just ys)+                Right ys -> pure (Just ys)+  where+      tryNext ys = do+          ys' <- ExceptT $ (tryCons ys <$> xs) <|> pure (Left ys)+          tryNext ys'+      tryCons ys x = case x of+          Nothing -> Left ys -- return successful reads so far+          Just x' -> Right $ x' NE.<| ys+{-# INLINABLE batch #-}++-- | Given a size and a initial tail, create a pipe that+-- will buffer the output of a producer.+-- This pipe is stateful, and will only buffer until the immediate connecting+-- producer is finished.+-- @+-- forever $ do+--   a <- await+--   yield a >-> buffer 2 [] -- will only ever result a producer of single 'a : []'.+-- @+-- @+-- (forever $ do+--   a <- await+--   yield a+-- ) >-> buffer 2 [] -- will buffer properly and produce '[latest, old]'+-- @+buffer :: Monad m => Int -> [a] -> P.Pipe a [a] m r+buffer n as = do+  a <- P.await+  let as' = take n $ a : as+  case forceSpine as' of -- TODO: can we leave this lazy?+    () -> do+      P.yield as'+      buffer n as'+ where+  -- from https://ro-che.info/articles/2015-05-28-force-list+  forceSpine = foldr (const id) ()+{-# INLINABLE buffer #-}++-- | Run a pipe in a larger stream, using view function and modify function+-- of the larger stream.+locally ::+  Monad m =>+     (s -> a)+  -> (b -> s -> t)+  -> P.Pipe a b m r+  -> P.Pipe s t m r+locally viewf modifyf p =+  PP.map (\s -> (s, s))+  P.>-> PS.runShaft (first $ PS.Shaft $ PP.map viewf P.>-> p)+  P.>-> PP.map (uncurry modifyf)+{-# INLINABLE locally #-}++-- | Given comparison function and an initial value.+-- yield the result of comparing the value await with the previously awaited value.+compare :: Monad m => (a -> a -> b) -> a -> P.Pipe a b m r+compare f i = do+    a <- P.await+    P.yield (f a i)+    go a+  where+    go a = do+        b <- P.await+        P.yield (f b a)+        go b+{-# INLINABLE compare #-}++-- | Given comparison function+-- yield the result of comparing the value await with the first awaited value.+compare' :: Monad m => (a -> a -> b) -> P.Pipe a b m r+compare' f = do+    i <- P.await+    P.yield (f i i)+    go i+  where+    go i = forever $ do+        a <- P.await+        P.yield (f a i)+{-# INLINABLE compare' #-}++-- | constantly yields the given value+always :: Monad m => a -> P.Producer a m r+always = forever . P.yield+{-# INLINABLE always #-}++-- | Makes the Producer return/pure the last value yielded, or the input value if nothing+-- was yielded+lastOr :: Monad m => a -> P.Producer a m () -> P.Producer a m a+lastOr = go+  where+    go i p =+        case p of+            Request a' fa -> Request a' (go i . fa)+            Respond b fb' -> Respond b (go b . fb')+            M m -> M (m >>= pure . go i)+            Pure () -> Pure i+{-# INLINABLE lastOr #-}

@@ -1,8 +1,11 @@ module Main where +import Control.Concurrent.STM import Control.Lens+import Control.Monad.Morph import Data.Foldable import qualified Pipes as P+import qualified Pipes.Concurrent as PC import qualified Pipes.Misc as PM import qualified Pipes.Prelude as PP import Test.Hspec@@ -27,3 +30,6 @@                         P.>-> PP.map (\a -> (a, a))                         P.>-> PM.locally (view _2) (set _2) (PP.map (+ 10)))               xs `shouldBe` zip data1 ((+ 10) <$> data1)+          it "Filtering STM Producer blocks" $ do+              sig <- PM.mkProducerSTM (PC.bounded 1) sig1+              PP.toListM (hoist atomically (sig P.>-> PP.filter even)) `shouldThrow` anyException