packages feed

streamly-examples 0.1.2 → 0.1.3

raw patch · 33 files changed

+915/−471 lines, 33 filesdep +streamly-coredep +tasty-benchdep ~basedep ~fusion-plugindep ~hashablenew-component:exe:DateTimeParsernew-component:exe:LogParser

Dependencies added: streamly-core, tasty-bench

Dependency ranges changed: base, fusion-plugin, hashable, streamly, transformers, vector

Files

Changelog.md view
@@ -1,5 +1,9 @@ # Changelog +## 0.1.3 (Mar 2023)++* Support streamly-0.9.0 and streamly-core-0.1.0+ ## 0.1.2 (Mar 2022)  * Support streamly-0.8.2
README.md view
@@ -11,16 +11,18 @@  ## Running The Examples -Executable names are the same as the filenames.  To run an example:+When running the unstable version (downloaded from the git repository)+you must use `--project-file cabal.project.user` option otherwise the+build might fail. For example:  ```-$ cabal run AcidRain+$ cabal run --project-file cabal.project.user AcidRain ``` -For performance sensitive examples use fusion-plugin for best performance:+Executable names are the same as the filenames.  To run an example:  ```-$ cabal run --flag fusion-plugin WordCount -- streamly-examples.cabal+$ cabal run AcidRain ```  To run SDL2 based examples, make sure that you have the OS package for@@ -43,7 +45,8 @@  * [Intro](examples/Intro.hs): Simple, introductory examples - loops, text   processing, networking, concurrency.-* [MergeSort](examples/MergeSort.hs): Merge sorted streams concurrently.+* [MergeSort](examples/MergeSort.hs): Sort a stream concurrently using merge+  sort. * [Rate](examples/Rate.hs): Run an action at a given rate.  ### FileSystem
examples/AcidRain.hs view
@@ -7,20 +7,17 @@ -- This example is adapted from Gabriel Gonzalez's pipes-concurrency package. -- https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html -#if !(MIN_VERSION_base(4,11,0))-import Data.Semigroup ((<>))-#endif-import Control.Monad (void) import Control.Monad.IO.Class (MonadIO(liftIO))-import Control.Monad.State (MonadState, get, modify, runStateT)+import Control.Monad.State (MonadState, get, modify) import Data.Function ((&))-import Streamly.Prelude (MonadAsync, SerialT)+import Streamly.Data.Stream.Prelude (MonadAsync, Stream) -import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream -data Event = Quit | Harm Int | Heal Int deriving (Show)+data Event = Quit | Harm Int | Heal Int deriving (Eq, Show) -userAction :: MonadAsync m => SerialT m Event+userAction :: MonadAsync m => Stream m Event userAction = Stream.repeatM $ liftIO askUser     where     askUser = do@@ -31,22 +28,26 @@             "quit"   -> return Quit             _        -> putStrLn "Type potion or harm or quit" >> askUser -acidRain :: MonadAsync m => SerialT m Event-acidRain =-      Stream.repeatM (liftIO $ return $ Harm 1) -- AsyncT m Event-    & Stream.constRate 1                        -- AsyncT m Event-    & Stream.fromAsync                          -- SerialT m Event+acidRain :: MonadAsync m => Stream m Event+acidRain = Stream.parRepeatM (Stream.constRate 1) (return $ Harm 1) +parallel :: MonadAsync m => [Stream m a] -> Stream m a+parallel = Stream.parList (Stream.eager True)+ data Result = Check | Done -runEvents :: (MonadAsync m, MonadState Int m) => SerialT m Result-runEvents = do-    event <- userAction `Stream.parallel` acidRain -- SerialT m Event-    case event of-        Harm n -> modify (\h -> h - n) >> return Check-        Heal n -> modify (\h -> h + n) >> return Check-        Quit -> return Done+runEvents :: (MonadAsync m, MonadState Int m) => Stream m Result+runEvents =+    Stream.mapM f $ parallel [userAction, acidRain] +    where++    f event =+        case event of+            Harm n -> modify (\h -> h - n) >> return Check+            Heal n -> modify (\h -> h + n) >> return Check+            Quit -> return Done+ data Status = Alive | GameOver deriving Eq  getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status@@ -55,15 +56,17 @@         Done  -> liftIO $ putStrLn "You quit!" >> return GameOver         Check -> do             h <- get-            liftIO $ if (h <= 0)-                     then putStrLn "You die!" >> return GameOver-                     else putStrLn ("Health = " <> show h) >> return Alive+            liftIO+                $ if (h <= 0)+                  then putStrLn "You die!" >> return GameOver+                  else putStrLn ("Health = " <> show h) >> return Alive  main :: IO () main = do     putStrLn "Your health is deteriorating due to acid rain,\\              \ type \"potion\" or \"quit\""-    let runGame =-            Stream.mapM getStatus runEvents -- SerialT (StateT Int IO) Status-          & Stream.drainWhile (== Alive)    -- StateT Int IO ()-    void $ runStateT runGame 60+    Stream.mapM getStatus runEvents  -- Stream (StateT Int IO) Status+        & Stream.runStateT (pure 60) -- Stream IO (Int, Status)+        & fmap snd                   -- Stream IO Status+        & Stream.fold (Fold.takeEndBy (/= Alive) Fold.drain) -- IO ()+    return ()
examples/CSVParser.hs view
@@ -4,32 +4,34 @@ import Data.Char (chr) import Data.Function ((&)) import Data.Word (Word8)-import Streamly.Prelude (SerialT)+import Streamly.Data.Stream (Stream) import System.Environment (getArgs) import System.IO (IOMode(..)) -import qualified Streamly.Data.Array.Foreign as Array-import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.FileSystem.Handle as Handle import qualified System.IO as IO-import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream+import qualified Streamly.Internal.Data.Stream as Stream (splitOn)+import qualified Streamly.Internal.Data.Stream.Chunked as ArrayStream (splitOn)  main :: IO () main = do     inFile <- fmap head getArgs     src <- IO.openFile inFile ReadMode -    Stream.unfold Handle.readChunks src -- SerialT IO (Array Word8)-        & ArrayStream.splitOn 10        -- SerialT IO (Array Word8)-        & Stream.mapM_ parseLine        -- IO ()+    Stream.unfold Handle.chunkReader src         -- Stream IO (Array Word8)+        & ArrayStream.splitOn 10                 -- Stream IO (Array Word8)+        & Stream.fold (Fold.drainMapM parseLine) -- IO ()      where      printList = putStr . map (chr . fromIntegral)+     parseLine arr =-        (Stream.unfold Array.read arr :: SerialT IO Word8)-            & Stream.splitOn (== 44) Fold.toList -- SerialT IO [Word8]-            & Stream.intersperse [32]            -- SerialT IO [Word8]-            & Stream.mapM_ printList             -- IO ()+        (Stream.unfold Array.reader arr :: Stream IO Word8)+            & Stream.splitOn (== 44) Fold.toList     -- Stream IO [Word8]+            & Stream.intersperse [32]                -- Stream IO [Word8]+            & Stream.fold (Fold.drainMapM printList) -- IO ()         >> putStrLn ""
examples/CamelCase.hs view
@@ -6,15 +6,16 @@ import System.Environment (getArgs) import System.IO (Handle, IOMode(..), openFile, stdout) -import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream import qualified Streamly.FileSystem.Handle as Handle  -- | @camelCase source-file dest-file@ camelCase :: Handle -> Handle -> IO () camelCase src dst =-      Stream.unfold Handle.read src      -- SerialT IO Word8-    & Stream.scanl' step (True, Nothing) -- SerialT IO (Bool, Maybe Word8)-    & Stream.mapMaybe snd                -- SerialT IO Word8+      Stream.unfold Handle.reader src    -- Stream IO Word8+    & Stream.postscan fold               -- Stream IO (Bool, Maybe Word8)+    & Stream.mapMaybe snd                -- Stream IO Word8     & Stream.fold (Handle.write dst)     -- IO ()      where@@ -33,6 +34,8 @@         -- Convert lower to upper if preceded by whitespace         | isLower x = (False, Just $ if wasSpace then x - 32 else x)         | otherwise = (True, Nothing)++    fold = Fold.foldl' step (True, Nothing)  main :: IO () main = do
examples/CirclingSquare.hs view
@@ -77,8 +77,8 @@     ref <- newIORef (0,0)      _ <- forkIO $ do-            Stream.repeatM (updateDisplay ref rend) -- SerialT IO ()-                & Stream.delay (1/60)               -- SerialT IO ()+            Stream.repeatM (updateDisplay ref rend) -- Stream IO ()+                & Stream.delay (1/60)               -- Stream IO ()                 & Stream.drain                      -- IO ()      -- MacOS requires pollEvents to run in the Main thread.
examples/CmdClient.hs view
@@ -4,12 +4,12 @@ import Data.Function ((&)) import Data.Word (Word8) import Network.Socket (PortNumber)-import Streamly.Prelude (SerialT)+import Streamly.Data.Stream.Prelude (Stream)  import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Unfold as Unfold-import qualified Streamly.Internal.Network.Inet.TCP as TCP (processBytes)-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.Internal.Network.Inet.TCP as TCP (pipeBytes) import qualified Streamly.Unicode.Stream as Unicode  remoteAddr :: (Word8,Word8,Word8,Word8)@@ -31,20 +31,28 @@         putStrLn $ tag ++ show (i * chunkSize)     return i -sender :: SerialT IO ()+sender :: Stream IO () sender =-      Stream.repeat "time\nrandom\n"               -- SerialT IO String-    & Stream.unfoldMany Unfold.fromList            -- SerialT IO Char-    & Unicode.encodeLatin1                         -- SerialT IO Word8-    & TCP.processBytes remoteAddr remotePort       -- SerialT IO Word8-    & Unicode.decodeLatin1                         -- SerialT IO Char-    & Stream.splitOnSuffix (== '\n') Fold.drain    -- SerialT IO String-    & Stream.chunksOf chunkSize Fold.drain         -- SerialT IO ()+      Stream.repeat "time\nrandom\n"               -- Stream IO String+    & Stream.unfoldMany Unfold.fromList            -- Stream IO Char+    & Unicode.encodeLatin1                         -- Stream IO Word8+    & TCP.pipeBytes remoteAddr remotePort          -- Stream IO Word8+    & Unicode.decodeLatin1                         -- Stream IO Char+    & split (== '\n') Fold.drain                   -- Stream IO String+    & Stream.foldMany chunk                        -- Stream IO () +    where++    chunk = Fold.take chunkSize Fold.drain+    split p f = Stream.foldMany (Fold.takeEndBy_ p f)+ main :: IO () main = do-      Stream.replicate 4 sender                    -- SerialT IO (SerialT IO ())-    & Stream.concatMapWith Stream.async id         -- SerialT IO ()-    & Stream.postscanlM' (counter "rcvd: ")-        (return 0 :: IO Int)                       -- SerialT IO Int-    & Stream.drain                                 -- IO ()+      Stream.replicate 4 sender                    -- Stream IO (Stream IO ())+    & Stream.parConcat id                          -- Stream IO ()+    & Stream.postscan counts                       -- Stream IO Int+    & Stream.fold Fold.drain                       -- IO ()++    where++    counts = Fold.foldlM' (counter "rcvd") (return 0 :: IO Int)
examples/CmdServer.hs view
@@ -14,20 +14,21 @@ import Control.Monad.Catch (catch, SomeException) import Data.Char (isSpace) import Data.Function ((&))+import Data.Map.Strict (Map) import Network.Socket (Socket) import Streamly.Data.Fold (Fold) import System.Random (randomIO) -import qualified Data.Map.Strict as Map+import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Parser as Parser+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Network.Inet.TCP as TCP import qualified Streamly.Network.Socket as Socket-import qualified Streamly.Prelude as Stream import qualified Streamly.Unicode.Stream as Unicode -import qualified Streamly.Internal.Data.Fold as Fold (demuxDefault)-import qualified Streamly.Internal.Data.Time.Clock as Clock (getTime, Clock(..))+import qualified Streamly.Internal.Data.Time.Clock as Clock+    (getTime, Clock(..))  ------------------------------------------------------------------------------ -- Utility functions@@ -38,7 +39,7 @@       Stream.fromList (show x ++ "\n")     & Unicode.encodeLatin1     & Stream.fold (Array.writeN 60)-    >>= Socket.writeChunk sk+    >>= Socket.putChunk sk  ------------------------------------------------------------------------------ -- Command Handlers@@ -50,17 +51,23 @@ random :: Socket -> IO () random sk = (randomIO :: IO Int) >>= sendValue sk -def :: (String, Socket) -> IO ()-def (str, sk) = sendValue sk ("Unknown command: " ++ str)+def :: String -> Socket -> IO ()+def str sk = sendValue sk ("Unknown command: " ++ str) -commands :: Map.Map String (Fold IO Socket ())-commands = Map.fromList-    [ ("time"  , Fold.drainBy time)-    , ("random", Fold.drainBy random)-    ]+commands :: String -> IO (Fold IO Socket ())+commands cmd =+    case cmd of+        "time"    -> return (Fold.drainMapM time)+        "random"  -> return (Fold.drainMapM random)+        _         -> return (Fold.drainMapM (def cmd)) +{-# INLINE demuxKvToMap #-}+demuxKvToMap :: (Monad m, Ord k) =>+    (k -> m (Fold m a b)) -> Fold m (k, a) (Map k b)+demuxKvToMap f = Fold.demuxToMap fst (\(k, _) -> fmap (Fold.lmap snd) (f k))+ demux :: Fold IO (String, Socket) ()-demux = snd <$> Fold.demuxDefault commands (Fold.drainBy def)+demux = void (demuxKvToMap commands)  ------------------------------------------------------------------------------ -- Parse and handle commands on a socket@@ -68,15 +75,17 @@  handler :: Socket -> IO () handler sk =-      Stream.unfold Socket.read sk       -- SerialT IO Word8-    & Unicode.decodeLatin1               -- SerialT IO Char-    & Stream.wordsBy isSpace Fold.toList -- SerialT IO String-    & Stream.map (, sk)                  -- SerialT IO (String, Socket)+      Stream.unfold Socket.reader sk     -- Stream IO Word8+    & Unicode.decodeLatin1               -- Stream IO Char+    & Stream.parseMany word              -- Stream IO String+    & Stream.catRights+    & fmap (, sk)                        -- Stream IO (String, Socket)     & Stream.fold demux                  -- IO () + Exceptions     & discard                            -- IO ()      where +    word = Parser.wordBy isSpace Fold.toList     discard action = void action `catch` (\(_ :: SomeException) -> return ())  ------------------------------------------------------------------------------@@ -85,11 +94,9 @@  server :: IO () server =-      Stream.unfold TCP.acceptOnPort 8091      -- SerialT IO Socket-    & Stream.fromSerial                        -- AsyncT IO Socket-    & Stream.mapM (Socket.forSocketM handler)  -- AsyncT IO ()-    & Stream.fromAsync                         -- SerialT IO ()-    & Stream.drain                             -- IO ()+      Stream.unfold TCP.acceptorOnPort 8091          -- Stream IO Socket+    & Stream.parMapM id (Socket.forSocketM handler)  -- Stream IO ()+    & Stream.fold Fold.drain                         -- IO ()  main :: IO () main = server
examples/ControlFlow.hs view
@@ -1,6 +1,6 @@-{-# LANGUAGE CPP                 #-} {-# LANGUAGE FlexibleContexts    #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-orphans #-}  ------------------------------------------------------------------------------- -- Combining control flow manipulating monad transformers (MaybeT, exceptT,@@ -15,12 +15,8 @@ -- This file provides an example where we enter a sequence of characters "x", -- and "y" on separate lines, on the command line. When any other sequence is -- entered the control flow short circuits at the first non-matching char and- -- exits. -#if !(MIN_VERSION_base(4,11,0))-import Data.Semigroup (Semigroup(..))-#endif import Control.Concurrent (threadDelay) import Control.Exception (catch, SomeException) import Control.Monad (when, mzero)@@ -30,9 +26,11 @@ import Control.Monad.Trans.Maybe (MaybeT (..)) import Control.Monad.Trans.Except (ExceptT, runExceptT, throwE, catchE) import Control.Monad.Trans.Cont (ContT(..), callCC)-import Streamly.Prelude (IsStream)+import Streamly.Data.Stream (Stream)+import Streamly.Internal.Data.Stream.StreamD (CrossStream, mkCross, unCross) -import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream  ------------------------------------------------------------------------------- -- Using MaybeT below streamly@@ -41,17 +39,11 @@ -- When streamly is on top MaybeT would terminate all iterations of -- non-determinism. ---getSequenceMaybeBelow-    :: ( IsStream t-       , Monad m-       , MonadTrans t-       , MonadIO (t (MaybeT m))-       )-    => t (MaybeT m) ()-getSequenceMaybeBelow = do+getSequenceMaybeBelow :: MonadIO m => Stream (MaybeT m) ()+getSequenceMaybeBelow = unCross $ do     liftIO $ putStrLn "MaybeT below streamly: Enter one char per line: " -    i <- Stream.fromFoldable [1..2 :: Int]+    i <- mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -62,7 +54,7 @@  mainMaybeBelow :: IO () mainMaybeBelow = do-    r <- runMaybeT (Stream.drain getSequenceMaybeBelow)+    r <- runMaybeT (Stream.fold Fold.drain getSequenceMaybeBelow)     case r of         Just _ -> putStrLn "Bingo"         Nothing -> putStrLn "Wrong"@@ -70,18 +62,18 @@ ------------------------------------------------------------------------------- -- Using MaybeT above streamly ----------------------------------------------------------------------------------+ -- When MaybeT is on top a Nothing would terminate only the current iteration -- of non-determinism below. -- -- Note that this is redundant configuration as the same behavior can be -- achieved with just streamly, using mzero. ---getSequenceMaybeAbove :: (IsStream t, MonadIO (t m)) => MaybeT (t m) ()+getSequenceMaybeAbove :: MonadIO m => MaybeT (CrossStream m) () getSequenceMaybeAbove = do     liftIO $ putStrLn "MaybeT above streamly: Enter one char per line: " -    i <- lift $ Stream.fromFoldable [1..2 :: Int]+    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -90,7 +82,7 @@     r2 <- liftIO getLine     when (r2 /= "y") mzero -mainMaybeAbove :: (IsStream t, MonadIO (t m)) => MaybeT (t m) ()+mainMaybeAbove :: MonadIO m => MaybeT (CrossStream m) () mainMaybeAbove = do     getSequenceMaybeAbove     liftIO $ putStrLn "Bingo"@@ -103,17 +95,11 @@ -- -- Note that throwE would terminate all iterations of non-determinism -- altogether.-getSequenceEitherBelow-    :: ( IsStream t-       , MonadTrans t-       , Monad m-       , MonadIO (t (ExceptT String m))-       )-    => t (ExceptT String m) ()-getSequenceEitherBelow = do+getSequenceEitherBelow :: MonadIO m => Stream (ExceptT String m) ()+getSequenceEitherBelow = unCross $ do     liftIO $ putStrLn "ExceptT below streamly: Enter one char per line: " -    i <- Stream.fromFoldable [1..2 :: Int]+    i <- mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -125,7 +111,7 @@ mainEitherBelow :: IO () mainEitherBelow = do     -- XXX Cannot lift catchE-    r <- runExceptT (Stream.drain getSequenceEitherBelow)+    r <- runExceptT (Stream.fold Fold.drain getSequenceEitherBelow)     case r of         Right _ -> liftIO $ putStrLn "Bingo"         Left s  -> liftIO $ putStrLn s@@ -133,30 +119,29 @@ ------------------------------------------------------------------------------- -- Using ExceptT below concurrent streamly ----------------------------------------------------------------------------------+ -- XXX does not work correctly yet ---getSequenceEitherAsyncBelow-    :: ( IsStream t-       , MonadTrans t-       , MonadIO m-       , MonadIO (t (ExceptT String m))-       , Semigroup (t (ExceptT String m) Integer)-       )-    => t (ExceptT String m) ()-getSequenceEitherAsyncBelow = do++getSequenceEitherAsyncBelow :: MonadIO m => Stream (ExceptT String m) ()+getSequenceEitherAsyncBelow = unCross $ do     liftIO $ putStrLn "ExceptT below concurrent streamly: " -    i <- (liftIO (threadDelay 1000)-            >> lift (throwE "First task")-            >> return 1)-            <> (lift (throwE "Second task") >> return 2)-            <> Stream.fromPure (3 :: Integer)+    i <- mkCross+            $ Stream.consM+                (liftIO (threadDelay 1000)+                    >> throwE "First task"+                    >> return 1+                )+                (Stream.consM+                    (throwE "Second task" >> return 2)+                    (Stream.fromPure (3 :: Integer))+                )     liftIO $ putStrLn $ "iteration = " <> show i  mainEitherAsyncBelow :: IO () mainEitherAsyncBelow = do-    r <- runExceptT (Stream.drain $ Stream.fromAsync getSequenceEitherAsyncBelow)+    r <- runExceptT (Stream.fold Fold.drain getSequenceEitherAsyncBelow)     case r of         Right _ -> liftIO $ putStrLn "Bingo"         Left s  -> liftIO $ putStrLn s@@ -173,12 +158,11 @@ -- -- Here we can use catchE directly but will have to use monad-control to lift -- stream operations with stream arguments.-getSequenceEitherAbove :: (IsStream t, MonadIO (t m))-    => ExceptT String (t m) ()+getSequenceEitherAbove :: MonadIO m => ExceptT String (CrossStream m) () getSequenceEitherAbove = do     liftIO $ putStrLn "ExceptT above streamly: Enter one char per line: " -    i <- lift $ Stream.fromFoldable [1..2 :: Int]+    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -187,7 +171,7 @@     r2 <- liftIO getLine     when (r2 /= "y") $ throwE $ "Expecting y got: " <> r2 -mainEitherAbove :: (IsStream t, MonadIO (t m)) => ExceptT String (t m) ()+mainEitherAbove :: MonadIO m => ExceptT String (CrossStream m) () mainEitherAbove =     catchE (getSequenceEitherAbove >> liftIO (putStrLn "Bingo"))            (liftIO . putStrLn)@@ -203,23 +187,26 @@ -- Note that unlike when ExceptT is used on top, MonadThrow terminates all -- iterations of non-determinism rather then just the current iteration. ---getSequenceMonadThrow :: (IsStream t, MonadIO (t m), MonadThrow (t m))-    => t m ()-getSequenceMonadThrow = do+getSequenceMonadThrow :: (MonadIO m, MonadThrow m) => Stream m ()+getSequenceMonadThrow = unCross $ do     liftIO $ putStrLn "MonadThrow in streamly: Enter one char per line: " -    i <- Stream.fromFoldable [1..2 :: Int]+    i <- mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine-    when (r1 /= "x") $ throwM $ Unexpected $ "Expecting x got: " <> r1+    when (r1 /= "x")+        $ mkCross+        $ Stream.fromEffect $ throwM $ Unexpected $ "Expecting x got: " <> r1      r2 <- liftIO getLine-    when (r2 /= "y") $ throwM $ Unexpected $ "Expecting y got: " <> r2+    when (r2 /= "y")+        $ mkCross+        $ Stream.fromEffect $ throwM $ Unexpected $ "Expecting y got: " <> r2  mainMonadThrow :: IO () mainMonadThrow =-    catch (Stream.drain getSequenceMonadThrow >> liftIO (putStrLn "Bingo"))+    catch (Stream.fold Fold.drain getSequenceMonadThrow >> liftIO (putStrLn "Bingo"))           (\(e :: SomeException) -> liftIO $ print e)  -------------------------------------------------------------------------------@@ -231,13 +218,11 @@ -- -- XXX need to have a specialized liftCallCC to actually lift callCC ---getSequenceContBelow-    :: (IsStream t, MonadTrans t, MonadIO m, MonadIO (t (ContT r m)))-    => t (ContT r m) (Either String ())+getSequenceContBelow :: MonadIO m => CrossStream (ContT r m) (Either String ()) getSequenceContBelow = do     liftIO $ putStrLn "ContT below streamly: Enter one char per line: " -    i <- Stream.fromFoldable [1..2 :: Int]+    i <- mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r <- lift $ callCC $ \exit -> do@@ -253,10 +238,8 @@     liftIO $ putStrLn $ "done iteration = " <> show i     return r -mainContBelow-    :: (IsStream t, MonadIO m, MonadTrans t, MonadIO (t (ContT r m)))-    => t (ContT r m) ()-mainContBelow = do+mainContBelow :: MonadIO m => Stream (ContT r m) ()+mainContBelow = unCross $ do     r <- getSequenceContBelow     case r of         Right _ -> liftIO $ putStrLn "Bingo"@@ -266,12 +249,11 @@ -- Using ContT above streamly ------------------------------------------------------------------------------- ---getSequenceContAbove :: (IsStream t, MonadIO (t m))-    => ContT r (t m) (Either String ())+getSequenceContAbove :: MonadIO m => ContT r (CrossStream m) (Either String ()) getSequenceContAbove = do     liftIO $ putStrLn "ContT above streamly: Enter one char per line: " -    i <- lift $ Stream.fromFoldable [1..2 :: Int]+    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      callCC $ \exit -> do@@ -285,7 +267,7 @@         then exit $ Left $ "Expecting y got: " <> r2         else return $ Right () -mainContAbove :: (IsStream t, MonadIO (t m)) => ContT r (t m) ()+mainContAbove :: MonadIO m => ContT r (CrossStream m) () mainContAbove = do     r <- getSequenceContAbove     case r of@@ -300,10 +282,10 @@ main :: IO () main = do     mainMaybeBelow-    Stream.drain $ runMaybeT mainMaybeAbove-    runContT (Stream.drain mainContBelow) return-    Stream.drain (runContT mainContAbove return)+    Stream.fold Fold.drain $ unCross $ runMaybeT mainMaybeAbove+    runContT (Stream.fold Fold.drain mainContBelow) return+    Stream.fold Fold.drain $ unCross $ runContT mainContAbove return     mainEitherBelow-    Stream.drain (runExceptT mainEitherAbove)+    Stream.fold Fold.drain $ unCross $ runExceptT mainEitherAbove     mainMonadThrow     mainEitherAsyncBelow
examples/CoreUtils.hs view
@@ -1,70 +1,68 @@ -- Implement some simple coreutils commands  import Control.Monad (void)-import Data.Char (ord) import Data.Function ((&)) import Data.Word (Word8) import System.Environment (getArgs)-import Streamly.Data.Array.Foreign (Array)+import Streamly.Data.Array (Array) import Streamly.Data.Fold (Fold)  import qualified Streamly.Console.Stdio as Stdio-import qualified Streamly.Data.Array.Foreign as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Internal.Data.Stream.IsStream as Stream-    (splitOnSeq, tapAsyncK)-import qualified Streamly.Internal.FileSystem.Dir as Dir (toFiles)+import qualified Streamly.Data.Stream as Stream++import qualified Streamly.Internal.FileSystem.Dir as Dir (readFiles) import qualified Streamly.Internal.FileSystem.File as File-import qualified Streamly.Prelude as Stream  -- | > cat input.txt cat :: IO () cat =-      File.toChunks "input.txt"     -- SerialT IO (Array Word8)+      File.readChunks "input.txt"     -- Stream IO (Array Word8)     & Stream.fold Stdio.writeChunks -- IO ()  -- | Read all files from a directory and write to a single output file. -- > cat dir/* > output.txt catDirTo :: IO () catDirTo =-      Dir.toFiles "dir"                 -- SerialT IO String-    & Stream.unfoldMany File.readChunks -- SerialT IO (Array Word8)-    & File.fromChunks "output.txt"      -- IO ()+      Dir.readFiles "dir"                -- Stream IO String+    & Stream.unfoldMany File.chunkReader -- Stream IO (Array Word8)+    & File.fromChunks "output.txt"       -- IO ()  -- | > cp input.txt output.txt cp :: IO () cp =-      File.toChunks "input.txt"    -- SerialT IO (Array Word8)-    & File.fromChunks "output.txt" -- IO ()+      File.readChunks "input.txt"    -- Stream IO (Array Word8)+    & File.fromChunks "output.txt"   -- IO ()  -- | > cat input.txt >> output.txt append :: IO () append =-      File.toChunks "input.txt"      -- SerialT IO (Array Word8)-    & File.appendChunks "output.txt" -- IO ()+      File.readChunks "input.txt"      -- Stream IO (Array Word8)+    & File.appendChunks "output.txt"   -- IO ()  -- | > cat input.txt | tee output1.txt > output.txt tap :: IO () tap =-      File.toChunks "input.txt"                   -- SerialT IO (Array Word8)-    & Stream.tap (File.writeChunks "output1.txt") -- SerialT IO (Array Word8)+      File.readChunks "input.txt"                   -- Stream IO (Array Word8)+    & Stream.tap (File.writeChunks "output1.txt") -- Stream IO (Array Word8)     & File.fromChunks "output.txt"                -- IO ()-+{- -- | > cat input.txt | tee output1.txt > output.txt -- output1.txt is processed/written to in a separate thread. tapAsync :: IO () tapAsync =-      Stream.unfold Stdio.readChunks ()   -- SerialT IO (Array Word8)+      Stream.unfold Stdio.readChunks ()   -- Stream IO (Array Word8)     & Stream.tapAsyncK-          (File.fromChunks "output1.txt") -- SerialT IO (Array Word8)+          (File.fromChunks "output1.txt") -- Stream IO (Array Word8)     & File.fromChunks "output.txt"        -- IO ()+-}  -- | > cat input.txt | tee output1.txt > output.txt tee :: IO () tee =-      File.toChunks "input.txt" -- SerialT IO (Array Word8)-    & Stream.fold t             -- IO ((),())-    & void                      -- IO ()+      File.readChunks "input.txt" -- Stream IO (Array Word8)+    & Stream.fold t               -- IO ((),())+    & void                        -- IO ()      where @@ -73,12 +71,12 @@     t = Fold.tee             (File.writeChunks "output1.txt") -- Fold IO (Array Word8) ()             (File.writeChunks "output.txt")  -- Fold IO (Array Word8) ()-+{- -- | > grep -c "the" input.txt grepc :: IO () grepc = do-    File.toBytes "input.txt"                                -- SerialT IO Word8-        & Stream.splitOnSeq (Array.fromList pat) Fold.drain -- SerialT IO ()+    File.toBytes "input.txt"                                -- Stream IO Word8+        & Stream.splitOnSeq (Array.fromList pat) Fold.drain -- Stream IO ()         & Stream.length                                     -- IO Int         >>= print . subtract 1                              -- IO () @@ -86,6 +84,7 @@      pat :: [Word8]     pat = map (fromIntegral . ord) "the"+-}  main :: IO () main = do@@ -97,7 +96,7 @@         "cp" -> putStrLn "cp" >> cp         "append" -> putStrLn "append" >> append         "tap" -> putStrLn "tap" >> tap-        "tapAsync" -> putStrLn "tapAsync" >> tapAsync+       -- "tapAsync" -> putStrLn "tapAsync" >> tapAsync         "tee" -> putStrLn "tee" >> tee-        "grepc" -> putStrLn "grepc" >> grepc+       -- "grepc" -> putStrLn "grepc" >> grepc         _ -> putStrLn $ "Unknown command: " ++ cmd
examples/CoreUtilsHandle.hs view
@@ -6,11 +6,11 @@ import System.IO (IOMode(..), stdin, stdout, Handle, openFile)  import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream import qualified Streamly.FileSystem.Handle as Handle-import qualified Streamly.Prelude as Stream import qualified Streamly.Unicode.Stream as Unicode -import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream (splitOn)+import qualified Streamly.Internal.Data.Stream.Chunked as ArrayStream (splitOn)  -- | Read the contents of a file to stdout. --@@ -20,21 +20,21 @@ -- catBytes :: Handle -> IO () catBytes src =-      Stream.unfold Handle.read src     -- SerialT IO Word8-    & Stream.fold (Handle.write stdout) -- IO ()+      Stream.unfold Handle.reader src     -- Stream IO Word8+    & Stream.fold (Handle.write stdout)   -- IO ()  -- | Chunked version, more efficient than the byte stream version above. Reads -- the file in 256KB chunks and writes those chunks to stdout. cat :: Handle -> IO () cat src =-      Stream.unfold Handle.readChunksWithBufferOf (256*1024, src) -- SerialT IO (Array Word8)+      Stream.unfold Handle.chunkReaderWith (256*1024, src) -- Stream IO (Array Word8)     & Stream.fold (Handle.writeChunks stdout) -- IO ()  -- | Read from standard input write to standard output echo :: IO () echo =-      Stream.unfold Handle.readChunks stdin   -- SerialT IO (Array Word8)-    & Stream.fold (Handle.writeChunks stdout) -- IO ()+      Stream.unfold Handle.chunkReader stdin   -- Stream IO (Array Word8)+    & Stream.fold (Handle.writeChunks stdout)  -- IO ()  -- | Copy a source file to a destination file. --@@ -43,15 +43,15 @@ -- 32KB and writes those chunks to the destination file. cpBytes :: Handle -> Handle -> IO () cpBytes src dst =-      Stream.unfold Handle.read src  -- SerialT IO Word8-    & Stream.fold (Handle.write dst) -- IO ()+      Stream.unfold Handle.reader src  -- Stream IO Word8+    & Stream.fold (Handle.write dst)   -- IO ()  -- | Chunked version, more efficient than the byte stream version above. Reads -- the file in 256KB chunks and writes those chunks to stdout. cp :: Handle -> Handle -> IO () cp src dst =       Stream.fold (Handle.writeChunks dst)-    $ Stream.unfold Handle.readChunksWithBufferOf (256*1024, src)+    $ Stream.unfold Handle.chunkReaderWith (256*1024, src)  -- | Count lines like wc -l. --@@ -59,19 +59,23 @@ -- and counts the lines.. wclChar :: Handle -> IO Int wclChar src =-      Stream.unfold Handle.read src             -- SerialT IO Word8-    & Unicode.decodeLatin1                      -- SerialT IO Char-    & Stream.splitOnSuffix (== '\n') Fold.drain -- SerialT IO ()-    & Stream.length                             -- IO ()+      Stream.unfold Handle.reader src           -- Stream IO Word8+    & Unicode.decodeLatin1                      -- Stream IO Char+    & split (== '\n') Fold.drain                -- Stream IO ()+    & Stream.fold Fold.length +    where++    split p f = Stream.foldMany (Fold.takeEndBy_ p f)   -- IO ()+ -- | More efficient chunked version. Reads chunks from the input handles and -- splits the chunks directly instead of converting them into byte stream -- first. wcl :: Handle -> IO Int wcl src =-      Stream.unfold Handle.readChunks src -- SerialT IO (Array Word8)-    & ArrayStream.splitOn 10              -- SerialT IO (Array Word8)-    & Stream.length                       -- IO ()+      Stream.unfold Handle.chunkReader src -- Stream IO (Array Word8)+    & ArrayStream.splitOn 10               -- Stream IO (Array Word8)+    & Stream.fold Fold.length              -- IO ()  main :: IO () main = do
+ examples/DateTimeParser.hs view
@@ -0,0 +1,233 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE BangPatterns #-}++-- IMPORTANT: Do not export anything other than "main" from this file. The+-- benchmark timings may change drastically otherwise.++module Main+    ( main+    )+where++import Data.Either (fromRight)+import Data.Maybe (fromJust)+import Streamly.Data.Array (Array)+import Streamly.Internal.Data.Fold (Fold(..), Step(..))+import Test.Tasty.Bench++import qualified Data.Char as Char+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Parser as Parser+import qualified Streamly.Data.ParserK as ParserK+import qualified Streamly.Data.StreamK as StreamK+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.Internal.Data.Fold as Fold (foldt', satisfy)+import qualified Streamly.Unicode.Parser as Parser++-------------------------------------------------------------------------------+-- Monolithic fold - fastest, same as rust speeddate perf+-------------------------------------------------------------------------------++mkTime :: Int -> Int -> Int -> Int -> Int -> Int -> Int+mkTime year month day hr mn sec = year + month + day + hr + mn + sec++{-# INLINE isDigit #-}+isDigit :: Char -> Bool+isDigit c = c >= '0' && c <= '9'++data FoldState =+      Year !Int+    | Month !Int !Int+    | Day !Int !Int+    | Hr !Int !Int+    | Min !Int !Int !Int+    | Sec !Int !Int !Int++{-# INLINE foldDateTime #-}+foldDateTime :: Array Char -> IO (Int, Int)+foldDateTime arr =+    Stream.fold t $ Stream.unfold Array.reader arr++    where++    t = Fold.foldt' step initial extract++    initial = Partial $ Year 0++    dec n ch = n * 10 + fromIntegral (Char.ord ch) - 48++    step (Year n) ch+      | isDigit ch = Partial $ Year (dec n ch)+      | ch == '-' = Partial $ Month (365 * n) 0+      | otherwise = error "parse error"+    step (Month y n) ch+      | isDigit ch = Partial $ Month y (dec n ch)+      | ch == '-' = Partial $ Day (y + n * 30) 0+      | otherwise = error "parse error"+    step (Day x n) ch+      | isDigit ch = Partial $ Day x (dec n ch)+      | ch == 'T' = Partial $ Hr (x + n) 0+      | otherwise = error "parse error"+    step (Hr d n) ch+      | isDigit ch = Partial $ Hr d (dec n ch)+      | ch == ':' = Partial $ Min d (n * 60) 0+      | otherwise = error "parse error"+    step (Min d m n) ch+      | isDigit ch = Partial $ Min d m (dec n ch)+      | ch == ':' = Partial $ Sec d ((m + n) * 60) 0+      | otherwise = error "parse error"+    step (Sec d s n) ch+      | isDigit ch = Partial $ Sec d s (dec n ch)+      | ch == 'Z' = Done (d, s + n)+      | otherwise = error "parse error"++    extract _ = error "incomplete"++-------------------------------------------------------------------------------+-- Modular applicative version - 4x slower+-------------------------------------------------------------------------------++{-# INLINE check #-}+check ::  (Char -> Bool) -> Fold m Char a -> Fold m Char a+check p = Fold.lmap (\x -> if p x then x else error "parse failed")++{-# INLINE decimal #-}+decimal :: Monad m  => Int -> Fold m Char Int+decimal n = Fold.take n (check isDigit (Fold.foldl' step 0))++    where++    step a c = a * 10 + fromIntegral (Char.ord c - 48)++{-# INLINE char #-}+char :: Monad m => Char -> Fold m Char Char+char c = fromJust <$> Fold.satisfy (== c)++{-# NOINLINE _foldDateTimeAp #-}+_foldDateTimeAp :: Array Char -> IO Int+_foldDateTimeAp arr =+    let t =+                mkTime+            <$> decimal 4  -- year+            <*  char '-'+            <*> decimal 2  -- month+            <*  char '-'+            <*> decimal 2  -- day+            <*  char 'T'+            <*> decimal 2  -- hr+            <*  char ':'+            <*> decimal 2  -- min+            <*  char ':'+            <*> decimal 2  -- sec+            <*  char 'Z'+    in Stream.fold t $ Stream.unfold Array.reader arr++-------------------------------------------------------------------------------+-- Using foldBreak - slower than applicative+-------------------------------------------------------------------------------++{-# INLINE _foldBreakDateTime #-}+_foldBreakDateTime :: Array Char -> IO Int+_foldBreakDateTime arr = do+    let s = Stream.unfold Array.reader arr+    (year, s1) <- Stream.foldBreak (decimal 4) s+    (_, s2) <- Stream.foldBreak (char '-') s1+    (month, s3) <- Stream.foldBreak (decimal 2) s2+    (_, s4) <- Stream.foldBreak (char '-') s3+    (day, s5) <- Stream.foldBreak (decimal 2) s4+    (_, s6) <- Stream.foldBreak (char 'T') s5+    (hr, s7) <- Stream.foldBreak (decimal 2) s6+    (_, s8) <- Stream.foldBreak (char ':') s7+    (mn, s9) <- Stream.foldBreak (decimal 2) s8+    (_, s10) <- Stream.foldBreak (char ':') s9+    (sec, s11) <- Stream.foldBreak (decimal 2) s10+    (_, _) <- Stream.foldBreak (char 'Z') s11+    return (year + month + day + hr + mn + sec)++-------------------------------------------------------------------------------+-- Using parseBreak - slower than foldBreak and parseK+-------------------------------------------------------------------------------++{-# NOINLINE _parseBreakDateTime #-}+_parseBreakDateTime :: Array Char -> IO Int+_parseBreakDateTime arr = do+    let s = StreamK.fromStream $ Stream.fromPure arr+        p = ParserK.fromParser . Parser.fromFold+    (Right year, s1) <- StreamK.parseBreakChunks (p $ decimal 4) s+    (_, s2) <- StreamK.parseBreakChunks (p $ char '-') s1+    (Right month, s3) <- StreamK.parseBreakChunks (p $ decimal 2) s2+    (_, s4) <- StreamK.parseBreakChunks (p $ char '-') s3+    (Right day, s5) <- StreamK.parseBreakChunks (p $ decimal 2) s4+    (_, s6) <- StreamK.parseBreakChunks (p $ char 'T') s5+    (Right hr, s7) <- StreamK.parseBreakChunks (p $ decimal 2) s6+    (_, s8) <- StreamK.parseBreakChunks (p $ char ':') s7+    (Right mn, s9) <- StreamK.parseBreakChunks (p $ decimal 2) s8+    (_, s10) <- StreamK.parseBreakChunks (p $ char ':') s9+    (Right sec, s11) <- StreamK.parseBreakChunks (p $ decimal 2) s10+    (_, _) <- StreamK.parseBreakChunks (p $ char 'Z') s11+    return (year + month + day + hr + mn + sec)++-------------------------------------------------------------------------------+-- Parser -- slower than foldBreak+-------------------------------------------------------------------------------++{-# NOINLINE _parseKDateTime #-}+_parseKDateTime :: Array Char -> IO Int+_parseKDateTime arr = do+    r <- StreamK.parseChunks dateParser $ StreamK.fromPure arr+    return $ fromRight (error "failed") r++    where++    p = ParserK.fromParser++    dateParser = do+        year <- p $ Parser.decimal <* Parser.char '-'+        month <- p $ Parser.decimal <* Parser.char '-'+        day <- p $ Parser.decimal <* Parser.char 'T'+        hr <- p $ Parser.decimal <* Parser.char ':'+        mi <- p $ Parser.decimal <* Parser.char ':'+        sec <- p $ Parser.decimal <* Parser.char 'Z'+        pure (mkTime year month day hr mi sec)++-- Parser Monad should not be used for more than 2-3 compositions, use ParserK+-- instead. Something like this is doomed to not perform well.+{-# NOINLINE _parseDateTime #-}+_parseDateTime :: Array Char -> IO Int+_parseDateTime arr = do+    r <- Stream.parse dateParser $ Stream.unfold Array.reader arr+    return $ fromRight (error "failed") r++    where++    dateParser = do+        year <- Parser.decimal <* Parser.char '-'+        month <- Parser.decimal <* Parser.char '-'+        day <- Parser.decimal <* Parser.char 'T'+        hr <- Parser.decimal <* Parser.char ':'+        mi <- Parser.decimal <* Parser.char ':'+        sec <- Parser.decimal <* Parser.char 'Z'+        pure (mkTime year month day hr mi sec)++-------------------------------------------------------------------------------+-- Benchmarks+-------------------------------------------------------------------------------++timeBench :: Benchmark+timeBench =+    let !(arr :: Array Char) = Array.fromListN 20 "2000-01-01T00:02:03Z"+    in bgroup "parseDateTime"+        -- IMPORTANT: Enable only one benchmark at a time, the benchmark+        -- timings may change drastically otherwise.+        [ bench "fold monolithic" $ nfIO $ foldDateTime arr  -- 20 ns+        {- , bench "fold applicative" $ nfIO $ _foldDateTimeAp arr -- 110 ns+        , bench "foldBreak" $ nfIO $ _foldBreakDateTime arr  -- 275 ns+        , bench "parseK" $ nfIO $ _parseKDateTime arr -- 340 ns+        , bench "parseBreak" $ nfIO $ _parseBreakDateTime arr -- 700 ns+        , bench "parseD" $ nfIO $ _parseDateTime arr -- 950 ns+        -}+        ]++main :: IO ()+main = defaultMain [timeBench]
examples/EchoServer.hs view
@@ -5,23 +5,23 @@ import Network.Socket (Socket)  import qualified Network.Socket as Net+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Network.Inet.TCP as TCP import qualified Streamly.Network.Socket as Socket-import qualified Streamly.Prelude as Stream  main :: IO () main =-      Stream.unfold TCP.acceptOnPort 8091 -- ParallelT IO Socket-    & Stream.mapM (handleWithM echo)      -- ParallelT IO ()-    & Stream.fromParallel                 -- SerialT IO ()-    & Stream.drain                        -- IO ()+      Stream.unfold TCP.acceptorOnPort 8091 -- Stream IO Socket+    & Stream.parMapM id (handleWithM echo)  -- Stream IO ()+    & Stream.fold Fold.drain                -- IO ()      where      echo :: Socket -> IO ()     echo sk =-          Stream.unfold Socket.readChunksWithBufferOf (32768, sk) -- SerialT IO (Array Word8)-        & Stream.fold (Socket.writeChunks sk)                     -- IO ()+          Stream.unfold Socket.chunkReaderWith (32768, sk) -- Stream IO (Array Word8)+        & Stream.fold (Socket.writeChunks sk)              -- IO ()      handleWithM :: (Socket -> IO ()) -> Socket -> IO ()     handleWithM f sk = finally (f sk) (Net.close sk)
examples/FileSender.hs view
@@ -12,10 +12,11 @@  import qualified Control.Monad.Catch as Catch import qualified Network.Socket as Net+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.FileSystem.Handle as Handle import qualified Streamly.Network.Inet.TCP as TCP import qualified Streamly.Network.Socket as Socket-import qualified Streamly.Prelude as Stream  main :: IO () main = do@@ -29,8 +30,8 @@      fileToSocket :: Handle -> Socket -> IO ()     fileToSocket fh sk =-          Stream.unfold Handle.readChunks fh  -- SerialT IO (Array Word8)-        & Stream.fold (Socket.writeChunks sk) -- IO ()+          Stream.unfold Handle.chunkReader fh  -- Stream IO (Array Word8)+        & Stream.fold (Socket.writeChunks sk)  -- IO ()      sendFile :: String -> IO ()     sendFile file =@@ -39,7 +40,6 @@      sendAll :: [String] -> IO ()     sendAll files =-          Stream.fromList files -- ParallelT IO String-        & Stream.mapM sendFile  -- ParallelT IO ()-        & Stream.fromParallel   -- SerialT IO ()-        & Stream.drain          -- IO ()+          Stream.fromList files        -- Stream IO String+        & Stream.parMapM id sendFile   -- Stream IO ()+        & Stream.fold Fold.drain       -- IO ()
examples/FileSystemEvent.hs view
@@ -6,11 +6,12 @@ import Data.Function ((&)) import Data.Word (Word8) import System.Environment (getArgs)-import Streamly.Data.Array.Foreign (Array)+import Streamly.Data.Array (Array)  import qualified Data.List.NonEmpty as NonEmpty-import qualified Streamly.Data.Array.Foreign as Array-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream import qualified Streamly.Unicode.Stream as Unicode  #if darwin_HOST_OS@@ -39,4 +40,4 @@     args <- getArgs     paths <- mapM toUtf8 args     Event.watch (NonEmpty.fromList paths)-        & Stream.mapM_ (putStrLn . Event.showEvent)+        & Stream.fold (Fold.drainMapM (putStrLn . Event.showEvent))
examples/Interop/Conduit.hs view
@@ -1,19 +1,20 @@ {-# LANGUAGE FlexibleContexts #-} -import Streamly.Prelude (IsStream, MonadAsync, SerialT)-import qualified Streamly.Prelude as Stream+import Streamly.Data.Stream (Stream)++import qualified Streamly.Data.Stream as Stream import qualified Data.Conduit as Conduit import qualified Data.Conduit.List as Conduit  -- | conduit to streamly-fromConduit :: (IsStream t, MonadAsync m) => Conduit.ConduitT () a m () -> t m a+fromConduit :: Monad m => Conduit.ConduitT () a m () -> Stream m a fromConduit = Stream.unfoldrM Conduit.unconsM . Conduit.sealConduitT  -- | streamly to conduit-toConduit :: Monad m => SerialT m a -> Conduit.ConduitT i a m ()+toConduit :: Monad m => Stream m a -> Conduit.ConduitT i a m () toConduit = Conduit.unfoldM Stream.uncons  main :: IO () main = do     Stream.toList (fromConduit (Conduit.sourceList ([1..3]::[Int]))) >>= print-    Conduit.runConduit (toConduit (Stream.fromFoldable ([1..3]::[Int])) Conduit..| Conduit.consume) >>= print+    Conduit.runConduit (toConduit (Stream.fromList ([1..3]::[Int])) Conduit..| Conduit.consume) >>= print
examples/Interop/Pipes.hs view
@@ -1,26 +1,27 @@ {-# LANGUAGE FlexibleContexts #-} -import Streamly.Prelude (IsStream, MonadAsync, SerialT)-import qualified Streamly.Prelude as Stream+import Streamly.Data.Stream (Stream)++import qualified Streamly.Data.Stream as Stream import qualified Pipes as Pipe import qualified Pipes.Prelude as Pipe  -- | pipes to streamly-fromPipes :: (IsStream t, MonadAsync m) => Pipe.Producer a m () -> t m a+fromPipes :: Monad m => Pipe.Producer a m () -> Stream m a fromPipes = Stream.unfoldrM unconsP     where     -- Adapt Pipe.next to return a Maybe instead of Either     unconsP p = either (const Nothing) Just <$> Pipe.next p  -- | streamly to pipes-toPipes :: Monad m => SerialT m a -> Pipe.Producer a m ()+toPipes :: Monad m => Stream m a -> Pipe.Producer a m () toPipes = Pipe.unfoldr unconsEither     where     -- Adapt Stream.uncons to return an Either instead of Maybe-    unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))+    unconsEither :: Monad m => Stream m a -> m (Either () (a, Stream m a))     unconsEither s = maybe (Left ()) Right <$> Stream.uncons s  main :: IO () main = do     Stream.toList (fromPipes (Pipe.each ([1..3]::[Int]))) >>= print-    Pipe.toListM (toPipes (Stream.fromFoldable ([1..3]::[Int]))) >>= print+    Pipe.toListM (toPipes (Stream.fromList ([1..3]::[Int]))) >>= print
examples/Interop/Streaming.hs view
@@ -1,23 +1,24 @@ {-# LANGUAGE FlexibleContexts #-} -import Streamly.Prelude (IsStream, MonadAsync, SerialT)-import qualified Streamly.Prelude as Stream+import Streamly.Data.Stream (Stream)++import qualified Streamly.Data.Stream as Stream import qualified Streaming import qualified Streaming.Prelude as Streaming  -- | streaming to streamly-fromStreaming :: (IsStream t, MonadAsync m) => Streaming.Stream (Streaming.Of a) m () -> t m a+fromStreaming :: Monad m => Streaming.Stream (Streaming.Of a) m () -> Stream m a fromStreaming = Stream.unfoldrM Streaming.uncons -- -- | streamly to streaming-toStreaming :: Monad m => SerialT m a -> Streaming.Stream (Streaming.Of a) m ()+toStreaming :: Monad m => Stream m a -> Streaming.Stream (Streaming.Of a) m () toStreaming = Streaming.unfoldr unconsEither     where     -- Adapt Stream.uncons to return an Either instead of Maybe-    unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))+    unconsEither :: Monad m => Stream m a -> m (Either () (a, Stream m a))     unconsEither s = maybe (Left ()) Right <$> Stream.uncons s  main :: IO () main = do     Stream.toList (fromStreaming (Streaming.each ([1..3]::[Int]))) >>= print-    Streaming.toList (toStreaming (Stream.fromFoldable ([1..3]::[Int]))) >>= print+    Streaming.toList (toStreaming (Stream.fromList ([1..3]::[Int]))) >>= print
examples/Interop/Vector.hs view
@@ -1,11 +1,12 @@ {-# LANGUAGE FlexibleContexts #-} -import Streamly.Prelude (MonadAsync, SerialT)-import qualified Streamly.Prelude as Stream+import Streamly.Data.Stream (Stream)++import qualified Streamly.Data.Stream as Stream import qualified Data.Vector.Fusion.Stream.Monadic as Vector  --  | vector to streamly-fromVector :: (Stream.IsStream t, MonadAsync  m) => Vector.Stream m a -> t m a+fromVector :: Monad m => Vector.Stream m a -> Stream m a fromVector = Stream.unfoldrM unconsV     where     unconsV v = do@@ -17,10 +18,10 @@             return $ Just (h, Vector.tail v)  --  | streamly to vector-toVector :: Monad m => SerialT m a -> Vector.Stream m a-toVector = Vector.unfoldrM (Stream.uncons . Stream.adapt)+toVector :: Monad m => Stream m a -> Vector.Stream m a+toVector = Vector.unfoldrM Stream.uncons  main :: IO () main = do     Stream.toList (fromVector (Vector.fromList ([1..3]::[Int])))   >>= print-    Vector.toList (toVector (Stream.fromFoldable ([1..3]::[Int]))) >>= print+    Vector.toList (toVector (Stream.fromList ([1..3]::[Int]))) >>= print
examples/Intro.hs view
@@ -8,22 +8,21 @@ import Data.Word (Word8) import System.Environment (getArgs) import System.IO (stdout)--import Streamly.Prelude (SerialT)-import Streamly.Data.Fold (Fold)-import Streamly.Data.Fold.Tee (Tee(..))+import Streamly.Data.Fold (Fold, Tee(..))+import Streamly.Data.Stream.Prelude (Stream) import Streamly.Data.Unfold (Unfold)+import Streamly.Internal.Data.Stream (CrossStream, mkCross, unCross) -import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Data.Unfold as Unfold-import qualified Streamly.Prelude as Stream import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.FileSystem.File as File import qualified Streamly.Unicode.Stream as Unicode +import qualified Streamly.Internal.Data.Stream as Stream (wordsBy) import qualified Streamly.Internal.Data.Unfold as Unfold (enumerateFromToIntegral)-import qualified Streamly.Internal.Data.Fold as Fold (classify)-import qualified Streamly.Internal.FileSystem.File as File (toBytes)  ------------------------------------------------------------------------------- -- Simple loops@@ -32,14 +31,14 @@ -- | Sum a list of Int sumInt :: Identity Int sumInt =-      Stream.unfold Unfold.fromList [1..10] -- SerialT Identity Int+      Stream.unfold Unfold.fromList [1..10] -- Stream Identity Int     & Stream.fold Fold.sum                  -- Identity Int  -- | Sum a list of Int sumInt1 :: Identity Int sumInt1 =-      Stream.fromList [1..10]       -- SerialT Identity Int-    & Stream.sum                    -- Identity Int+      Stream.fromList [1..10]       -- Stream Identity Int+    & Stream.fold Fold.sum          -- Identity Int  ------------------------------------------------------------------------------- -- Nested loops@@ -60,28 +59,31 @@                   (Unfold.lmap fst Unfold.enumerateFromToIntegral)                   (Unfold.lmap snd Unfold.enumerateFromToIntegral) -     in Stream.unfold xmult (range1,range2) -- SerialT Identity Int+     in Stream.unfold xmult (range1,range2) -- Stream Identity Int             & Stream.fold Fold.sum          -- Identity Int  -- | Nested looping similar to 'cross' above but more general and less -- efficient. The second stream may depend on the first stream. The loops -- cannot fuse completely. ---nestedLoops :: SerialT IO ()+nestedLoops :: CrossStream IO () nestedLoops = do-    x <- Stream.fromList [3,4 :: Int]-    y <- Stream.fromList [1..x]-    Stream.fromEffect $ print (x, y)+    x <- mkCross $  Stream.fromList [3,4 :: Int]+    y <- mkCross $  Stream.fromList [1..x]+    mkCross $  Stream.fromEffect $ print (x, y)  ------------------------------------------------------------------------------- -- Text processing ------------------------------------------------------------------------------- +splitOn :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b+splitOn p f = Stream.foldMany (Fold.takeEndBy_ p f)+ -- | Find average line length for lines in a text file avgLineLength :: IO Double avgLineLength =-      File.toBytes "input.txt"                    -- SerialT IO Word8-    & Stream.splitOnSuffix isNewLine Fold.length  -- SerialT IO Int+      File.read "input.txt"                    -- Stream IO Word8+    & splitOn isNewLine Fold.length               -- Stream IO Int     & Stream.fold avg                             -- IO Double      where@@ -93,17 +95,21 @@     toDouble = fmap (fromIntegral :: Int -> Double)      avg :: Fold IO Int Double-    avg = toFold $ (/)+    avg = unTee $ (/)             <$> Tee (toDouble Fold.sum)             <*> Tee (toDouble Fold.length) +{-# INLINE kvMap #-}+kvMap :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)+kvMap = Fold.toMap fst . Fold.lmap snd+ -- | Read text from a file and generate a histogram of line length lineLengthHistogram :: IO (Map Int Int) lineLengthHistogram =-      File.toBytes "input.txt"                   -- SerialT IO Word8-    & Stream.splitOnSuffix isNewLine Fold.length -- SerialT IO Int-    & Stream.map bucket                          -- SerialT IO (Int, Int)-    & Stream.fold (Fold.classify Fold.length)    -- IO (Map Int Int)+      File.read "input.txt"                      -- Stream IO Word8+    & splitOn isNewLine Fold.length              -- Stream IO Int+    & fmap bucket                                -- Stream IO (Int, Int)+    & Stream.fold (kvMap Fold.length)            -- IO (Map Int Int)      where @@ -116,11 +122,11 @@ -- | Read text from a file and generate a histogram of word length wordLengthHistogram :: IO (Map Int Int) wordLengthHistogram =-      File.toBytes "input.txt"                -- SerialT IO Word8-    & Unicode.decodeLatin1                    -- SerialT IO Char-    & Stream.wordsBy isSpace Fold.length      -- SerialT IO Int-    & Stream.map bucket                       -- SerialT IO (Int, Int)-    & Stream.fold (Fold.classify Fold.length) -- IO (Map (Int, Int))+      File.read "input.txt"                   -- Stream IO Word8+    & Unicode.decodeLatin1                    -- Stream IO Char+    & Stream.wordsBy isSpace Fold.length      -- Stream IO Int+    & fmap bucket                             -- Stream IO (Int, Int)+    & Stream.fold (kvMap Fold.length)         -- IO (Map (Int, Int))      where @@ -146,12 +152,13 @@ -- getWords :: IO () getWords =-      Stream.fromListM meanings                -- AheadT  IO (String, String)-    & Stream.fromAhead                         -- SerialT IO (String, String)-    & Stream.map show                          -- SerialT IO String-    & unlinesBy "\n"                           -- SerialT IO String-    & Stream.map Array.fromList                -- SerialT IO (Array Word8)-    & Stream.fold (Handle.writeChunks stdout)  -- IO ()+      Stream.fromList meanings                -- Stream IO (IO (String, String))+    & Stream.parSequence+        (Stream.ordered True)                 -- Stream IO (String, String)+    & fmap show                               -- Stream IO String+    & unlinesBy "\n"                          -- Stream IO String+    & fmap Array.fromList                     -- Stream IO (Array Word8)+    & Stream.fold (Handle.writeChunks stdout) -- IO ()      where unlinesBy = Stream.intercalateSuffix (Unfold.function id) @@ -180,7 +187,7 @@             print $ runIdentity $ crossProduct (1,1000) (1000,2000)         "nestedLoops" -> do             putStrLn "nestedLoops"-            Stream.drain nestedLoops+            Stream.fold Fold.drain $ unCross nestedLoops         "avgLineLength" -> do             putStrLn "avgLineLength"             avgLineLength >>= print
examples/ListDir.hs view
@@ -1,27 +1,36 @@+{-# LANGUAGE FlexibleContexts #-}+{-# OPTIONS_GHC -Wno-deprecations #-}+ module Main (main) where -import Data.Bifunctor (bimap)-import Data.Function ((&))-import Streamly.Prelude (SerialT) import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering)) -import qualified Streamly.Prelude as Stream-import qualified Streamly.Internal.Data.Stream.IsStream as Stream-       (iterateMapLeftsWith)-import qualified Streamly.Internal.FileSystem.Dir as Dir (toEither)---- Lists a dir as a stream of (Either Dir File)-listDir :: String -> SerialT IO (Either String String)-listDir dir =-      Dir.toEither dir               -- SerialT IO (Either String String)-    & Stream.map (bimap mkAbs mkAbs) -- SerialT IO (Either String String)--    where mkAbs x = dir ++ "/" ++ x+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream+-- import qualified Streamly.Internal.Data.Stream as Stream+-- import qualified Streamly.Internal.Data.Unfold as Unfold (either, nil)+import qualified Streamly.Internal.FileSystem.Dir as Dir  -- | List the current directory recursively using concurrent processing main :: IO () main = do     hSetBuffering stdout LineBuffering-    let start = Stream.fromPure (Left ".")-    Stream.iterateMapLeftsWith Stream.ahead listDir start-        & Stream.mapM_ print+    Stream.fold (Fold.drainMapM print)+        -- $ Stream.unfoldIterateDfs unfoldOne+        -- $ Stream.unfoldIterateBfs unfoldOne+        -- $ Stream.unfoldIterateBfsRev unfoldOne+        -- $ Stream.concatIterateDfs streamOneMaybe+        -- $ Stream.concatIterateBfs streamOneMaybe+        -- $ Stream.concatIterateBfsRev streamOneMaybe+        -- $ Stream.concatIterateWith Stream.append streamOne+        -- $ Stream.mergeIterateWith Stream.interleave streamOne+        $ Stream.parConcatIterate id streamOne+        -- $ Stream.parConcatIterate (Stream.interleaved True) streamOne+        -- $ Stream.parConcatIterate (Stream.ordered True) streamOne+        $ Stream.fromPure (Left ".")++    where++    -- unfoldOne = Unfold.either Dir.eitherReaderPaths Unfold.nil+    -- streamOneMaybe = either (Just . Dir.readEitherPaths) (const Nothing)+    streamOne = either Dir.readEitherPaths (const Stream.nil)
+ examples/LogParser.hs view
@@ -0,0 +1,58 @@+{-# LANGUAGE QuasiQuotes #-}+{-# OPTIONS_GHC -fspec-constr-recursive=4 #-}++import Data.Function ((&))+import Data.Functor.Identity (runIdentity)+import Streamly.Data.Parser (Parser)+import Streamly.Unicode.String (str)++import qualified Data.Char as Char+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Parser as Parser+import qualified Streamly.Data.Stream as Stream++-- Example of a quoted log string to be parsed+quoted :: String+quoted = [str|+"[2023-02-04T09:15:20.549Z] \"GET /cards?customer_id=XXXXP/1.1\" 200+- \"-\" \"-\" 0 75 23 23 \"131.26.22.133,127.16.101.49,127.18.4.69\"+\"edge\" \"7ccc821a-03ff-49c1-a721-977e7cbd78f3\" \"api.example.in\"+\"127.18.55.25:1108\" outbound|443|v21945461|svc.cluster.local+127.18.44.67:75523 127.18.44.67:8443 127.18.4.69:14462+mum.example.net confirmtkt\n"+|]++-- Use double quote as the quoting char+isQuote :: Char -> Maybe Char+isQuote x =+    case x of+        '"' -> Just x+        _ -> Nothing++-- Transliterate \" and \n+tr :: Char -> Char -> Maybe Char+tr q x =+    case x of+        _ | x == q -> Just x+        'n' -> Just '\n'+        _ -> Nothing++-- quoted string parser, reads everything inside quotes as a single word+-- and processes the quotes and escapes to expose the words inside it.+parser :: Monad m => Parser Char m [Char]+parser = Parser.wordWithQuotes False tr '\\' isQuote Char.isSpace Fold.toList++main :: IO ()+main = do+    -- Strip outer quotes+    let res = runIdentity $ Stream.parse parser $ Stream.fromList quoted++    -- parse words inside quotes+    case res of+        Left err -> error $ "Malformed quoted string: " ++ show err+        Right unquoted ->+              Stream.fromList unquoted+            & Stream.parseMany parser+            & Stream.catRights+            & Stream.toList+            >>= print
examples/MergeServer.hs view
@@ -1,39 +1,44 @@ import Control.Monad.IO.Class (liftIO) import Data.Function ((&)) import Network.Socket (Socket, close)-import Streamly.Data.Array.Foreign (Array)-import Streamly.Prelude (SerialT)+import Streamly.Data.Array (Array)+import Streamly.Data.Stream.Prelude (Stream) import System.IO (Handle, withFile, IOMode(..)) -import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.FileSystem.Handle as Handle import qualified Streamly.Network.Socket as Socket import qualified Streamly.Network.Inet.TCP as TCP-import qualified Streamly.Prelude as Stream import qualified Streamly.Unicode.Stream as Unicode  -- | Read a line stream from a socket. Note, lines are buffered, we could add -- a limit to the buffering for safety.-readLines :: Socket -> SerialT IO (Array Char)+readLines :: Socket -> Stream IO (Array Char) readLines sk =-    Stream.unfold Socket.read sk                 -- SerialT IO Word8-  & Unicode.decodeLatin1                         -- SerialT IO Char-  & Stream.splitWithSuffix (== '\n') Array.write -- SerialT IO String+    Stream.unfold Socket.reader sk               -- Stream IO Word8+  & Unicode.decodeLatin1                         -- Stream IO Char+  & split (== '\n') Array.write                  -- Stream IO String -recv :: Socket -> SerialT IO (Array Char)-recv sk = Stream.finally (liftIO $ close sk) (readLines sk)+  where +  split p f = Stream.foldMany (Fold.takeEndBy p f)++recv :: Socket -> Stream IO (Array Char)+recv sk = Stream.finallyIO (liftIO $ close sk) (readLines sk)+ -- | Starts a server at port 8091 listening for lines with space separated -- words. Multiple clients can connect to the server and send streams of lines. -- The server handles all the connections concurrently, merges the incoming -- streams at line boundaries and writes the merged stream to a file. server :: Handle -> IO () server file =-      Stream.unfold TCP.acceptOnPort 8090        -- SerialT IO Socket-    & Stream.concatMapWith Stream.parallel recv  -- SerialT IO (Array Char)-    & Stream.unfoldMany Array.read               -- SerialT IO Char-    & Unicode.encodeLatin1                       -- SerialT IO Word8-    & Stream.fold (Handle.write file)            -- IO ()+      Stream.unfold TCP.acceptorOnPort 8090         -- Stream IO Socket+    & Stream.parConcatMap (Stream.eager True) recv  -- Stream IO (Array Char)+    & Stream.unfoldMany Array.reader                -- Stream IO Char+    & Unicode.encodeLatin1                          -- Stream IO Word8+    & Stream.fold (Handle.write file)               -- IO ()  main :: IO () main = withFile "output.txt" AppendMode server
examples/MergeSort.hs view
@@ -1,26 +1,106 @@-{-# LANGUAGE FlexibleContexts    #-}+module Main+    ( main+    , sortMergeCombined+    , sortMergeChunks+    )+where --- This example generates two random streams sorted in ascending order and--- merges them in ascending order, concurrently.------ Compile with '-threaded -with-rtsopts "-N"' GHC options to use the--- parallelism.+import Control.Monad (void)+import Data.Function ((&))+import Streamly.Data.Array (Array)+import Streamly.Data.Stream (Stream) -import Data.Word (Word16)-import Streamly.Prelude (SerialT)-import System.Random (getStdGen, randoms)+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.Data.StreamK as K -import qualified Data.List as List-import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Stream as Stream (reduceIterateBfs) -getRandomSorted :: IO (SerialT IO Word16)-getRandomSorted = do-    g <- getStdGen-    let ls = take 100000 (randoms g) :: [Word16]-    return $ Stream.fromList (List.sort ls)+input :: [Int]+input = [1000000,999999..1] +chunkSize :: Int+chunkSize = 32*1024++streamChunk :: Array Int -> Stream IO Int+streamChunk =+    K.toStream+        . K.sortBy compare+        . K.fromStream+        . Stream.unfold Array.reader++sortChunk :: Array Int -> IO (Array Int)+sortChunk = Stream.fold Array.write . streamChunk++-------------------------------------------------------------------------------+-- Stream the unsorted chunks and sort, merge those streams.+-------------------------------------------------------------------------------+-- In contrast to sortMergeSeparate this uses much more peak memory because all+-- the streams are open in memory at the same time.+sortMergeCombined :: (Array Int -> Stream IO Int) -> IO ()+sortMergeCombined f =+    Stream.fromList input+        & Stream.chunksOf chunkSize+        & K.fromStream+        & K.mergeMapWith (K.mergeBy compare) (K.fromStream . f)+        & K.toStream+        & Stream.fold Fold.drain++-------------------------------------------------------------------------------+-- First create a stream of sorted chunks, then stream sorted chunks and merge+-- the streams+-------------------------------------------------------------------------------++sortMergeSeparate ::+       (   (Array Int -> IO (Array Int))+        -> Stream IO (Array Int)+        -> Stream IO (Array Int)+       )+    -> IO ()+sortMergeSeparate f =+    Stream.fromList input+        & Stream.chunksOf chunkSize+        & f sortChunk+        & K.fromStream+        & K.mergeMapWith+            (K.mergeBy compare) (K.fromStream . Stream.unfold Array.reader)+        & K.toStream+        & Stream.fold Fold.drain++-------------------------------------------------------------------------------+-- First create a stream of sorted chunks, merge sorted chunks into sorted+-- chunks recursively.+-------------------------------------------------------------------------------++reduce :: Array Int -> Array Int -> IO (Array Int)+reduce arr1 arr2 =+    Stream.mergeBy+        compare+        (Stream.unfold Array.reader arr1)+        (Stream.unfold Array.reader arr2)+        & Stream.fold Array.write++sortMergeChunks ::+       (  (Array Int -> IO (Array Int))+       -> Stream IO (Array Int)+       -> Stream IO (Array Int)+       )+    -> IO ()+sortMergeChunks f =+    Stream.fromList input+        & Stream.chunksOf chunkSize+        & f sortChunk+        & Stream.reduceIterateBfs reduce+        & void++-- | Divide a stream in chunks, sort the chunks and merge them. main :: IO () main = do-    s1 <- getRandomSorted-    s2 <- getRandomSorted-    Stream.last (Stream.mergeAsyncBy compare s1 s2) >>= print+    -- Sorted in best performing first order+    sortMergeSeparate (Stream.parMapM id)+    -- sortMergeSeparate Stream.mapM+    -- sortMergeCombined (Stream.parEval id . streamChunk)+    -- sortMergeCombined streamChunk+    -- sortMergeChunks (Stream.parMapM id)+    -- sortMergeChunks Stream.mapM
examples/Rate.hs view
@@ -1,11 +1,11 @@ import Data.Function ((&))-import qualified Streamly.Prelude as Stream-import qualified Streamly.Internal.Data.Stream.IsStream as Stream (timestamped)+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.Internal.Data.Stream as Stream (timestamped)  main :: IO () main =-      Stream.repeatM (pure "tick")  -- AsyncT IO String-    & Stream.timestamped            -- AsyncT IO (AbsTime, String)-    & Stream.avgRate 1              -- AsyncT IO (AbsTime, String)-    & Stream.fromAsync              -- SerialT IO (AbsTime, String)-    & Stream.mapM_ print            -- IO ()+      Stream.sequence (Stream.repeat (pure "tick"))+    & Stream.timestamped+    & Stream.parEval (Stream.avgRate 1)+    & Stream.fold (Fold.drainMapM print)
examples/Split.hs view
@@ -4,13 +4,14 @@ import Data.Word (Word8) import System.Environment (getArgs) import System.IO (Handle, IOMode(..), openFile, hClose)-import Streamly.Prelude (SerialT)+import Streamly.Data.Stream (Stream) -import qualified Streamly.Prelude as Stream-import qualified Streamly.Internal.Data.Stream.IsStream as Stream (refoldMany)+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.Internal.Data.Stream as Stream (refoldMany) import qualified Streamly.Internal.Data.Refold.Type as Refold (take) import qualified Streamly.FileSystem.Handle as Handle-import qualified Streamly.Internal.FileSystem.Handle as Handle (consumer)+import qualified Streamly.Internal.FileSystem.Handle as Handle (writer)  newHandle :: StateT (Maybe (Handle, Int)) IO Handle newHandle = do@@ -28,13 +29,13 @@ -- of directory names. splitFile :: Handle -> IO () splitFile inHandle =-      (Stream.unfold Handle.read inHandle :: SerialT IO Word8) -- SerialT IO Word8-    & Stream.liftInner -- SerialT (StateT (Maybe (Handle, Int)) IO) Word8-    -- SerialT (StateT (Maybe (Handle, Int)) IO) ()-    & Stream.refoldMany (Refold.take (180 * mb) Handle.consumer) newHandle-    & Stream.runStateT (return Nothing)  -- SerialT IO (Maybe (Handle, Int), ())-    & Stream.map snd                     -- SerialT IO ()-    & Stream.drain                       -- SerialT IO ()+      (Stream.unfold Handle.reader inHandle :: Stream IO Word8) -- Stream IO Word8+    & Stream.liftInner                   -- Stream (StateT (Maybe (Handle, Int)) IO) Word8+    -- Stream (StateT (Maybe (Handle, Int)) IO) ()+    & Stream.refoldMany (Refold.take (180 * mb) Handle.writer) newHandle+    & Stream.runStateT (return Nothing)  -- Stream IO (Maybe (Handle, Int), ())+    & fmap snd                           -- Stream IO ()+    & Stream.fold Fold.drain             -- Stream IO ()      where
examples/WordCount.hs view
@@ -8,8 +8,9 @@ import Data.Function ((&)) import System.Environment (getArgs) -import qualified Streamly.Internal.FileSystem.File as File-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.FileSystem.File as File import qualified Streamly.Unicode.Stream as Stream  -------------------------------------------------------------------------------@@ -40,10 +41,10 @@  wc :: String -> IO Counts wc file =-      Stream.unfold File.read file            -- SerialT IO Word8-    & Stream.decodeLatin1                     -- SerialT IO Char- -- & Stream.decodeUtf8                       -- SerialT IO Char-    & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts+      File.read file                                      -- Stream IO Word8+    & Stream.decodeLatin1                                 -- Stream IO Char+ -- & Stream.decodeUtf8                                   -- Stream IO Char+    & Stream.fold (Fold.foldl' count (Counts 0 0 0 True)) -- IO Counts  ------------------------------------------------------------------------------- -- Main
examples/WordCountModular.hs view
@@ -7,14 +7,12 @@ import Data.Char (chr, ord) import Data.Function ((&)) import Data.Word (Word8)-import Streamly.Data.Fold (Fold)-import Streamly.Data.Fold.Tee (Tee(..))+import Streamly.Data.Fold (Fold, Tee(..)) import System.Environment (getArgs)  import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Data.Fold.Tee as Tee-import qualified Streamly.Internal.FileSystem.File as File (toBytes)-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.FileSystem.File as File  {-# INLINE isSpace #-} isSpace :: Char -> Bool@@ -28,7 +26,7 @@ -- The fold accepts a stream of `Word8` and returns a value of type "a". foldWith :: Fold IO Word8 a -> String -> IO a foldWith f file =-    File.toBytes file -- SerialT IO Word8+    File.read file    -- Stream IO Word8   & Stream.fold f     -- IO a  -------------------------------------------------------------------------------@@ -62,7 +60,7 @@  -- The fold accepts a stream of `Word8` and returns the three counts countAll :: Fold IO Word8 (Int, Int, Int)-countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords+countAll = unTee $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords  ------------------------------------------------------------------------------- -- Main
examples/WordCountParallel.hs view
@@ -7,20 +7,21 @@ import Data.Word (Word8) import GHC.Conc (numCapabilities) import System.Environment (getArgs)-import Streamly.Data.Array.Foreign (Array)+import Streamly.Data.Array (Array) import WordCount (count, Counts(..), isSpace) -import qualified Streamly.Data.Array.Foreign as Array-import qualified Streamly.Internal.FileSystem.File as File (readChunks)-import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.FileSystem.File as File (readChunks) import qualified Streamly.Unicode.Stream as Stream  -- Get the line, word, char counts in one chunk. countArray :: Array Word8 -> IO Counts countArray arr =-      Stream.unfold Array.read arr            -- SerialT IO Word8-    & Stream.decodeLatin1                     -- SerialT IO Char-    & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts+      Stream.unfold Array.reader arr          -- Stream IO Word8+    & Stream.decodeLatin1                     -- Stream IO Char+    & Stream.fold (Fold.foldl' count (Counts 0 0 0 True)) -- IO Counts  -- When combining the counts in two contiguous chunks, we would also need to -- know whether the first element of the next chunk was a space char or@@ -29,7 +30,7 @@ {-# NOINLINE partialCounts #-} partialCounts :: Array Word8 -> IO (Bool, Counts) partialCounts arr = do-    let r = Array.getIndex arr 0+    let r = Array.getIndex 0 arr     case r of         Just x -> do             counts <- countArray arr@@ -49,11 +50,17 @@ -- apply our counting function to each array and then combine all the counts. wc :: String -> IO (Bool, Counts) wc file = do-      Stream.unfold File.readChunks file -- AheadT IO (Array Word8)-    & Stream.mapM partialCounts          -- AheadT IO (Bool, Counts)-    & Stream.maxThreads numCapabilities  -- AheadT IO (Bool, Counts)-    & Stream.fromAhead                   -- SerialT IO (Bool, Counts)-    & Stream.foldl' addCounts (False, Counts 0 0 0 True) -- IO (Bool, Counts)+      File.readChunks file               -- Stream IO (Array Word8)+    & Stream.parMapM+        ( Stream.maxThreads numCapabilities+        . Stream.ordered True+        )+        partialCounts                   -- Stream IO (Bool, Counts)+    & Stream.fold foldCounts            -- IO (Bool, Counts)++    where++    foldCounts = Fold.foldl' addCounts (False, Counts 0 0 0 True)  ------------------------------------------------------------------------------- -- Main
examples/WordCountParallelUTF8.hs view
@@ -34,19 +34,23 @@ import Data.Char (isSpace) import Data.Word (Word8) import GHC.Conc (numCapabilities)+import Streamly.Data.MutArray (MutArray)+import Streamly.Data.Stream.Prelude (Stream) import System.Environment (getArgs) import System.IO (Handle, openFile, IOMode(..)) -import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.MutArray as MArray+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.FileSystem.Handle as Handle-import qualified Streamly.Prelude as Stream  -- Internal modules import qualified Streamly.Internal.Unicode.Stream as Unicode        (DecodeState, DecodeError(..), CodePoint, decodeUtf8Either        , resumeDecodeUtf8Either)-import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MArray-       (getIndexUnsafe, putIndexUnsafe, modifyIndexUnsafe, Array, newArray)+import qualified Streamly.Internal.Data.Array.Mut.Type as MArray+       (getIndexUnsafe, putIndexUnsafe, modifyIndexUnsafe)   -------------------------------------------------------------------------------@@ -178,20 +182,20 @@ -- Default/initial state of the block ------------------------------------------------------------------------------- -readField :: MArray.Array Int -> Field -> IO Int+readField :: MutArray Int -> Field -> IO Int readField v fld = MArray.getIndexUnsafe (fromEnum fld) v -writeField :: MArray.Array Int -> Field -> Int -> IO ()-writeField v fld i = void $ MArray.putIndexUnsafe (fromEnum fld) i v+writeField :: MutArray Int -> Field -> Int -> IO ()+writeField v fld i = void $ MArray.putIndexUnsafe i v (fromEnum fld) -modifyField :: MArray.Array Int -> Field -> (Int -> Int) -> IO ()+modifyField :: MutArray Int -> Field -> (Int -> Int) -> IO () modifyField v fld f = do   let index = fromEnum fld-  MArray.modifyIndexUnsafe index (\x -> (f x, ())) v+  MArray.modifyIndexUnsafe index v (\x -> (f x, ())) -newCounts :: IO (MArray.Array Int)+newCounts :: IO (MutArray Int) newCounts = do-    counts <- MArray.newArray (fromEnum (maxBound :: Field) + 1)+    counts <- MArray.newPinned (fromEnum (maxBound :: Field) + 1)     writeField counts LineCount 0     writeField counts WordCount 0     writeField counts CharCount 0@@ -206,7 +210,7 @@ -- Counting chars ------------------------------------------------------------------------------- -accountChar :: MArray.Array Int -> Bool -> IO ()+accountChar :: MutArray Int -> Bool -> IO () accountChar counts isSp = do     c <- readField counts CharCount     let space = if isSp then 1 else 0@@ -218,7 +222,7 @@ -- Manipulating the header bytes ------------------------------------------------------------------------------- -addToHeader :: MArray.Array Int -> Int -> IO Bool+addToHeader :: MutArray Int -> Int -> IO Bool addToHeader counts cp = do     cnt <- readField counts HeaderWordCount     case cnt of@@ -237,7 +241,7 @@             return True         _ -> return False -resetHeaderOnNewChar :: MArray.Array Int -> IO ()+resetHeaderOnNewChar :: MutArray Int -> IO () resetHeaderOnNewChar counts = do     hdone <- readField counts HeaderDone     when (hdone == 0) $ writeField counts HeaderDone 1@@ -246,13 +250,13 @@ -- Manipulating the trailer ------------------------------------------------------------------------------- -setTrailer :: MArray.Array Int -> Unicode.DecodeState -> Unicode.CodePoint -> IO ()+setTrailer :: MutArray Int -> Unicode.DecodeState -> Unicode.CodePoint -> IO () setTrailer counts st cp = do     writeField counts TrailerState (fromIntegral st)     writeField counts TrailerCodePoint cp     writeField counts TrailerPresent 1 -resetTrailerOnNewChar :: MArray.Array Int -> IO ()+resetTrailerOnNewChar :: MutArray Int -> IO () resetTrailerOnNewChar counts = do     trailer <- readField counts TrailerPresent     when (trailer /= 0) $ do@@ -264,7 +268,7 @@ -------------------------------------------------------------------------------  {-# INLINE countChar #-}-countChar :: MArray.Array Int -> Either Unicode.DecodeError Char -> IO ()+countChar :: MutArray Int -> Either Unicode.DecodeError Char -> IO () countChar counts inp =     case inp of         Right ch -> do@@ -301,7 +305,7 @@                     then accountChar counts True                     else setTrailer counts st cp -printCounts :: MArray.Array Int -> IO ()+printCounts :: MutArray Int -> IO () printCounts v = do     l <- readField v LineCount     w <- readField v WordCount@@ -318,9 +322,9 @@ -- combine trailing bytes in preceding block with leading bytes in the next -- block and decode them into a codepoint reconstructChar :: Int-                -> MArray.Array Int-                -> MArray.Array Int-                -> IO (Stream.SerialT IO (Either Unicode.DecodeError Char))+                -> MutArray Int+                -> MutArray Int+                -> IO (Stream IO (Either Unicode.DecodeError Char)) reconstructChar hdrCnt v1 v2 = do     when (hdrCnt > 3 || hdrCnt < 0) $ error "reconstructChar: hdrCnt > 3"     stream1 <-@@ -346,7 +350,7 @@     cp <- readField v1 TrailerCodePoint     return $ Unicode.resumeDecodeUtf8Either (fromIntegral state) cp stream3 -getHdrChar :: MArray.Array Int -> IO (Maybe Int)+getHdrChar :: MutArray Int -> IO (Maybe Int) getHdrChar v = do     hdrCnt <- readField v HeaderWordCount     case hdrCnt of@@ -372,7 +376,7 @@  -- If the header of the first block is not done then combine the header -- with the header of the next block.-combineHeaders :: MArray.Array Int -> MArray.Array Int -> IO ()+combineHeaders :: MutArray Int -> MutArray Int -> IO () combineHeaders v1 v2 = do     hdone1 <- readField v1 HeaderDone     when (hdone1 == 0) $ do@@ -387,7 +391,7 @@ -- the first vector and returning it. -- XXX This is a quick hack and can be refactored to reduce the size -- and understandability considerably.-addCounts :: MArray.Array Int -> MArray.Array Int -> IO (MArray.Array Int)+addCounts :: MutArray Int -> MutArray Int -> IO (MutArray Int) addCounts v1 v2 = do     hdone1 <- readField v1 HeaderDone     hdone2 <- readField v2 HeaderDone@@ -477,7 +481,7 @@                     case res of                         Nothing -> error "addCounts: Bug. empty reconstructed char"                         Just (h, t) -> do-                            tlength <- Stream.length t+                            tlength <- Stream.fold Fold.length t                             case h of                                 Right ch -> do                                     -- If we have an error case after this@@ -516,7 +520,7 @@                                     -- in partially decoded char to be written                                     -- as trailer. Check if the last error is                                     -- an incomplete decode.-                                    r <- Stream.last t+                                    r <- Stream.fold Fold.latest t                                     let (st', cp') =                                             case r of                                                 Nothing -> (st, cp)@@ -557,34 +561,36 @@ -- Individual array processing is an isolated loop, fusing it with the bigger -- loop may be counter productive. {-# NOINLINE countArray #-}-countArray :: Array.Array Word8 -> IO (MArray.Array Int)+countArray :: Array.Array Word8 -> IO (MutArray Int) countArray src = do     counts <- newCounts-    Stream.mapM_ (countChar counts)+    Stream.fold (Fold.drainMapM (countChar counts))         $ Unicode.decodeUtf8Either-        $ Stream.unfold Array.read src+        $ Stream.unfold Array.reader src     return counts  {-# INLINE wcMwlParallel #-}-wcMwlParallel :: Handle -> Int -> IO (MArray.Array Int)+wcMwlParallel :: Handle -> Int -> IO (MutArray Int) wcMwlParallel src n = do-    Stream.foldlM' addCounts newCounts-        $ Stream.fromAhead-        $ Stream.maxThreads numCapabilities-        $ Stream.mapM countArray-        $ Stream.unfold Handle.readChunksWithBufferOf (n, src)+    Stream.fold (Fold.foldlM' addCounts newCounts)+        $ Stream.parMapM+            ( Stream.maxThreads numCapabilities+            . Stream.ordered True+            )+            countArray+        $ Stream.unfold Handle.chunkReaderWith (n, src)  ------------------------------------------------------------------------------- -- Serial counting using parallel version of countChar ------------------------------------------------------------------------------- -- -- This is only for perf comparison-wcMwlParserial :: Handle -> IO (MArray.Array Int)+wcMwlParserial :: Handle -> IO (MutArray Int) wcMwlParserial src = do     counts <- newCounts-    Stream.mapM_ (countChar counts)+    Stream.fold (Fold.drainMapM (countChar counts))         $ Unicode.decodeUtf8Either-        $ Stream.unfold Handle.read src+        $ Stream.unfold Handle.reader src     return counts  -------------------------------------------------------------------------------
examples/WordFrequency.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleInstances, CPP #-}  {-# OPTIONS_GHC -Wno-orphans #-} @@ -7,28 +7,34 @@ import Data.Function ((&)) import Data.Functor.Identity (Identity(..)) import Data.Hashable (Hashable(..))-import Data.IORef (newIORef, readIORef, modifyIORef')-import Foreign.Storable (Storable(..)) import System.Environment (getArgs)+import Streamly.Data.Array (Unbox)+import Streamly.Internal.Data.IsMap.HashMap ()  import qualified Data.Char as Char import qualified Data.HashMap.Strict as Map import qualified Data.List as List import qualified Data.Ord as Ord-import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Internal.FileSystem.File as File (toBytes)-import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Fold.Container as Fold (toContainerIO)+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.Internal.Data.Stream as Stream (wordsBy)+import qualified Streamly.FileSystem.File as File import qualified Streamly.Unicode.Stream as Unicode -hashArray :: (Storable a, Integral b, Num c) =>+hashArray :: (Unbox a, Integral b, Num c) =>     Fold.Fold Identity a b -> Array.Array a -> c hashArray f arr =-      Stream.unfold Array.read arr+      Stream.unfold Array.reader arr     & Stream.fold f     & fromIntegral . runIdentity -instance (Enum a, Storable a) => Hashable (Array.Array a) where+#if MIN_VERSION_hashable(1,4,0)+instance (Eq a, Enum a, Unbox a) => Hashable (Array.Array a) where+#else+instance (Enum a, Unbox a) => Hashable (Array.Array a) where+#endif     hash = hashArray Fold.rollingHash      hashWithSalt salt =@@ -54,25 +60,17 @@ main = do     inFile <- fmap head getArgs +    let counter = Fold.foldl' (\n _ -> n + 1) (0 :: Int)+        classifier = Fold.toContainerIO id counter     -- Write the stream to a hashmap consisting of word counts     mp <--        let-            alter Nothing    = Just <$> newIORef (1 :: Int)-            alter (Just ref) = modifyIORef' ref (+ 1) >> return (Just ref)-        in File.toBytes inFile                  -- SerialT IO Word8-         & Unicode.decodeLatin1                 -- SerialT IO Char-         & Stream.map toLower                   -- SerialT IO Char-         & Stream.wordsBy isSpace Fold.toList   -- SerialT IO String-         & Stream.filter (all isAlpha)          -- SerialT IO String-         & Stream.foldlM' (flip (Map.alterF alter)) (return Map.empty) -- IO (Map String (IORef Int))--    -- Print the top hashmap entries-    counts <--        let readRef (w, ref) = do-                cnt <- readIORef ref-                return (w, cnt)-         in Map.toList mp-          & mapM readRef+        File.read inFile                       -- Stream IO Word8+         & Unicode.decodeLatin1                -- Stream IO Char+         & fmap toLower                        -- Stream IO Char+         & Stream.wordsBy isSpace Fold.toList  -- Stream IO String+         & Stream.filter (all isAlpha)         -- Stream IO String+         & Stream.fold classifier              -- IO (HashMap String Int) -    traverse_ print $ List.sortOn (Ord.Down . snd) counts-                    & List.take 25+    traverse_ print+        $ List.sortOn (Ord.Down . snd) (Map.toList mp)+        & List.take 25
examples/WordServer.hs view
@@ -4,8 +4,9 @@ import Data.Function ((&)) import Network.Socket (Socket, close) -import qualified Streamly.Prelude as Stream import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Parser as Parser+import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Network.Socket as Socket import qualified Streamly.Network.Inet.TCP as TCP import qualified Streamly.Unicode.Stream as Unicode@@ -20,17 +21,20 @@ -- connection is closed. lookupWords :: Socket -> IO () lookupWords sk =-      Stream.unfold Socket.read sk               -- SerialT IO Word8-    & Unicode.decodeLatin1                       -- SerialT IO Char-    & Stream.wordsBy isSpace Fold.toList         -- SerialT IO String-    & Stream.fromSerial                          -- AheadT  IO String-    & Stream.mapM fetch                          -- AheadT  IO (String, String)-    & Stream.fromAhead                           -- SerialT IO (String, String)-    & Stream.map show                            -- SerialT IO String-    & Stream.intersperse "\n"                    -- SerialT IO String-    & Unicode.encodeStrings Unicode.encodeLatin1 -- SerialT IO (Array Word8)+      Stream.unfold Socket.reader sk             -- Stream IO Word8+    & Unicode.decodeLatin1                       -- Stream IO Char+    & Stream.parseMany word+    & Stream.catRights                           -- Stream IO String+    & Stream.parMapM (Stream.ordered True) fetch+    & fmap show                                  -- Stream IO String+    & Stream.intersperse "\n"                    -- Stream IO String+    & Unicode.encodeStrings Unicode.encodeLatin1 -- Stream IO (Array Word8)     & Stream.fold (Socket.writeChunks sk)        -- IO () +  where++  word = Parser.wordBy isSpace Fold.toList+ serve :: Socket -> IO () serve sk = finally (lookupWords sk) (close sk) @@ -39,8 +43,6 @@ -- "nc" as a client to try it out. main :: IO () main =-      Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket-    & Stream.fromSerial                   -- AsyncT IO ()-    & Stream.mapM serve                   -- AsyncT IO ()-    & Stream.fromAsync                    -- SerialT IO ()-    & Stream.drain                        -- IO ()+      Stream.unfold TCP.acceptorOnPort 8091             -- Stream IO Socket+    & Stream.parMapM id (Socket.forSocketM serve)       -- Stream IO ()+    & Stream.fold Fold.drain                            -- IO ()
streamly-examples.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name:          streamly-examples-version:       0.1.2+version:       0.1.3 license:       Apache-2.0 license-file:  LICENSE author:        Composewell Technologies@@ -22,7 +22,7 @@ category:     Streamly, Streaming, Concurrency, Text, Filesystem, Network, Reactivity stability:     provisional-tested-with: GHC==9.0.1, GHC==8.10.5, GHC==8.8.4, GHC==8.6.5+tested-with: GHC==9.4.4, GHC==9.2.7, GHC==9.0.1, GHC==8.10.7 build-type:    Simple extra-source-files:   Changelog.md@@ -42,7 +42,7 @@ flag fusion-plugin   description: Use fusion plugin for for best performance   manual: True-  default: False+  default: True  flag sdl2   description: Include graphics examples using SDL2@@ -60,18 +60,19 @@  common exe-dependencies   build-depends:-      streamly              == 0.8.2.*-    , base                  >= 4.9   && < 4.17+      streamly              == 0.9.0+    , streamly-core         == 0.1.0+    , base                  >= 4.9   && < 4.18     , directory             >= 1.2   && < 1.4-    , transformers          >= 0.4   && < 0.6+    , transformers          >= 0.4   && < 0.7     , containers            >= 0.5   && < 0.7     , random                >= 1.0.0 && < 2     , exceptions            >= 0.8   && < 0.11     , transformers-base     >= 0.4   && < 0.5     , network               >= 2.6   && < 4-    , hashable              >= 1.2   && < 1.4+    , hashable              >= 1.2   && < 1.5     , unordered-containers  >= 0.2   && < 0.3-    , vector                >= 0.12  && < 0.13+    , vector                >= 0.12  && < 0.14     , mtl                   >= 2.2   && < 3  common exe-options@@ -96,7 +97,7 @@   if flag(fusion-plugin) && !impl(ghcjs) && !impl(ghc < 8.6)     ghc-options: -fplugin Fusion.Plugin     build-depends:-        fusion-plugin     >= 0.2   && < 0.3+        fusion-plugin >= 0.2.6 && < 0.3  common exe-options-threaded   import: exe-options@@ -322,5 +323,23 @@   main-is: FileSystemEvent.hs   if !impl(ghcjs)     buildable: True+  else+    buildable: False++executable DateTimeParser+  import: exe-options+  main-is: DateTimeParser.hs+  if !impl(ghcjs)+    buildable: True+    build-depends: tasty-bench >= 0.3+  else+    buildable: False++executable LogParser+  import: exe-options+  main-is: LogParser.hs+  if !impl(ghcjs)+    buildable: True+    build-depends: tasty-bench >= 0.3   else     buildable: False