3dc7269 · streamly · git.iko.soy

git.iko.soy/streamly

Replace SerialT with Stream

commit3dc72698cfe23da0e128590ada8f71be647c49eb authorRanjeet Kumar Ranjan <ranjeetr@composewell.com> dateWed 20 Jul 2022, 15:33 (4 years ago) parentd58ebdc

17 files changed+92−92

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Array.hs+6−6

@@ -62,7 +62,7 @@ import GHC.Base (Int(..))

6262import GHC.IO (unsafePerformIO)

6363

6464import Streamly.Internal.Data.Fold.Type (Fold(..))

65−import Streamly.Internal.Data.Stream.Serial (SerialT)

65+import Streamly.Internal.Data.Stream (Stream)

6666import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))

6767import Streamly.Internal.Data.Unfold.Type (Unfold(..))

6868

@@ -152,13 +152,13 @@ fromStreamD :: MonadIO m => D.Stream m a -> m (Array a)

152152fromStreamD = D.fold write

153153

154154{-# INLINE fromStreamN #-}

155−fromStreamN :: MonadIO m => Int -> SerialT m a -> m (Array a)

155+fromStreamN :: MonadIO m => Int -> Stream m a -> m (Array a)

156156fromStreamN n m = do

157157 when (n < 0) $ error "fromStreamN: negative write count specified"

158158 fromStreamDN n $ Stream.toStreamD m

159159

160160{-# INLINE fromStream #-}

161−fromStream :: MonadIO m => SerialT m a -> m (Array a)

161+fromStream :: MonadIO m => Stream m a -> m (Array a)

162162fromStream = fromStreamD . Stream.toStreamD

163163

164164{-# INLINABLE fromListN #-}

@@ -218,11 +218,11 @@ toStreamDRev arr = D.Stream step (length arr - 1)

218218 (# x #) -> D.Yield x (I# i - 1)

219219

220220{-# INLINE_EARLY toStream #-}

221−toStream :: Monad m => Array a -> SerialT m a

221+toStream :: Monad m => Array a -> Stream m a

222222toStream = Stream.fromStreamD . toStreamD

223223

224224{-# INLINE_EARLY toStreamRev #-}

225−toStreamRev :: Monad m => Array a -> SerialT m a

225+toStreamRev :: Monad m => Array a -> Stream m a

226226toStreamRev = Stream.fromStreamD . toStreamDRev

227227

228228-------------------------------------------------------------------------------

@@ -248,7 +248,7 @@ fold :: Monad m => Fold m a b -> Array a -> m b

248248fold f arr = D.fold f (toStreamD arr)

249249

250250{-# INLINE streamFold #-}

251−streamFold :: Monad m => (SerialT m a -> m b) -> Array a -> m b

251+streamFold :: Monad m => (Stream m a -> m b) -> Array a -> m b

252252streamFold f arr = f (toStream arr)

253253

254254-------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Array/Foreign.hs+7−7

@@ -131,7 +131,7 @@ import Streamly.Internal.Data.Array.Foreign.Type

131131 (Array(..), length, asPtrUnsafe)

132132import Streamly.Internal.Data.Fold.Type (Fold(..))

133133import Streamly.Internal.Data.Producer.Type (Producer(..))

134−import Streamly.Internal.Data.Stream.Serial (SerialT)

134+import Streamly.Internal.Data.Stream (Stream)

135135import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))

136136import Streamly.Internal.Data.Unfold.Type (Unfold(..))

137137import Streamly.Internal.System.IO (unsafeInlineIO)

@@ -158,7 +158,7 @@ import qualified Streamly.Internal.Data.Unfold as Unfold

158158--

159159-- /Pre-release/

160160{-# INLINE fromStreamN #-}

161−fromStreamN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)

161+fromStreamN :: (MonadIO m, Storable a) => Int -> Stream m a -> m (Array a)

162162fromStreamN n m = do

163163 when (n < 0) $ error "writeN: negative write count specified"

164164 A.fromStreamDN n $ Stream.toStreamD m

@@ -173,7 +173,7 @@ fromStreamN n m = do

173173--

174174-- /Pre-release/

175175{-# INLINE fromStream #-}

176−fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

176+fromStream :: (MonadIO m, Storable a) => Stream m a -> m (Array a)

177177fromStream m = P.fold A.write $ Stream.toStreamK m

178178-- write m = A.fromStreamD $ D.fromStreamK m

179179

@@ -369,7 +369,7 @@ getSliceUnsafe index len (Array contents start e) =

369369-- /Pre-release/

370370{-# INLINE splitOn #-}

371371splitOn :: (Monad m, Storable a) =>

372− (a -> Bool) -> Array a -> SerialT m (Array a)

372+ (a -> Bool) -> Array a -> Stream m (Array a)

373373splitOn predicate arr =

374374 Stream.fromStreamD

375375 $ fmap (\(i, len) -> getSliceUnsafe i len arr)

@@ -435,7 +435,7 @@ getIndex i arr =

435435--

436436-- /Pre-release/

437437{-# INLINE getIndices #-}

438−getIndices :: (Monad m, Storable a) => SerialT m Int -> Unfold m (Array a) a

438+getIndices :: (Monad m, Storable a) => Stream m Int -> Unfold m (Array a) a

439439getIndices m =

440440 let unf = MA.getIndicesD (return . unsafeInlineIO) $ D.fromStreamK $ Stream.toStreamK m

441441 in Unfold.lmap A.unsafeThaw unf

@@ -486,7 +486,7 @@ runPipe f arr = P.runPipe (toArrayMinChunk (length arr)) $ f (A.read arr)

486486-- /Pre-release/

487487{-# INLINE streamTransform #-}

488488streamTransform :: forall m a b. (MonadIO m, Storable a, Storable b)

489− => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)

489+ => (Stream m a -> Stream m b) -> Array a -> m (Array b)

490490streamTransform f arr =

491491 P.fold (A.writeWith (length arr)) $ Stream.toStreamK $ f (A.toStream arr)

492492

@@ -559,5 +559,5 @@ fold f arr = P.fold f (Stream.toStreamK (A.toStream arr))

559559--

560560-- /Pre-release/

561561{-# INLINE streamFold #-}

562−streamFold :: (Monad m, Storable a) => (SerialT m a -> m b) -> Array a -> m b

562+streamFold :: (Monad m, Storable a) => (Stream m a -> m b) -> Array a -> m b

563563streamFold f arr = f (A.toStream arr)

core/src/Streamly/Internal/Data/Array/Foreign/Mut.hs+4−4

@@ -34,7 +34,7 @@ where

3434

3535import Control.Monad.IO.Class (MonadIO(..))

3636import Streamly.Internal.Data.Unboxed (Storable)

37−import Streamly.Internal.Data.Stream.Serial (SerialT)

37+import Streamly.Internal.Data.Stream (Stream)

3838import Streamly.Internal.Data.Unfold.Type (Unfold(..))

3939

4040import qualified Streamly.Internal.Data.Stream.StreamD as D

@@ -51,7 +51,7 @@ import Streamly.Internal.Data.Array.Foreign.Mut.Type

5151-- /Pre-release/

5252{-# INLINE splitOn #-}

5353splitOn :: (MonadIO m, Storable a) =>

54− (a -> Bool) -> Array a -> SerialT m (Array a)

54+ (a -> Bool) -> Array a -> Stream m (Array a)

5555splitOn predicate arr =

5656 Stream.fromStreamD

5757 $ fmap (\(i, len) -> getSliceUnsafe i len arr)

@@ -98,6 +98,6 @@ getSlicesFromLen from len =

9898--

9999-- /Pre-release/

100100{-# INLINE fromStream #-}

101−fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

101+fromStream :: (MonadIO m, Storable a) => Stream m a -> m (Array a)

102102fromStream = fromStreamD . Stream.toStreamD

103−-- fromStream (SerialT m) = P.fold write m

103+-- fromStream (Stream m) = P.fold write m

core/src/Streamly/Internal/Data/Array/Stream/Mut/Foreign.hs+6−6

@@ -34,7 +34,7 @@ import Data.Bifunctor (first)

3434import Streamly.Internal.Data.Unboxed (Storable, sizeOf)

3535import Streamly.Internal.Data.Array.Foreign.Mut.Type (Array(..))

3636import Streamly.Internal.Data.Fold.Type (Fold(..))

37−import Streamly.Internal.Data.Stream.Serial (SerialT)

37+import Streamly.Internal.Data.Stream.Type (Stream)

3838import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

3939

4040import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MArray

@@ -53,7 +53,7 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream

5353-- /Pre-release/

5454{-# INLINE arraysOf #-}

5555arraysOf :: (MonadIO m, Storable a)

56− => Int -> SerialT m a -> SerialT m (Array a)

56+ => Int -> Stream m a -> Stream m (Array a)

5757arraysOf n = Stream.fromStreamD . MArray.arraysOf n . Stream.toStreamD

5858

5959-------------------------------------------------------------------------------

@@ -192,7 +192,7 @@ lpackArraysChunksOf n (Fold step1 initial1 extract1) =

192192-- /Internal/

193193{-# INLINE compact #-}

194194compact :: (MonadIO m, Storable a)

195− => Int -> SerialT m (Array a) -> SerialT m (Array a)

195+ => Int -> Stream m (Array a) -> Stream m (Array a)

196196compact n = Stream.fromStreamD . packArraysChunksOf n . Stream.toStreamD

197197

198198-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

@@ -299,7 +299,7 @@ compactGEFold n = Fold step initial extract

299299--

300300-- /Internal/

301301compactLE :: (MonadThrow m, MonadIO m, Storable a) =>

302− Int -> SerialT m (Array a) -> SerialT m (Array a)

302+ Int -> Stream m (Array a) -> Stream m (Array a)

303303compactLE n =

304304 Stream.fromStreamD . D.parseMany (compactLEParserD n) . Stream.toStreamD

305305

@@ -309,7 +309,7 @@ compactLE n =

309309-- /Unimplemented/

310310{-# INLINE compactEQ #-}

311311compactEQ :: -- (MonadIO m, Storable a) =>

312− Int -> SerialT m (Array a) -> SerialT m (Array a)

312+ Int -> Stream m (Array a) -> Stream m (Array a)

313313compactEQ _n _xs = undefined

314314 -- IsStream.fromStreamD $ D.foldMany (compactEQFold n) (IsStream.toStreamD xs)

315315

@@ -320,6 +320,6 @@ compactEQ _n _xs = undefined

320320{-# INLINE compactGE #-}

321321compactGE ::

322322 (MonadIO m, Storable a)

323− => Int -> SerialT m (Array a) -> SerialT m (Array a)

323+ => Int -> Stream m (Array a) -> Stream m (Array a)

324324compactGE n =

325325 Stream.fromStreamD . D.foldMany (compactGEFold n) . Stream.toStreamD

core/src/Streamly/Internal/Data/Fold.hs+1−1

@@ -2474,7 +2474,7 @@ zip = zipWithM (curry return)

24742474{-# INLINE indexed #-}

24752475indexed :: -- forall m a b. Monad m =>

24762476 Fold m (Int, a) b -> Fold m a b

2477−indexed = undefined -- zip (Stream.enumerateFrom 0 :: SerialT m Int)

2477+indexed = undefined -- zip (Stream.enumerateFrom 0 :: Stream m Int)

24782478

24792479-- | Change the predicate function of a Fold from @a -> b@ to accept an

24802480-- additional state input @(s, a) -> b@. Convenient to filter with an

core/src/Streamly/Internal/Data/Fold/Type.hs+4−3

@@ -414,15 +414,16 @@ import Prelude hiding (concatMap, filter, foldr, map, take)

414414-- >>> :set -XFlexibleContexts

415415-- >>> :set -package streamly

416416-- >>> import Streamly.Data.Fold (Fold)

417−-- >>> import Prelude hiding (concatMap, filter, map)

418−-- >>> import Streamly.Prelude (SerialT)

417+-- >>> import Streamly.Internal.Data.Stream.Type (Stream)

419418-- >>> import qualified Data.Foldable as Foldable

420419-- >>> import qualified Streamly.Prelude as Stream

421420-- >>> import qualified Streamly.Data.Fold as Fold

422421-- >>> import qualified Streamly.Internal.Data.Fold as Fold

423422-- >>> import qualified Streamly.Internal.Data.Fold.Type as Fold

423+-- >>> import qualified Streamly.Internal.Data.Stream.Type as Stream

424424-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream

425425-- >>> import qualified Streamly.Internal.Data.Stream.StreamK as StreamK

426+-- >>> import Prelude hiding (concatMap, filter, map)

426427

427428------------------------------------------------------------------------------

428429-- The Fold type

@@ -1227,7 +1228,7 @@ take n (Fold fstep finitial fextract) = Fold step initial extract

12271228-- We can append a stream to a fold as follows:

12281229--

12291230-- >>> :{

1230−-- foldAppend :: Monad m => Fold m a b -> SerialT m a -> m (Fold m a b)

1231+-- foldAppend :: Monad m => Fold m a b -> Stream m a -> m (Fold m a b)

12311232-- foldAppend f = Stream.fold (Fold.duplicate f)

12321233-- :}

12331234--

core/src/Streamly/Internal/Data/List.hs+12−13

@@ -9,9 +9,9 @@

99-- Stability : pre-release

1010-- Portability : GHC

1111--

12−-- Lists are just a special case of monadic streams. The stream type @SerialT

12+-- Lists are just a special case of monadic streams. The stream type @Stream

1313-- Identity a@ can be used as a replacement for @[a]@. The 'List' type in this

14−-- module is just a newtype wrapper around @SerialT Identity@ for better type

14+-- module is just a newtype wrapper around @Stream Identity@ for better type

1515-- inference when using the 'OverloadedLists' GHC extension. @List a@ provides

1616-- better performance compared to @[a]@. Standard list, string and list

1717-- comprehension syntax can be used with the 'List' type by enabling

@@ -25,12 +25,12 @@

2525--

2626--

2727-- @

28−-- List $ S.map (+ 1) $ toSerial (1 \`Cons\` Nil)

28+-- List $ S.map (+ 1) $ toStream (1 \`Cons\` Nil)

2929-- @

3030--

3131-- To convert a 'List' to regular lists, you can use any of the following:

3232--

33−-- * @toList . toSerial@ and @toSerial . fromList@

33+-- * @toList . toStream@ and @toStream . fromList@

3434-- * 'Data.Foldable.toList' from "Data.Foldable"

3535-- * 'GHC.Exts.toList' and 'GHC.Exts.fromList' from 'IsList' in "GHC.Exts"

3636--

@@ -69,10 +69,9 @@ import GHC.Exts (IsList(..), IsString(..))

6969#if __GLASGOW_HASKELL__ < 808

7070import Data.Semigroup (Semigroup(..))

7171#endif

72−import Streamly.Internal.Data.Stream.Serial (SerialT)

72+import Streamly.Internal.Data.Stream.Type (Stream)

7373import Streamly.Internal.Data.Stream.Zip (ZipSerialM(..))

7474

75−import qualified Streamly.Internal.Data.Stream.Serial as Serial

7675import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

7776import qualified Streamly.Internal.Data.Stream.Type as Stream

7877

@@ -90,8 +89,8 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream

9089--

9190-- | @List a@ is a replacement for @[a]@.

9291--

93−-- @since 0.6.0

94−newtype List a = List { toSerial :: SerialT Identity a }

92+-- /Pre-release/

93+newtype List a = List { toStream :: Stream Identity a }

9594 deriving

9695 ( Show, Read, Eq, Ord, NFData , NFData1

9796 , Semigroup, Monoid, Functor, Foldable

@@ -107,7 +106,7 @@ instance IsList (List a) where

107106 {-# INLINE fromList #-}

108107 fromList = List . fromList

109108 {-# INLINE toList #-}

110− toList = toList . toSerial

109+ toList = toList . toStream

111110

112111------------------------------------------------------------------------------

113112-- Patterns

@@ -124,7 +123,7 @@ instance IsList (List a) where

124123--

125124-- @since 0.6.0

126125pattern Nil :: List a

127−pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toSerial -> True) where

126+pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toStream -> True) where

128127 Nil = List $ Stream.fromStreamK K.nil

129128

130129infixr 5 `Cons`

@@ -136,13 +135,13 @@ infixr 5 `Cons`

136135pattern Cons :: a -> List a -> List a

137136pattern Cons x xs <-

138137 (fmap (second (List . Stream.fromStreamK))

139− . runIdentity . K.uncons . Stream.toStreamK . toSerial

138+ . runIdentity . K.uncons . Stream.toStreamK . toStream

140139 -> Just (x, xs)

141140 )

142141

143142 where

144143

145− Cons x xs = List $ Serial.cons x (toSerial xs)

144+ Cons x xs = List $ Stream.cons x (toStream xs)

146145

147146{-# COMPLETE Nil, Cons #-}

148147

@@ -183,4 +182,4 @@ fromZipList (ZipList zs) = List $ Stream.fromStreamK $ getZipSerialM zs

183182--

184183-- @since 0.6.0

185184toZipList :: List a -> ZipList a

186−toZipList = ZipList . ZipSerialM . Stream.toStreamK . toSerial

185+toZipList = ZipList . ZipSerialM . Stream.toStreamK . toStream

core/src/Streamly/Internal/Data/Parser.hs+1−1

@@ -1220,7 +1220,7 @@ takeP i p = D.toParserK $ D.takeP i $ D.fromParserK p

12201220-- of the parsers fail.

12211221--

12221222-- An even more efficient implementation can use ParserD type Parser in

1223−-- the SerialT stream.

1223+-- the stream.

12241224--

12251225-- /Pre-release/

12261226--

src/Streamly/Internal/Data/SmallArray.hs+5−5

@@ -53,7 +53,7 @@ import Streamly.Internal.Data.SmallArray.Type

5353import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

5454import Streamly.Internal.Data.Unfold.Type (Unfold(..))

5555import Streamly.Internal.Data.Fold.Type (Fold(..))

56−import Streamly.Internal.Data.Stream.Serial (SerialT)

56+import Streamly.Internal.Data.Stream.Type (Stream)

5757

5858import qualified Streamly.Internal.Data.Stream.StreamD as D

5959import qualified Streamly.Internal.Data.Fold.Type as FL

@@ -161,17 +161,17 @@ instance NFData a => NFData (SmallArray a) where

161161--

162162-- For optimal performance use this with @n@ <= 128.

163163{-# INLINE fromStreamN #-}

164−fromStreamN :: MonadIO m => Int -> SerialT m a -> m (SmallArray a)

164+fromStreamN :: MonadIO m => Int -> Stream m a -> m (SmallArray a)

165165fromStreamN n m = do

166166 when (n < 0) $ error "fromStreamN: negative write count specified"

167167 fromStreamDN n $ D.fromStreamK $ Stream.toStreamK m

168168

169169{-# INLINE_EARLY toStream #-}

170−toStream :: Monad m => SmallArray a -> SerialT m a

170+toStream :: Monad m => SmallArray a -> Stream m a

171171toStream = Stream.fromStreamK . D.toStreamK . toStreamD

172172

173173{-# INLINE_EARLY toStreamRev #-}

174−toStreamRev :: Monad m => SmallArray a -> SerialT m a

174+toStreamRev :: Monad m => SmallArray a -> Stream m a

175175toStreamRev = Stream.fromStreamK . D.toStreamK . toStreamDRev

176176

177177{-# INLINE fold #-}

@@ -179,7 +179,7 @@ fold :: Monad m => Fold m a b -> SmallArray a -> m b

179179fold f arr = D.fold f (toStreamD arr)

180180

181181{-# INLINE streamFold #-}

182−streamFold :: Monad m => (SerialT m a -> m b) -> SmallArray a -> m b

182+streamFold :: Monad m => (Stream m a -> m b) -> SmallArray a -> m b

183183streamFold f arr = f (toStream arr)

184184

185185{-# INLINE_NORMAL read #-}

src/Streamly/Internal/Data/Stream/IsStream/Eliminate.hs+1−1

@@ -364,7 +364,7 @@ foldxM = IsStream.foldlMx'

364364-- /Since: 0.8.0 (signature change)/

365365{-# INLINE foldlM' #-}

366366foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b

367−foldlM' step begin m = S.foldlM' step begin $ IsStream.toStreamS m

367+foldlM' step begin = S.foldlM' step begin . IsStream.toStreamS

368368

369369------------------------------------------------------------------------------

370370-- Running a sink

src/Streamly/Internal/Data/Stream/SVar/Eliminate.hs+3−3

@@ -36,7 +36,7 @@ import Streamly.Internal.Control.ForkLifted (doFork)

3636import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

3737import Streamly.Internal.Data.Fold.SVar (write, writeLimited)

3838import Streamly.Internal.Data.Fold.Type (Fold(..))

39−import Streamly.Internal.Data.Stream.Serial (SerialT)

39+import Streamly.Internal.Data.Stream.Type (Stream)

4040import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

4141

4242import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

@@ -207,7 +207,7 @@ fromProducer sv = K.mkStream $ \st yld sng stp -> do

207207--

208208{-# INLINE newFoldSVar #-}

209209newFoldSVar :: MonadAsync m

210− => State K.Stream m a -> (SerialT m a -> m b) -> m (SVar K.Stream m a)

210+ => State K.Stream m a -> (Stream m a -> m b) -> m (SVar K.Stream m a)

211211newFoldSVar stt f = do

212212 -- Buffer size for the SVar is derived from the current state

213213 sv <- newParallelVar StopAny (adaptState stt)

@@ -369,7 +369,7 @@ pushToFold sv a = do

369369--

370370{-# INLINE teeToSVar #-}

371371teeToSVar :: MonadAsync m =>

372− SVar K.Stream m a -> SerialT m a -> SerialT m a

372+ SVar K.Stream m a -> Stream m a -> Stream m a

373373teeToSVar svr m = Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do

374374 K.foldStreamShared st yld sng stp (go False $ Stream.toStreamK m)

375375

src/Streamly/Internal/Data/Stream/SVar/Generate.hs+3−3

@@ -38,7 +38,7 @@ import Control.Monad.IO.Class (MonadIO(liftIO))

3838import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)

3939import Data.Maybe (isNothing)

4040import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

41−import Streamly.Internal.Data.Stream.Serial (SerialT)

41+import Streamly.Internal.Data.Stream.Type (Stream)

4242import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

4343import System.Mem (performMajorGC)

4444

@@ -89,7 +89,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)

8989-- XXX this errors out for Parallel/Ahead SVars

9090-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

9191-- be read back from the SVar using 'fromSVar'.

92−toSVar :: MonadAsync m => SVar SerialT m a -> SerialT m a -> m ()

92+toSVar :: MonadAsync m => SVar Stream m a -> Stream m a -> m ()

9393toSVar sv m = do

9494 runIn <- askRunInIO

9595 liftIO $ enqueue sv (runIn, m)

@@ -186,7 +186,7 @@ inspect $ hasNoTypeClassesExcept 'fromStreamVar

186186-- combinators.

187187--

188188{-# INLINE fromSVar #-}

189−fromSVar :: MonadAsync m => SVar K.Stream m a -> SerialT m a

189+fromSVar :: MonadAsync m => SVar K.Stream m a -> Stream m a

190190fromSVar sv =

191191 Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do

192192 ref <- liftIO $ newIORef ()

src/Streamly/Internal/FileSystem/FD.hs+9−9

@@ -131,7 +131,7 @@ import qualified GHC.IO.Device as RawIO

131131import Streamly.Internal.Data.Array.Foreign.Type

132132 (Array(..), byteLength, unsafeFreeze, asPtrUnsafe)

133133import Streamly.Internal.System.IO (defaultChunkSize)

134−import Streamly.Internal.Data.Stream.Serial (SerialT)

134+import Streamly.Internal.Data.Stream.Type (Stream)

135135import Streamly.Internal.Data.Stream.IsStream.Type

136136 (IsStream, mkStream, fromStreamD)

137137#if !defined(mingw32_HOST_OS)

@@ -353,7 +353,7 @@ read = AS.concat . readArrays

353353--

354354-- @since 0.7.0

355355{-# INLINE writeArrays #-}

356−writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()

356+writeArrays :: (MonadIO m, Storable a) => Handle -> Stream m (Array a) -> m ()

357357writeArrays h = S.mapM_ (liftIO . writeArray h)

358358

359359-- | 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)

364364-- @since 0.7.0

365365{-# INLINE writeArraysPackedUpto #-}

366366writeArraysPackedUpto :: (MonadIO m, Storable a)

367− => Int -> Handle -> SerialT m (Array a) -> m ()

367+ => Int -> Handle -> Stream m (Array a) -> m ()

368368writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

369369

370370#if !defined(mingw32_HOST_OS)

@@ -374,7 +374,7 @@ writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

374374--

375375-- @since 0.7.0

376376{-# INLINE writev #-}

377−writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()

377+writev :: MonadIO m => Handle -> Stream m (Array RawIO.IOVec) -> m ()

378378writev h = S.mapM_ (liftIO . writeIOVec h)

379379

380380-- XXX this is incomplete

@@ -386,7 +386,7 @@ writev h = S.mapM_ (liftIO . writeIOVec h)

386386-- @since 0.7.0

387387{-# INLINE _writevArraysPackedUpto #-}

388388_writevArraysPackedUpto :: MonadIO m

389− => Int -> Handle -> SerialT m (Array a) -> m ()

389+ => Int -> Handle -> Stream m (Array a) -> m ()

390390_writevArraysPackedUpto n h xs =

391391 writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)

392392-}

@@ -406,7 +406,7 @@ _writevArraysPackedUpto n h xs =

406406--

407407-- @since 0.7.0

408408{-# INLINE writeInChunksOf #-}

409−writeInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

409+writeInChunksOf :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()

410410writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m

411411

412412-- > write = 'writeInChunksOf' A.defaultChunkSize

@@ -417,12 +417,12 @@ writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m

417417--

418418-- @since 0.7.0

419419{-# INLINE write #-}

420−write :: MonadIO m => Handle -> SerialT m Word8 -> m ()

420+write :: MonadIO m => Handle -> Stream m Word8 -> m ()

421421write = writeInChunksOf defaultChunkSize

422422

423423{-

424424{-# INLINE write #-}

425−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

425+write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()

426426write = toHandleWith A.defaultChunkSize

427427-}

428428

@@ -449,7 +449,7 @@ readUtf8 = decodeUtf8 . read

449449--

450450-- @since 0.7.0

451451{-# INLINE writeUtf8 #-}

452−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

452+writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()

453453writeUtf8 h s = write h $ encodeUtf8 s

454454

455455-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

src/Streamly/Internal/FileSystem/File.hs+11−11

@@ -98,7 +98,7 @@ import Streamly.Internal.Control.Concurrent (MonadAsync)

9898import Streamly.Internal.Data.Fold.Type (Fold(..))

9999import Streamly.Internal.Data.Unfold.Type (Unfold(..))

100100import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)

101−import Streamly.Internal.Data.Stream.Serial (SerialT)

101+import Streamly.Internal.Data.Stream.Type (Stream)

102102import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)

103103-- import Streamly.Data.Fold (Fold)

104104-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

@@ -324,7 +324,7 @@ readShared = undefined

324324

325325{-# INLINE fromChunksMode #-}

326326fromChunksMode :: (MonadAsync m, MonadCatch m)

327− => IOMode -> FilePath -> SerialT m (Array a) -> m ()

327+ => IOMode -> FilePath -> Stream m (Array a) -> m ()

328328fromChunksMode mode file xs = S.drain $

329329 withFile file mode (\h -> S.mapM (FH.putChunk h) xs)

330330

@@ -333,7 +333,7 @@ fromChunksMode mode file xs = S.drain $

333333-- @since 0.7.0

334334{-# INLINE fromChunks #-}

335335fromChunks :: (MonadAsync m, MonadCatch m)

336− => FilePath -> SerialT m (Array a) -> m ()

336+ => FilePath -> Stream m (Array a) -> m ()

337337fromChunks = fromChunksMode WriteMode

338338

339339-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

@@ -351,7 +351,7 @@ fromChunks = fromChunksMode WriteMode

351351-- @since 0.9.0

352352{-# INLINE fromBytesWith #-}

353353fromBytesWith :: (MonadAsync m, MonadCatch m)

354− => Int -> FilePath -> SerialT m Word8 -> m ()

354+ => Int -> FilePath -> Stream m Word8 -> m ()

355355fromBytesWith n file xs = fromChunks file $ AS.arraysOf n xs

356356

357357-- > write = 'writeWith' defaultChunkSize

@@ -363,12 +363,12 @@ fromBytesWith n file xs = fromChunks file $ AS.arraysOf n xs

363363--

364364-- /Pre-release/

365365{-# INLINE fromBytes #-}

366−fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()

366+fromBytes :: (MonadAsync m, MonadCatch m) => FilePath -> Stream m Word8 -> m ()

367367fromBytes = fromBytesWith defaultChunkSize

368368

369369{-

370370{-# INLINE write #-}

371−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

371+write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()

372372write = toHandleWith A.defaultChunkSize

373373-}

374374

@@ -424,7 +424,7 @@ write = writeWith defaultChunkSize

424424-- @since 0.7.0

425425{-# INLINE appendChunks #-}

426426appendChunks :: (MonadAsync m, MonadCatch m)

427− => FilePath -> SerialT m (Array a) -> m ()

427+ => FilePath -> Stream m (Array a) -> m ()

428428appendChunks = fromChunksMode AppendMode

429429

430430-- | Like 'append' but provides control over the write buffer. Output will

@@ -434,7 +434,7 @@ appendChunks = fromChunksMode AppendMode

434434-- @since 0.9.0

435435{-# INLINE appendWith #-}

436436appendWith :: (MonadAsync m, MonadCatch m)

437− => Int -> FilePath -> SerialT m Word8 -> m ()

437+ => Int -> FilePath -> Stream m Word8 -> m ()

438438appendWith n file xs = appendChunks file $ AS.arraysOf n xs

439439

440440-- | 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

444444--

445445-- @since 0.7.0

446446{-# INLINE append #-}

447−append :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()

447+append :: (MonadAsync m, MonadCatch m) => FilePath -> Stream m Word8 -> m ()

448448append = appendWith defaultChunkSize

449449

450450{-

@@ -452,7 +452,7 @@ append = appendWith defaultChunkSize

452452--

453453-- @since 0.7.0

454454{-# INLINE appendShared #-}

455−appendShared :: MonadIO m => Handle -> SerialT m Word8 -> m ()

455+appendShared :: MonadIO m => Handle -> Stream m Word8 -> m ()

456456appendShared = undefined

457457-}

458458

@@ -479,7 +479,7 @@ readUtf8 = decodeUtf8 . read

479479--

480480-- @since 0.7.0

481481{-# INLINE writeUtf8 #-}

482−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

482+writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()

483483writeUtf8 h s = write h $ encodeUtf8 s

484484

485485-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

src/Streamly/Internal/FileSystem/Handle.hs+9−9

@@ -123,7 +123,7 @@ import Streamly.Internal.Data.Refold.Type (Refold(..))

123123import Streamly.Internal.Data.Unfold.Type (Unfold(..))

124124import Streamly.Internal.Data.Array.Foreign.Type

125125 (Array(..), writeNUnsafe, unsafeFreezeWithShrink, byteLength)

126−import Streamly.Internal.Data.Stream.Serial (SerialT)

126+import Streamly.Internal.Data.Stream.Type (Stream)

127127import Streamly.Internal.Data.Stream.IsStream.Type

128128 (IsStream, mkStream, fromStreamD)

129129import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

@@ -424,7 +424,7 @@ putChunk h arr = A.asPtrUnsafe arr $ \ptr ->

424424--

425425-- @since 0.7.0

426426{-# INLINE putChunks #-}

427−putChunks :: MonadIO m => Handle -> SerialT m (Array a) -> m ()

427+putChunks :: MonadIO m => Handle -> Stream m (Array a) -> m ()

428428putChunks h = S.mapM_ (putChunk h)

429429

430430-- XXX AS.compact can be written idiomatically in terms of foldMany, just like

@@ -439,7 +439,7 @@ putChunks h = S.mapM_ (putChunk h)

439439-- @since 0.9.0

440440{-# INLINE putChunksWith #-}

441441putChunksWith :: (MonadIO m, Storable a)

442− => Int -> Handle -> SerialT m (Array a) -> m ()

442+ => Int -> Handle -> Stream m (Array a) -> m ()

443443putChunksWith n h xs = putChunks h $ AS.compact n xs

444444

445445-- | @putBytesWith bufsize handle stream@ writes @stream@ to @handle@

@@ -450,7 +450,7 @@ putChunksWith n h xs = putChunks h $ AS.compact n xs

450450--

451451-- @since 0.9.0

452452{-# INLINE putBytesWith #-}

453−putBytesWith :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

453+putBytesWith :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()

454454putBytesWith n h m = putChunks h $ S.arraysOf n m

455455

456456-- putBytesWith n h m = putChunks h $ AS.arraysOf n m

@@ -465,7 +465,7 @@ putBytesWith n h m = putChunks h $ S.arraysOf n m

465465--

466466-- @since 0.7.0

467467{-# INLINE putBytes #-}

468−putBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()

468+putBytes :: MonadIO m => Handle -> Stream m Word8 -> m ()

469469putBytes = putBytesWith defaultChunkSize

470470

471471-- | Write a stream of arrays to a handle. Each array in the stream is written

@@ -578,7 +578,7 @@ consumer = consumerWith defaultChunkSize

578578

579579{-

580580{-# INLINE write #-}

581−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

581+write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()

582582write = toHandleWith A.defaultChunkSize

583583-}

584584

@@ -609,7 +609,7 @@ readUtf8 = decodeUtf8 . read

609609--

610610-- @since 0.7.0

611611{-# INLINE writeUtf8 #-}

612−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

612+writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()

613613writeUtf8 h s = write h $ encodeUtf8 s

614614

615615-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

@@ -662,13 +662,13 @@ writeByFrames = undefined

662662-- gets exceeded.

663663{-# INLINE writeByChunksOrSessionsOf #-}

664664writeByChunksOrSessionsOf :: MonadIO m

665− => Int -> Double -> Handle -> SerialT m Word8 -> m ()

665+ => Int -> Double -> Handle -> Stream m Word8 -> m ()

666666writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined

667667

668668-- | Write collecting the input in sessions of n seconds or if defaultChunkSize

669669-- gets exceeded.

670670{-# INLINE writeBySessionsOf #-}

671−writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()

671+writeBySessionsOf :: MonadIO m => Double -> Handle -> Stream m Word8 -> m ()

672672writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n

673673

674674-------------------------------------------------------------------------------

src/Streamly/Internal/Network/Socket.hs+8−8

@@ -96,7 +96,7 @@ import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

9696import Streamly.Internal.Data.Fold (Fold)

9797import Streamly.Internal.Data.Stream.IsStream.Type

9898 (IsStream, mkStream, fromStreamD)

99−import Streamly.Internal.Data.Stream.Serial (SerialT)

99+import Streamly.Internal.Data.Stream.Type (Stream)

100100import Streamly.Internal.Data.Unfold.Type (Unfold(..))

101101-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

102102import Streamly.Internal.System.IO (defaultChunkSize)

@@ -224,7 +224,7 @@ connectFrom spec local = connectCommon spec (Just local)

224224

225225{-# INLINE recvConnectionTuplesWith #-}

226226recvConnectionTuplesWith :: MonadAsync m

227− => Int -> SockSpec -> SockAddr -> SerialT m (Socket, SockAddr)

227+ => Int -> SockSpec -> SockAddr -> Stream m (Socket, SockAddr)

228228recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing

229229 where

230230 step Nothing = do

@@ -243,7 +243,7 @@ recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing

243243--

244244-- /Pre-release/

245245{-# INLINE connections #-}

246−connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> SerialT m Socket

246+connections :: MonadAsync m => Int -> SockSpec -> SockAddr -> Stream m Socket

247247connections tcpListenQ spec addr =

248248 fst <$> recvConnectionTuplesWith tcpListenQ spec addr

249249

@@ -468,7 +468,7 @@ read = UF.first defaultChunkSize readWith

468468-- @since 0.7.0

469469{-# INLINE putChunks #-}

470470putChunks :: (MonadIO m, Storable a)

471− => Socket -> SerialT m (Array a) -> m ()

471+ => Socket -> Stream m (Array a) -> m ()

472472putChunks h = S.mapM_ (liftIO . writeChunk h)

473473

474474-- | Write a stream of arrays to a socket. Each array in the stream is written

@@ -514,7 +514,7 @@ writeChunksWithBufferOf = writeChunksWith

514514--

515515-- @since 0.9.0

516516{-# INLINE putBytesWith #-}

517−putBytesWith :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

517+putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()

518518putBytesWith n h m = putChunks h $ AS.arraysOf n m

519519

520520-- | Write a byte stream to a socket. Accumulates the input in chunks of

@@ -554,7 +554,7 @@ writeMaybesWith n h =

554554--

555555-- @since 0.7.0

556556{-# INLINE putBytes #-}

557−putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

557+putBytes :: MonadIO m => Socket -> Stream m Word8 -> m ()

558558putBytes = putBytesWith defaultChunkSize

559559

560560-- | Write a byte stream to a socket. Accumulates the input in chunks of

@@ -571,7 +571,7 @@ write = writeWith defaultChunkSize

571571

572572{-

573573{-# INLINE write #-}

574−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

574+write :: (MonadIO m, Storable a) => Handle -> Stream m a -> m ()

575575write = toHandleWith defaultChunkSize

576576-}

577577

@@ -598,7 +598,7 @@ readUtf8 = decodeUtf8 . read

598598--

599599-- @since 0.7.0

600600{-# INLINE writeUtf8 #-}

601−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

601+writeUtf8 :: MonadIO m => Handle -> Stream m Char -> m ()

602602writeUtf8 h s = write h $ encodeUtf8 s

603603

604604-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

test/Streamly/Test/Data/List.hs+2−2

@@ -110,7 +110,7 @@ main = hspec $

110110 it "Show instance" $ do

111111 show ([1..3] :: List Int) `shouldBe`

112112#ifdef USE_STREAMLY_LIST

113− "List {toSerial = fromList [1,2,3]}"

113+ "List {toStream = fromList [1,2,3]}"

114114#else

115115 "[1,2,3]"

116116#endif

@@ -118,7 +118,7 @@ main = hspec $

118118 it "Read instance" $ do

119119 (read

120120#ifdef USE_STREAMLY_LIST

121− "List {toSerial = fromList [1,2,3]}"

121+ "List {toStream = fromList [1,2,3]}"

122122#else

123123 "[1,2,3]"

124124#endif