pipes-misc 0.2.3.0 → 0.2.4.0

raw patch · 3 files changed

+29/−27 lines, 3 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• pipes-misc.cabal +1/−1
• src/Pipes/Misc/Concurrent.hs +24/−0
• src/Pipes/Misc/Util.hs +4/−26

@@ -1,5 +1,5 @@ name:                pipes-misc-version:             0.2.3.0+version:             0.2.4.0 synopsis:            Miscellaneous utilities for pipes, required by glazier-tutorial description:         Please see README.md homepage:            https://github.com/louispan/pipes-misc#readme

@@ -2,11 +2,14 @@  module Pipes.Misc.Concurrent where +import Control.Applicative import Control.Concurrent import Control.Concurrent.STM import Control.Monad import Control.Monad.Trans.Class+import Control.Monad.Trans.Except import Control.Monad.Trans.Maybe+import qualified Data.List.NonEmpty as NE import qualified Pipes as P import qualified Pipes.Concurrent as PC @@ -53,3 +56,24 @@     void . forkIO . void . forever . P.runEffect $ xs P.>-> PC.toOutput output     pure (seal, fromInputSTM input) {-# INLINABLE mkProducerSTM' #-}++-- | 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 #-}

@@ -1,52 +1,30 @@ module Pipes.Misc.Util where -import Control.Applicative+import Control.Monad 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