mirror of
https://github.com/composewell/streamly.git
synced 2024-09-11 11:37:22 +03:00
Replace SerialT with Stream
This commit is contained in:
parent
d58ebdc7b3
commit
3dc72698cf
@ -62,7 +62,7 @@ import GHC.Base (Int(..))
|
||||
import GHC.IO (unsafePerformIO)
|
||||
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream (Stream)
|
||||
import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
|
||||
@ -152,13 +152,13 @@ fromStreamD :: MonadIO m => D.Stream m a -> m (Array a)
|
||||
fromStreamD = D.fold write
|
||||
|
||||
{-# INLINE fromStreamN #-}
|
||||
fromStreamN :: MonadIO m => Int -> SerialT m a -> m (Array a)
|
||||
fromStreamN :: MonadIO m => Int -> Stream m a -> m (Array a)
|
||||
fromStreamN n m = do
|
||||
when (n < 0) $ error "fromStreamN: negative write count specified"
|
||||
fromStreamDN n $ Stream.toStreamD m
|
||||
|
||||
{-# INLINE fromStream #-}
|
||||
fromStream :: MonadIO m => SerialT m a -> m (Array a)
|
||||
fromStream :: MonadIO m => Stream m a -> m (Array a)
|
||||
fromStream = fromStreamD . Stream.toStreamD
|
||||
|
||||
{-# INLINABLE fromListN #-}
|
||||
@ -218,11 +218,11 @@ toStreamDRev arr = D.Stream step (length arr - 1)
|
||||
(# x #) -> D.Yield x (I# i - 1)
|
||||
|
||||
{-# INLINE_EARLY toStream #-}
|
||||
toStream :: Monad m => Array a -> SerialT m a
|
||||
toStream :: Monad m => Array a -> Stream m a
|
||||
toStream = Stream.fromStreamD . toStreamD
|
||||
|
||||
{-# INLINE_EARLY toStreamRev #-}
|
||||
toStreamRev :: Monad m => Array a -> SerialT m a
|
||||
toStreamRev :: Monad m => Array a -> Stream m a
|
||||
toStreamRev = Stream.fromStreamD . toStreamDRev
|
||||
|
||||
-------------------------------------------------------------------------------
|
||||
@ -248,7 +248,7 @@ fold :: Monad m => Fold m a b -> Array a -> m b
|
||||
fold f arr = D.fold f (toStreamD arr)
|
||||
|
||||
{-# INLINE streamFold #-}
|
||||
streamFold :: Monad m => (SerialT m a -> m b) -> Array a -> m b
|
||||
streamFold :: Monad m => (Stream m a -> m b) -> Array a -> m b
|
||||
streamFold f arr = f (toStream arr)
|
||||
|
||||
-------------------------------------------------------------------------------
|
||||
|
@ -131,7 +131,7 @@ import Streamly.Internal.Data.Array.Foreign.Type
|
||||
(Array(..), length, asPtrUnsafe)
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Producer.Type (Producer(..))
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream (Stream)
|
||||
import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
import Streamly.Internal.System.IO (unsafeInlineIO)
|
||||
@ -158,7 +158,7 @@ import qualified Streamly.Internal.Data.Unfold as Unfold
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE fromStreamN #-}
|
||||
fromStreamN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)
|
||||
fromStreamN :: (MonadIO m, Storable a) => Int -> Stream m a -> m (Array a)
|
||||
fromStreamN n m = do
|
||||
when (n < 0) $ error "writeN: negative write count specified"
|
||||
A.fromStreamDN n $ Stream.toStreamD m
|
||||
@ -173,7 +173,7 @@ fromStreamN n m = do
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE fromStream #-}
|
||||
fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)
|
||||
fromStream :: (MonadIO m, Storable a) => Stream m a -> m (Array a)
|
||||
fromStream m = P.fold A.write $ Stream.toStreamK m
|
||||
-- write m = A.fromStreamD $ D.fromStreamK m
|
||||
|
||||
@ -369,7 +369,7 @@ getSliceUnsafe index len (Array contents start e) =
|
||||
-- /Pre-release/
|
||||
{-# INLINE splitOn #-}
|
||||
splitOn :: (Monad m, Storable a) =>
|
||||
(a -> Bool) -> Array a -> SerialT m (Array a)
|
||||
(a -> Bool) -> Array a -> Stream m (Array a)
|
||||
splitOn predicate arr =
|
||||
Stream.fromStreamD
|
||||
$ fmap (\(i, len) -> getSliceUnsafe i len arr)
|
||||
@ -435,7 +435,7 @@ getIndex i arr =
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE getIndices #-}
|
||||
getIndices :: (Monad m, Storable a) => SerialT m Int -> Unfold m (Array a) a
|
||||
getIndices :: (Monad m, Storable a) => Stream m Int -> Unfold m (Array a) a
|
||||
getIndices m =
|
||||
let unf = MA.getIndicesD (return . unsafeInlineIO) $ D.fromStreamK $ Stream.toStreamK m
|
||||
in Unfold.lmap A.unsafeThaw unf
|
||||
@ -486,7 +486,7 @@ runPipe f arr = P.runPipe (toArrayMinChunk (length arr)) $ f (A.read arr)
|
||||
-- /Pre-release/
|
||||
{-# INLINE streamTransform #-}
|
||||
streamTransform :: forall m a b. (MonadIO m, Storable a, Storable b)
|
||||
=> (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
|
||||
=> (Stream m a -> Stream m b) -> Array a -> m (Array b)
|
||||
streamTransform f arr =
|
||||
P.fold (A.writeWith (length arr)) $ Stream.toStreamK $ f (A.toStream arr)
|
||||
|
||||
@ -559,5 +559,5 @@ fold f arr = P.fold f (Stream.toStreamK (A.toStream arr))
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE streamFold #-}
|
||||
streamFold :: (Monad m, Storable a) => (SerialT m a -> m b) -> Array a -> m b
|
||||
streamFold :: (Monad m, Storable a) => (Stream m a -> m b) -> Array a -> m b
|
||||
streamFold f arr = f (A.toStream arr)
|
||||
|
@ -34,7 +34,7 @@ where
|
||||
|
||||
import Control.Monad.IO.Class (MonadIO(..))
|
||||
import Streamly.Internal.Data.Unboxed (Storable)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream (Stream)
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
|
||||
import qualified Streamly.Internal.Data.Stream.StreamD as D
|
||||
@ -51,7 +51,7 @@ import Streamly.Internal.Data.Array.Foreign.Mut.Type
|
||||
-- /Pre-release/
|
||||
{-# INLINE splitOn #-}
|
||||
splitOn :: (MonadIO m, Storable a) =>
|
||||
(a -> Bool) -> Array a -> SerialT m (Array a)
|
||||
(a -> Bool) -> Array a -> Stream m (Array a)
|
||||
splitOn predicate arr =
|
||||
Stream.fromStreamD
|
||||
$ fmap (\(i, len) -> getSliceUnsafe i len arr)
|
||||
@ -98,6 +98,6 @@ getSlicesFromLen from len =
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE fromStream #-}
|
||||
fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)
|
||||
fromStream :: (MonadIO m, Storable a) => Stream m a -> m (Array a)
|
||||
fromStream = fromStreamD . Stream.toStreamD
|
||||
-- fromStream (SerialT m) = P.fold write m
|
||||
-- fromStream (Stream m) = P.fold write m
|
||||
|
@ -34,7 +34,7 @@ import Data.Bifunctor (first)
|
||||
import Streamly.Internal.Data.Unboxed (Storable, sizeOf)
|
||||
import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array(..))
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
|
||||
|
||||
import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MArray
|
||||
@ -53,7 +53,7 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream
|
||||
-- /Pre-release/
|
||||
{-# INLINE arraysOf #-}
|
||||
arraysOf :: (MonadIO m, Storable a)
|
||||
=> Int -> SerialT m a -> SerialT m (Array a)
|
||||
=> Int -> Stream m a -> Stream m (Array a)
|
||||
arraysOf n = Stream.fromStreamD . MArray.arraysOf n . Stream.toStreamD
|
||||
|
||||
-------------------------------------------------------------------------------
|
||||
@ -192,7 +192,7 @@ lpackArraysChunksOf n (Fold step1 initial1 extract1) =
|
||||
-- /Internal/
|
||||
{-# INLINE compact #-}
|
||||
compact :: (MonadIO m, Storable a)
|
||||
=> Int -> SerialT m (Array a) -> SerialT m (Array a)
|
||||
=> Int -> Stream m (Array a) -> Stream m (Array a)
|
||||
compact n = Stream.fromStreamD . packArraysChunksOf n . Stream.toStreamD
|
||||
|
||||
-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
|
||||
@ -299,7 +299,7 @@ compactGEFold n = Fold step initial extract
|
||||
--
|
||||
-- /Internal/
|
||||
compactLE :: (MonadThrow m, MonadIO m, Storable a) =>
|
||||
Int -> SerialT m (Array a) -> SerialT m (Array a)
|
||||
Int -> Stream m (Array a) -> Stream m (Array a)
|
||||
compactLE n =
|
||||
Stream.fromStreamD . D.parseMany (compactLEParserD n) . Stream.toStreamD
|
||||
|
||||
@ -309,7 +309,7 @@ compactLE n =
|
||||
-- /Unimplemented/
|
||||
{-# INLINE compactEQ #-}
|
||||
compactEQ :: -- (MonadIO m, Storable a) =>
|
||||
Int -> SerialT m (Array a) -> SerialT m (Array a)
|
||||
Int -> Stream m (Array a) -> Stream m (Array a)
|
||||
compactEQ _n _xs = undefined
|
||||
-- IsStream.fromStreamD $ D.foldMany (compactEQFold n) (IsStream.toStreamD xs)
|
||||
|
||||
@ -320,6 +320,6 @@ compactEQ _n _xs = undefined
|
||||
{-# INLINE compactGE #-}
|
||||
compactGE ::
|
||||
(MonadIO m, Storable a)
|
||||
=> Int -> SerialT m (Array a) -> SerialT m (Array a)
|
||||
=> Int -> Stream m (Array a) -> Stream m (Array a)
|
||||
compactGE n =
|
||||
Stream.fromStreamD . D.foldMany (compactGEFold n) . Stream.toStreamD
|
||||
|
@ -2474,7 +2474,7 @@ zip = zipWithM (curry return)
|
||||
{-# INLINE indexed #-}
|
||||
indexed :: -- forall m a b. Monad m =>
|
||||
Fold m (Int, a) b -> Fold m a b
|
||||
indexed = undefined -- zip (Stream.enumerateFrom 0 :: SerialT m Int)
|
||||
indexed = undefined -- zip (Stream.enumerateFrom 0 :: Stream m Int)
|
||||
|
||||
-- | Change the predicate function of a Fold from @a -> b@ to accept an
|
||||
-- additional state input @(s, a) -> b@. Convenient to filter with an
|
||||
|
@ -414,15 +414,16 @@ import Prelude hiding (concatMap, filter, foldr, map, take)
|
||||
-- >>> :set -XFlexibleContexts
|
||||
-- >>> :set -package streamly
|
||||
-- >>> import Streamly.Data.Fold (Fold)
|
||||
-- >>> import Prelude hiding (concatMap, filter, map)
|
||||
-- >>> import Streamly.Prelude (SerialT)
|
||||
-- >>> import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
-- >>> import qualified Data.Foldable as Foldable
|
||||
-- >>> import qualified Streamly.Prelude as Stream
|
||||
-- >>> import qualified Streamly.Data.Fold as Fold
|
||||
-- >>> import qualified Streamly.Internal.Data.Fold as Fold
|
||||
-- >>> import qualified Streamly.Internal.Data.Fold.Type as Fold
|
||||
-- >>> import qualified Streamly.Internal.Data.Stream.Type as Stream
|
||||
-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
|
||||
-- >>> import qualified Streamly.Internal.Data.Stream.StreamK as StreamK
|
||||
-- >>> import Prelude hiding (concatMap, filter, map)
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
-- The Fold type
|
||||
@ -1227,7 +1228,7 @@ take n (Fold fstep finitial fextract) = Fold step initial extract
|
||||
-- We can append a stream to a fold as follows:
|
||||
--
|
||||
-- >>> :{
|
||||
-- foldAppend :: Monad m => Fold m a b -> SerialT m a -> m (Fold m a b)
|
||||
-- foldAppend :: Monad m => Fold m a b -> Stream m a -> m (Fold m a b)
|
||||
-- foldAppend f = Stream.fold (Fold.duplicate f)
|
||||
-- :}
|
||||
--
|
||||
|
@ -9,9 +9,9 @@
|
||||
-- Stability : pre-release
|
||||
-- Portability : GHC
|
||||
--
|
||||
-- Lists are just a special case of monadic streams. The stream type @SerialT
|
||||
-- Lists are just a special case of monadic streams. The stream type @Stream
|
||||
-- Identity a@ can be used as a replacement for @[a]@. The 'List' type in this
|
||||
-- module is just a newtype wrapper around @SerialT Identity@ for better type
|
||||
-- module is just a newtype wrapper around @Stream Identity@ for better type
|
||||
-- inference when using the 'OverloadedLists' GHC extension. @List a@ provides
|
||||
-- better performance compared to @[a]@. Standard list, string and list
|
||||
-- comprehension syntax can be used with the 'List' type by enabling
|
||||
@ -25,12 +25,12 @@
|
||||
--
|
||||
--
|
||||
-- @
|
||||
-- List $ S.map (+ 1) $ toSerial (1 \`Cons\` Nil)
|
||||
-- List $ S.map (+ 1) $ toStream (1 \`Cons\` Nil)
|
||||
-- @
|
||||
--
|
||||
-- To convert a 'List' to regular lists, you can use any of the following:
|
||||
--
|
||||
-- * @toList . toSerial@ and @toSerial . fromList@
|
||||
-- * @toList . toStream@ and @toStream . fromList@
|
||||
-- * 'Data.Foldable.toList' from "Data.Foldable"
|
||||
-- * 'GHC.Exts.toList' and 'GHC.Exts.fromList' from 'IsList' in "GHC.Exts"
|
||||
--
|
||||
@ -69,10 +69,9 @@ import GHC.Exts (IsList(..), IsString(..))
|
||||
#if __GLASGOW_HASKELL__ < 808
|
||||
import Data.Semigroup (Semigroup(..))
|
||||
#endif
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Stream.Zip (ZipSerialM(..))
|
||||
|
||||
import qualified Streamly.Internal.Data.Stream.Serial as Serial
|
||||
import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
|
||||
import qualified Streamly.Internal.Data.Stream.Type as Stream
|
||||
|
||||
@ -90,8 +89,8 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream
|
||||
--
|
||||
-- | @List a@ is a replacement for @[a]@.
|
||||
--
|
||||
-- @since 0.6.0
|
||||
newtype List a = List { toSerial :: SerialT Identity a }
|
||||
-- /Pre-release/
|
||||
newtype List a = List { toStream :: Stream Identity a }
|
||||
deriving
|
||||
( Show, Read, Eq, Ord, NFData , NFData1
|
||||
, Semigroup, Monoid, Functor, Foldable
|
||||
@ -107,7 +106,7 @@ instance IsList (List a) where
|
||||
{-# INLINE fromList #-}
|
||||
fromList = List . fromList
|
||||
{-# INLINE toList #-}
|
||||
toList = toList . toSerial
|
||||
toList = toList . toStream
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
-- Patterns
|
||||
@ -124,7 +123,7 @@ instance IsList (List a) where
|
||||
--
|
||||
-- @since 0.6.0
|
||||
pattern Nil :: List a
|
||||
pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toSerial -> True) where
|
||||
pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toStream -> True) where
|
||||
Nil = List $ Stream.fromStreamK K.nil
|
||||
|
||||
infixr 5 `Cons`
|
||||
@ -136,13 +135,13 @@ infixr 5 `Cons`
|
||||
pattern Cons :: a -> List a -> List a
|
||||
pattern Cons x xs <-
|
||||
(fmap (second (List . Stream.fromStreamK))
|
||||
. runIdentity . K.uncons . Stream.toStreamK . toSerial
|
||||
. runIdentity . K.uncons . Stream.toStreamK . toStream
|
||||
-> Just (x, xs)
|
||||
)
|
||||
|
||||
where
|
||||
|
||||
Cons x xs = List $ Serial.cons x (toSerial xs)
|
||||
Cons x xs = List $ Stream.cons x (toStream xs)
|
||||
|
||||
{-# COMPLETE Nil, Cons #-}
|
||||
|
||||
@ -183,4 +182,4 @@ fromZipList (ZipList zs) = List $ Stream.fromStreamK $ getZipSerialM zs
|
||||
--
|
||||
-- @since 0.6.0
|
||||
toZipList :: List a -> ZipList a
|
||||
toZipList = ZipList . ZipSerialM . Stream.toStreamK . toSerial
|
||||
toZipList = ZipList . ZipSerialM . Stream.toStreamK . toStream
|
||||
|
@ -1220,7 +1220,7 @@ takeP i p = D.toParserK $ D.takeP i $ D.fromParserK p
|
||||
-- of the parsers fail.
|
||||
--
|
||||
-- An even more efficient implementation can use ParserD type Parser in
|
||||
-- the SerialT stream.
|
||||
-- the stream.
|
||||
--
|
||||
-- /Pre-release/
|
||||
--
|
||||
|
@ -53,7 +53,7 @@ import Streamly.Internal.Data.SmallArray.Type
|
||||
import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
|
||||
import qualified Streamly.Internal.Data.Stream.StreamD as D
|
||||
import qualified Streamly.Internal.Data.Fold.Type as FL
|
||||
@ -161,17 +161,17 @@ instance NFData a => NFData (SmallArray a) where
|
||||
--
|
||||
-- For optimal performance use this with @n@ <= 128.
|
||||
{-# INLINE fromStreamN #-}
|
||||
fromStreamN :: MonadIO m => Int -> SerialT m a -> m (SmallArray a)
|
||||
fromStreamN :: MonadIO m => Int -> Stream m a -> m (SmallArray a)
|
||||
fromStreamN n m = do
|
||||
when (n < 0) $ error "fromStreamN: negative write count specified"
|
||||
fromStreamDN n $ D.fromStreamK $ Stream.toStreamK m
|
||||
|
||||
{-# INLINE_EARLY toStream #-}
|
||||
toStream :: Monad m => SmallArray a -> SerialT m a
|
||||
toStream :: Monad m => SmallArray a -> Stream m a
|
||||
toStream = Stream.fromStreamK . D.toStreamK . toStreamD
|
||||
|
||||
{-# INLINE_EARLY toStreamRev #-}
|
||||
toStreamRev :: Monad m => SmallArray a -> SerialT m a
|
||||
toStreamRev :: Monad m => SmallArray a -> Stream m a
|
||||
toStreamRev = Stream.fromStreamK . D.toStreamK . toStreamDRev
|
||||
|
||||
{-# INLINE fold #-}
|
||||
@ -179,7 +179,7 @@ fold :: Monad m => Fold m a b -> SmallArray a -> m b
|
||||
fold f arr = D.fold f (toStreamD arr)
|
||||
|
||||
{-# INLINE streamFold #-}
|
||||
streamFold :: Monad m => (SerialT m a -> m b) -> SmallArray a -> m b
|
||||
streamFold :: Monad m => (Stream m a -> m b) -> SmallArray a -> m b
|
||||
streamFold f arr = f (toStream arr)
|
||||
|
||||
{-# INLINE_NORMAL read #-}
|
||||
|
@ -364,7 +364,7 @@ foldxM = IsStream.foldlMx'
|
||||
-- /Since: 0.8.0 (signature change)/
|
||||
{-# INLINE foldlM' #-}
|
||||
foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b
|
||||
foldlM' step begin m = S.foldlM' step begin $ IsStream.toStreamS m
|
||||
foldlM' step begin = S.foldlM' step begin . IsStream.toStreamS
|
||||
|
||||
------------------------------------------------------------------------------
|
||||
-- Running a sink
|
||||
|
@ -36,7 +36,7 @@ import Streamly.Internal.Control.ForkLifted (doFork)
|
||||
import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
|
||||
import Streamly.Internal.Data.Fold.SVar (write, writeLimited)
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
|
||||
|
||||
import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
|
||||
@ -207,7 +207,7 @@ fromProducer sv = K.mkStream $ \st yld sng stp -> do
|
||||
--
|
||||
{-# INLINE newFoldSVar #-}
|
||||
newFoldSVar :: MonadAsync m
|
||||
=> State K.Stream m a -> (SerialT m a -> m b) -> m (SVar K.Stream m a)
|
||||
=> State K.Stream m a -> (Stream m a -> m b) -> m (SVar K.Stream m a)
|
||||
newFoldSVar stt f = do
|
||||
-- Buffer size for the SVar is derived from the current state
|
||||
sv <- newParallelVar StopAny (adaptState stt)
|
||||
@ -369,7 +369,7 @@ pushToFold sv a = do
|
||||
--
|
||||
{-# INLINE teeToSVar #-}
|
||||
teeToSVar :: MonadAsync m =>
|
||||
SVar K.Stream m a -> SerialT m a -> SerialT m a
|
||||
SVar K.Stream m a -> Stream m a -> Stream m a
|
||||
teeToSVar svr m = Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do
|
||||
K.foldStreamShared st yld sng stp (go False $ Stream.toStreamK m)
|
||||
|
||||
|
@ -38,7 +38,7 @@ import Control.Monad.IO.Class (MonadIO(liftIO))
|
||||
import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
|
||||
import Data.Maybe (isNothing)
|
||||
import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
|
||||
import System.Mem (performMajorGC)
|
||||
|
||||
@ -89,7 +89,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)
|
||||
-- XXX this errors out for Parallel/Ahead SVars
|
||||
-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
|
||||
-- be read back from the SVar using 'fromSVar'.
|
||||
toSVar :: MonadAsync m => SVar SerialT m a -> SerialT m a -> m ()
|
||||
toSVar :: MonadAsync m => SVar Stream m a -> Stream m a -> m ()
|
||||
toSVar sv m = do
|
||||
runIn <- askRunInIO
|
||||
liftIO $ enqueue sv (runIn, m)
|
||||
@ -186,7 +186,7 @@ inspect $ hasNoTypeClassesExcept 'fromStreamVar
|
||||
-- combinators.
|
||||
--
|
||||
{-# INLINE fromSVar #-}
|
||||
fromSVar :: MonadAsync m => SVar K.Stream m a -> SerialT m a
|
||||
fromSVar :: MonadAsync m => SVar K.Stream m a -> Stream m a
|
||||
fromSVar sv =
|
||||
Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do
|
||||
ref <- liftIO $ newIORef ()
|
||||
|
@ -131,7 +131,7 @@ import qualified GHC.IO.Device as RawIO
|
||||
import Streamly.Internal.Data.Array.Foreign.Type
|
||||
(Array(..), byteLength, unsafeFreeze, asPtrUnsafe)
|
||||
import Streamly.Internal.System.IO (defaultChunkSize)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Stream.IsStream.Type
|
||||
(IsStream, mkStream, fromStreamD)
|
||||
#if !defined(mingw32_HOST_OS)
|
||||
@ -353,7 +353,7 @@ read = AS.concat . readArrays
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeArrays #-}
|
||||
writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
|
||||
writeArrays :: (MonadIO m, Storable a) => Handle -> Stream m (Array a) -> m ()
|
||||
writeArrays h = S.mapM_ (liftIO . writeArray h)
|
||||
|
||||
-- | Write a stream of arrays to a handle after coalescing them in chunks of
|
||||
@ -364,7 +364,7 @@ writeArrays h = S.mapM_ (liftIO . writeArray h)
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeArraysPackedUpto #-}
|
||||
writeArraysPackedUpto :: (MonadIO m, Storable a)
|
||||
=> Int -> Handle -> SerialT m (Array a) -> m ()
|
||||
=> Int -> Handle -> Stream m (Array a) -> m ()
|
||||
writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs
|
||||
|
||||
#if !defined(mingw32_HOST_OS)
|
||||
@ -374,7 +374,7 @@ writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writev #-}
|
||||
writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()
|
||||
writev :: MonadIO m => Handle -> Stream m (Array RawIO.IOVec) -> m ()
|
||||
writev h = S.mapM_ (liftIO . writeIOVec h)
|
||||
|
||||
-- XXX this is incomplete
|
||||
@ -386,7 +386,7 @@ writev h = S.mapM_ (liftIO . writeIOVec h)
|
||||
-- @since 0.7.0
|
||||
{-# INLINE _writevArraysPackedUpto #-}
|
||||
_writevArraysPackedUpto :: MonadIO m
|
||||
=> Int -> Handle -> SerialT m (Array a) -> m ()
|
||||
=> Int -> Handle -> Stream m (Array a) -> m ()
|
||||
_writevArraysPackedUpto n h xs =
|
||||
writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)
|
||||
-}
|
||||
@ -406,7 +406,7 @@ _writevArraysPackedUpto n h xs =
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeInChunksOf #-}
|
||||
writeInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
|
||||
writeInChunksOf :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()
|
||||
writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m
|
||||
|
||||
-- > write = 'writeInChunksOf' A.defaultChunkSize
|
||||
@ -417,12 +417,12 @@ writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE write #-}
|
||||
write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
|
||||
write :: MonadIO m => Handle -> Stream m Word8 -> m ()
|
||||
write = writeInChunksOf defaultChunkSize
|
||||
|
||||
{-
|
||||
{-# INLINE write #-}
|
||||
write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
|
||||
write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()
|
||||
write = toHandleWith A.defaultChunkSize
|
||||
-}
|
||||
|
||||
@ -449,7 +449,7 @@ readUtf8 = decodeUtf8 . read
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeUtf8 #-}
|
||||
writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
|
||||
writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
|
||||
writeUtf8 h s = write h $ encodeUtf8 s
|
||||
|
||||
-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
|
||||
|
@ -98,7 +98,7 @@ import Streamly.Internal.Control.Concurrent (MonadAsync)
|
||||
import Streamly.Internal.Data.Fold.Type (Fold(..))
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)
|
||||
-- import Streamly.Data.Fold (Fold)
|
||||
-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
|
||||
@ -324,7 +324,7 @@ readShared = undefined
|
||||
|
||||
{-# INLINE fromChunksMode #-}
|
||||
fromChunksMode :: (MonadAsync m, MonadCatch m)
|
||||
=> IOMode -> FilePath -> SerialT m (Array a) -> m ()
|
||||
=> IOMode -> FilePath -> Stream m (Array a) -> m ()
|
||||
fromChunksMode mode file xs = S.drain $
|
||||
withFile file mode (\h -> S.mapM (FH.putChunk h) xs)
|
||||
|
||||
@ -333,7 +333,7 @@ fromChunksMode mode file xs = S.drain $
|
||||
-- @since 0.7.0
|
||||
{-# INLINE fromChunks #-}
|
||||
fromChunks :: (MonadAsync m, MonadCatch m)
|
||||
=> FilePath -> SerialT m (Array a) -> m ()
|
||||
=> FilePath -> Stream m (Array a) -> m ()
|
||||
fromChunks = fromChunksMode WriteMode
|
||||
|
||||
-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
|
||||
@ -351,7 +351,7 @@ fromChunks = fromChunksMode WriteMode
|
||||
-- @since 0.9.0
|
||||
{-# INLINE fromBytesWith #-}
|
||||
fromBytesWith :: (MonadAsync m, MonadCatch m)
|
||||
=> Int -> FilePath -> SerialT m Word8 -> m ()
|
||||
=> Int -> FilePath -> Stream m Word8 -> m ()
|
||||
fromBytesWith n file xs = fromChunks file $ AS.arraysOf n xs
|
||||
|
||||
-- > write = 'writeWith' defaultChunkSize
|
||||
@ -363,12 +363,12 @@ fromBytesWith n file xs = fromChunks file $ AS.arraysOf n xs
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE fromBytes #-}
|
||||
fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
|
||||
fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> Stream m Word8 -> m ()
|
||||
fromBytes = fromBytesWith defaultChunkSize
|
||||
|
||||
{-
|
||||
{-# INLINE write #-}
|
||||
write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
|
||||
write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()
|
||||
write = toHandleWith A.defaultChunkSize
|
||||
-}
|
||||
|
||||
@ -424,7 +424,7 @@ write = writeWith defaultChunkSize
|
||||
-- @since 0.7.0
|
||||
{-# INLINE appendChunks #-}
|
||||
appendChunks :: (MonadAsync m, MonadCatch m)
|
||||
=> FilePath -> SerialT m (Array a) -> m ()
|
||||
=> FilePath -> Stream m (Array a) -> m ()
|
||||
appendChunks = fromChunksMode AppendMode
|
||||
|
||||
-- | Like 'append' but provides control over the write buffer. Output will
|
||||
@ -434,7 +434,7 @@ appendChunks = fromChunksMode AppendMode
|
||||
-- @since 0.9.0
|
||||
{-# INLINE appendWith #-}
|
||||
appendWith :: (MonadAsync m, MonadCatch m)
|
||||
=> Int -> FilePath -> SerialT m Word8 -> m ()
|
||||
=> Int -> FilePath -> Stream m Word8 -> m ()
|
||||
appendWith n file xs = appendChunks file $ AS.arraysOf n xs
|
||||
|
||||
-- | Append a byte stream to a file. Combines the bytes in chunks of size up to
|
||||
@ -444,7 +444,7 @@ appendWith n file xs = appendChunks file $ AS.arraysOf n xs
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE append #-}
|
||||
append :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
|
||||
append :: (MonadAsync m, MonadCatch m) => FilePath -> Stream m Word8 -> m ()
|
||||
append = appendWith defaultChunkSize
|
||||
|
||||
{-
|
||||
@ -452,7 +452,7 @@ append = appendWith defaultChunkSize
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE appendShared #-}
|
||||
appendShared :: MonadIO m => Handle -> SerialT m Word8 -> m ()
|
||||
appendShared :: MonadIO m => Handle -> Stream m Word8 -> m ()
|
||||
appendShared = undefined
|
||||
-}
|
||||
|
||||
@ -479,7 +479,7 @@ readUtf8 = decodeUtf8 . read
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeUtf8 #-}
|
||||
writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
|
||||
writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
|
||||
writeUtf8 h s = write h $ encodeUtf8 s
|
||||
|
||||
-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
|
||||
|
@ -123,7 +123,7 @@ import Streamly.Internal.Data.Refold.Type (Refold(..))
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
import Streamly.Internal.Data.Array.Foreign.Type
|
||||
(Array(..), writeNUnsafe, unsafeFreezeWithShrink, byteLength)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Stream.IsStream.Type
|
||||
(IsStream, mkStream, fromStreamD)
|
||||
import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
|
||||
@ -424,7 +424,7 @@ putChunk h arr = A.asPtrUnsafe arr $ \ptr ->
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE putChunks #-}
|
||||
putChunks :: MonadIO m => Handle -> SerialT m (Array a) -> m ()
|
||||
putChunks :: MonadIO m => Handle -> Stream m (Array a) -> m ()
|
||||
putChunks h = S.mapM_ (putChunk h)
|
||||
|
||||
-- XXX AS.compact can be written idiomatically in terms of foldMany, just like
|
||||
@ -439,7 +439,7 @@ putChunks h = S.mapM_ (putChunk h)
|
||||
-- @since 0.9.0
|
||||
{-# INLINE putChunksWith #-}
|
||||
putChunksWith :: (MonadIO m, Storable a)
|
||||
=> Int -> Handle -> SerialT m (Array a) -> m ()
|
||||
=> Int -> Handle -> Stream m (Array a) -> m ()
|
||||
putChunksWith n h xs = putChunks h $ AS.compact n xs
|
||||
|
||||
-- | @putBytesWith bufsize handle stream@ writes @stream@ to @handle@
|
||||
@ -450,7 +450,7 @@ putChunksWith n h xs = putChunks h $ AS.compact n xs
|
||||
--
|
||||
-- @since 0.9.0
|
||||
{-# INLINE putBytesWith #-}
|
||||
putBytesWith :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
|
||||
putBytesWith :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()
|
||||
putBytesWith n h m = putChunks h $ S.arraysOf n m
|
||||
|
||||
-- putBytesWith n h m = putChunks h $ AS.arraysOf n m
|
||||
@ -465,7 +465,7 @@ putBytesWith n h m = putChunks h $ S.arraysOf n m
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE putBytes #-}
|
||||
putBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
|
||||
putBytes :: MonadIO m => Handle -> Stream m Word8 -> m ()
|
||||
putBytes = putBytesWith defaultChunkSize
|
||||
|
||||
-- | Write a stream of arrays to a handle. Each array in the stream is written
|
||||
@ -578,7 +578,7 @@ consumer = consumerWith defaultChunkSize
|
||||
|
||||
{-
|
||||
{-# INLINE write #-}
|
||||
write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
|
||||
write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()
|
||||
write = toHandleWith A.defaultChunkSize
|
||||
-}
|
||||
|
||||
@ -609,7 +609,7 @@ readUtf8 = decodeUtf8 . read
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeUtf8 #-}
|
||||
writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
|
||||
writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
|
||||
writeUtf8 h s = write h $ encodeUtf8 s
|
||||
|
||||
-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
|
||||
@ -662,13 +662,13 @@ writeByFrames = undefined
|
||||
-- gets exceeded.
|
||||
{-# INLINE writeByChunksOrSessionsOf #-}
|
||||
writeByChunksOrSessionsOf :: MonadIO m
|
||||
=> Int -> Double -> Handle -> SerialT m Word8 -> m ()
|
||||
=> Int -> Double -> Handle -> Stream m Word8 -> m ()
|
||||
writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
|
||||
|
||||
-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
|
||||
-- gets exceeded.
|
||||
{-# INLINE writeBySessionsOf #-}
|
||||
writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
|
||||
writeBySessionsOf :: MonadIO m => Double -> Handle -> Stream m Word8 -> m ()
|
||||
writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
|
||||
|
||||
-------------------------------------------------------------------------------
|
||||
|
@ -96,7 +96,7 @@ import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
|
||||
import Streamly.Internal.Data.Fold (Fold)
|
||||
import Streamly.Internal.Data.Stream.IsStream.Type
|
||||
(IsStream, mkStream, fromStreamD)
|
||||
import Streamly.Internal.Data.Stream.Serial (SerialT)
|
||||
import Streamly.Internal.Data.Stream.Type (Stream)
|
||||
import Streamly.Internal.Data.Unfold.Type (Unfold(..))
|
||||
-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
|
||||
import Streamly.Internal.System.IO (defaultChunkSize)
|
||||
@ -224,7 +224,7 @@ connectFrom spec local = connectCommon spec (Just local)
|
||||
|
||||
{-# INLINE recvConnectionTuplesWith #-}
|
||||
recvConnectionTuplesWith :: MonadAsync m
|
||||
=> Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)
|
||||
=> Int -> SockSpec -> SockAddr -> Stream m (Socket, SockAddr)
|
||||
recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
|
||||
where
|
||||
step Nothing = do
|
||||
@ -243,7 +243,7 @@ recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing
|
||||
--
|
||||
-- /Pre-release/
|
||||
{-# INLINE connections #-}
|
||||
connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket
|
||||
connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> Stream m Socket
|
||||
connections tcpListenQ spec addr =
|
||||
fst <$> recvConnectionTuplesWith tcpListenQ spec addr
|
||||
|
||||
@ -468,7 +468,7 @@ read = UF.first defaultChunkSize readWith
|
||||
-- @since 0.7.0
|
||||
{-# INLINE putChunks #-}
|
||||
putChunks :: (MonadIO m, Storable a)
|
||||
=> Socket -> SerialT m (Array a) -> m ()
|
||||
=> Socket -> Stream m (Array a) -> m ()
|
||||
putChunks h = S.mapM_ (liftIO . writeChunk h)
|
||||
|
||||
-- | Write a stream of arrays to a socket. Each array in the stream is written
|
||||
@ -514,7 +514,7 @@ writeChunksWithBufferOf = writeChunksWith
|
||||
--
|
||||
-- @since 0.9.0
|
||||
{-# INLINE putBytesWith #-}
|
||||
putBytesWith :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()
|
||||
putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()
|
||||
putBytesWith n h m = putChunks h $ AS.arraysOf n m
|
||||
|
||||
-- | Write a byte stream to a socket. Accumulates the input in chunks of
|
||||
@ -554,7 +554,7 @@ writeMaybesWith n h =
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE putBytes #-}
|
||||
putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()
|
||||
putBytes :: MonadIO m => Socket -> Stream m Word8 -> m ()
|
||||
putBytes = putBytesWith defaultChunkSize
|
||||
|
||||
-- | Write a byte stream to a socket. Accumulates the input in chunks of
|
||||
@ -571,7 +571,7 @@ write = writeWith defaultChunkSize
|
||||
|
||||
{-
|
||||
{-# INLINE write #-}
|
||||
write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
|
||||
write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()
|
||||
write = toHandleWith defaultChunkSize
|
||||
-}
|
||||
|
||||
@ -598,7 +598,7 @@ readUtf8 = decodeUtf8 . read
|
||||
--
|
||||
-- @since 0.7.0
|
||||
{-# INLINE writeUtf8 #-}
|
||||
writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
|
||||
writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()
|
||||
writeUtf8 h s = write h $ encodeUtf8 s
|
||||
|
||||
-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
|
||||
|
@ -110,7 +110,7 @@ main = hspec $
|
||||
it "Show instance" $ do
|
||||
show ([1..3] :: List Int) `shouldBe`
|
||||
#ifdef USE_STREAMLY_LIST
|
||||
"List {toSerial = fromList [1,2,3]}"
|
||||
"List {toStream = fromList [1,2,3]}"
|
||||
#else
|
||||
"[1,2,3]"
|
||||
#endif
|
||||
@ -118,7 +118,7 @@ main = hspec $
|
||||
it "Read instance" $ do
|
||||
(read
|
||||
#ifdef USE_STREAMLY_LIST
|
||||
"List {toSerial = fromList [1,2,3]}"
|
||||
"List {toStream = fromList [1,2,3]}"
|
||||
#else
|
||||
"[1,2,3]"
|
||||
#endif
|
||||
|
Loading…
Reference in New Issue
Block a user