diff --git a/Changelog.md b/Changelog.md
--- a/Changelog.md
+++ b/Changelog.md
@@ -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
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -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
diff --git a/examples/AcidRain.hs b/examples/AcidRain.hs
--- a/examples/AcidRain.hs
+++ b/examples/AcidRain.hs
@@ -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 ()
diff --git a/examples/CSVParser.hs b/examples/CSVParser.hs
--- a/examples/CSVParser.hs
+++ b/examples/CSVParser.hs
@@ -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 ""
diff --git a/examples/CamelCase.hs b/examples/CamelCase.hs
--- a/examples/CamelCase.hs
+++ b/examples/CamelCase.hs
@@ -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
diff --git a/examples/CirclingSquare.hs b/examples/CirclingSquare.hs
--- a/examples/CirclingSquare.hs
+++ b/examples/CirclingSquare.hs
@@ -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.
diff --git a/examples/CmdClient.hs b/examples/CmdClient.hs
--- a/examples/CmdClient.hs
+++ b/examples/CmdClient.hs
@@ -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)
diff --git a/examples/CmdServer.hs b/examples/CmdServer.hs
--- a/examples/CmdServer.hs
+++ b/examples/CmdServer.hs
@@ -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
diff --git a/examples/ControlFlow.hs b/examples/ControlFlow.hs
--- a/examples/ControlFlow.hs
+++ b/examples/ControlFlow.hs
@@ -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
diff --git a/examples/CoreUtils.hs b/examples/CoreUtils.hs
--- a/examples/CoreUtils.hs
+++ b/examples/CoreUtils.hs
@@ -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
diff --git a/examples/CoreUtilsHandle.hs b/examples/CoreUtilsHandle.hs
--- a/examples/CoreUtilsHandle.hs
+++ b/examples/CoreUtilsHandle.hs
@@ -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
diff --git a/examples/DateTimeParser.hs b/examples/DateTimeParser.hs
new file mode 100644
--- /dev/null
+++ b/examples/DateTimeParser.hs
@@ -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]
diff --git a/examples/EchoServer.hs b/examples/EchoServer.hs
--- a/examples/EchoServer.hs
+++ b/examples/EchoServer.hs
@@ -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)
diff --git a/examples/FileSender.hs b/examples/FileSender.hs
--- a/examples/FileSender.hs
+++ b/examples/FileSender.hs
@@ -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 ()
diff --git a/examples/FileSystemEvent.hs b/examples/FileSystemEvent.hs
--- a/examples/FileSystemEvent.hs
+++ b/examples/FileSystemEvent.hs
@@ -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))
diff --git a/examples/Interop/Conduit.hs b/examples/Interop/Conduit.hs
--- a/examples/Interop/Conduit.hs
+++ b/examples/Interop/Conduit.hs
@@ -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
diff --git a/examples/Interop/Pipes.hs b/examples/Interop/Pipes.hs
--- a/examples/Interop/Pipes.hs
+++ b/examples/Interop/Pipes.hs
@@ -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
diff --git a/examples/Interop/Streaming.hs b/examples/Interop/Streaming.hs
--- a/examples/Interop/Streaming.hs
+++ b/examples/Interop/Streaming.hs
@@ -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
diff --git a/examples/Interop/Vector.hs b/examples/Interop/Vector.hs
--- a/examples/Interop/Vector.hs
+++ b/examples/Interop/Vector.hs
@@ -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
diff --git a/examples/Intro.hs b/examples/Intro.hs
--- a/examples/Intro.hs
+++ b/examples/Intro.hs
@@ -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
diff --git a/examples/ListDir.hs b/examples/ListDir.hs
--- a/examples/ListDir.hs
+++ b/examples/ListDir.hs
@@ -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)
diff --git a/examples/LogParser.hs b/examples/LogParser.hs
new file mode 100644
--- /dev/null
+++ b/examples/LogParser.hs
@@ -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
diff --git a/examples/MergeServer.hs b/examples/MergeServer.hs
--- a/examples/MergeServer.hs
+++ b/examples/MergeServer.hs
@@ -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
diff --git a/examples/MergeSort.hs b/examples/MergeSort.hs
--- a/examples/MergeSort.hs
+++ b/examples/MergeSort.hs
@@ -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
diff --git a/examples/Rate.hs b/examples/Rate.hs
--- a/examples/Rate.hs
+++ b/examples/Rate.hs
@@ -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)
diff --git a/examples/Split.hs b/examples/Split.hs
--- a/examples/Split.hs
+++ b/examples/Split.hs
@@ -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
 
diff --git a/examples/WordCount.hs b/examples/WordCount.hs
--- a/examples/WordCount.hs
+++ b/examples/WordCount.hs
@@ -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
diff --git a/examples/WordCountModular.hs b/examples/WordCountModular.hs
--- a/examples/WordCountModular.hs
+++ b/examples/WordCountModular.hs
@@ -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
diff --git a/examples/WordCountParallel.hs b/examples/WordCountParallel.hs
--- a/examples/WordCountParallel.hs
+++ b/examples/WordCountParallel.hs
@@ -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
diff --git a/examples/WordCountParallelUTF8.hs b/examples/WordCountParallelUTF8.hs
--- a/examples/WordCountParallelUTF8.hs
+++ b/examples/WordCountParallelUTF8.hs
@@ -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
 
 -------------------------------------------------------------------------------
diff --git a/examples/WordFrequency.hs b/examples/WordFrequency.hs
--- a/examples/WordFrequency.hs
+++ b/examples/WordFrequency.hs
@@ -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
diff --git a/examples/WordServer.hs b/examples/WordServer.hs
--- a/examples/WordServer.hs
+++ b/examples/WordServer.hs
@@ -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 ()
diff --git a/streamly-examples.cabal b/streamly-examples.cabal
--- a/streamly-examples.cabal
+++ b/streamly-examples.cabal
@@ -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