3dc7269 · streamly · git.iko.soy

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

compare with master

unifiedsplit

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

62import GHC.IO (unsafePerformIO)

63

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

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

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

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

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

68

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

152fromStreamD = D.fold write

153

154{-# INLINE fromStreamN #-}

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

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

156fromStreamN n m = do

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

158 fromStreamDN n $ Stream.toStreamD m

159

160{-# INLINE fromStream #-}

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

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

162fromStream = fromStreamD . Stream.toStreamD

163

164{-# INLINABLE fromListN #-}

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

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

219

220{-# INLINE_EARLY toStream #-}

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

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

222toStream = Stream.fromStreamD . toStreamD

223

224{-# INLINE_EARLY toStreamRev #-}

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

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

226toStreamRev = Stream.fromStreamD . toStreamDRev

227

228-------------------------------------------------------------------------------

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

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

249

250{-# 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

252streamFold f arr = f (toStream arr)

253

254-------------------------------------------------------------------------------

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

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

131 (Array(..), length, asPtrUnsafe)

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

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

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

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

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

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

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

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

158--

159-- /Pre-release/

160{-# 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)

162fromStreamN n m = do

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

164 A.fromStreamDN n $ Stream.toStreamD m

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

173--

174-- /Pre-release/

175{-# 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)

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

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

179

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

369-- /Pre-release/

370{-# INLINE splitOn #-}

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

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

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

373splitOn predicate arr =

374 Stream.fromStreamD

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

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

435--

436-- /Pre-release/

437{-# 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

439getIndices m =

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

441 in Unfold.lmap A.unsafeThaw unf

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

486-- /Pre-release/

487{-# INLINE streamTransform #-}

488streamTransform :: 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)

490streamTransform f arr =

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

492

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

559--

560-- /Pre-release/

561{-# 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

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

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

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

34

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

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

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

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

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

39

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

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

51-- /Pre-release/

52{-# INLINE splitOn #-}

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

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

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

55splitOn predicate arr =

56 Stream.fromStreamD

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

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

98--

99-- /Pre-release/

100{-# 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)

102fromStream = 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)

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

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

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

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

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

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

39

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

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

53-- /Pre-release/

54{-# INLINE arraysOf #-}

55arraysOf :: (MonadIO m, Storable a)

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

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

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

58

59-------------------------------------------------------------------------------

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

192-- /Internal/

193{-# INLINE compact #-}

194compact :: (MonadIO m, Storable a)

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

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

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

197

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

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

299--

300-- /Internal/

301compactLE :: (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)

303compactLE n =

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

305

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

309-- /Unimplemented/

310{-# INLINE compactEQ #-}

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

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

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

313compactEQ _n _xs = undefined

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

315

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

320{-# INLINE compactGE #-}

321compactGE ::

322 (MonadIO m, Storable a)

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

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

324compactGE n =

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

2474{-# INLINE indexed #-}

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

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

2478

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

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

414-- >>> :set -XFlexibleContexts

415-- >>> :set -package streamly

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

426

427

427------------------------------------------------------------------------------

428------------------------------------------------------------------------------

428-- The Fold type

429-- The Fold type

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

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

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

1228--

1229--

1229-- >>> :{

1230-- >>> :{

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)

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

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

1232-- :}

1233-- :}

1233--

1234--

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

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

9-- Stability : pre-release

10-- Portability : GHC

11--

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

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

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

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

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

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

25--

26--

27-- @

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

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

29-- @

30--

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

32--

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

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

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

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

36--

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

69#if __GLASGOW_HASKELL__ < 808

70import Data.Semigroup (Semigroup(..))

71#endif

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

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

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

74

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

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

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

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

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

78

77

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

90--

89--

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

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

92--

91--

93−-- @since 0.6.0

92+-- /Pre-release/

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

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

95 deriving

94 deriving

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

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

97 , Semigroup, Monoid, Functor, Foldable

96 , Semigroup, Monoid, Functor, Foldable

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

107 {-# INLINE fromList #-}

106 {-# INLINE fromList #-}

108 fromList = List . fromList

107 fromList = List . fromList

109 {-# INLINE toList #-}

108 {-# INLINE toList #-}

110− toList = toList . toSerial

109+ toList = toList . toStream

111

110

112------------------------------------------------------------------------------

111------------------------------------------------------------------------------

113-- Patterns

112-- Patterns

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

124--

123--

125-- @since 0.6.0

124-- @since 0.6.0

126pattern Nil :: List a

125pattern 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

128 Nil = List $ Stream.fromStreamK K.nil

127 Nil = List $ Stream.fromStreamK K.nil

129

128

130infixr 5 `Cons`

129infixr 5 `Cons`

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

136pattern Cons :: a -> List a -> List a

135pattern Cons :: a -> List a -> List a

137pattern Cons x xs <-

136pattern Cons x xs <-

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

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

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

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

140 -> Just (x, xs)

139 -> Just (x, xs)

141 )

140 )

142

141

143 where

142 where

144

143

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

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

146

145

147{-# COMPLETE Nil, Cons #-}

146{-# COMPLETE Nil, Cons #-}

148

147

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

183--

182--

184-- @since 0.6.0

183-- @since 0.6.0

185toZipList :: List a -> ZipList a

184toZipList :: 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

1220-- of the parsers fail.

1221--

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

1223−-- the SerialT stream.

1223+-- the stream.

1224--

1225-- /Pre-release/

1226--

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

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

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

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

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

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

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

57

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

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

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

161--

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

163{-# INLINE fromStreamN #-}

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

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

165fromStreamN n m = do

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

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

168

169{-# INLINE_EARLY toStream #-}

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

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

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

172

173{-# INLINE_EARLY toStreamRev #-}

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

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

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

176

177{-# INLINE fold #-}

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

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

180

181{-# 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

183streamFold f arr = f (toStream arr)

184

185{-# INLINE_NORMAL read #-}

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

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

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

365{-# INLINE foldlM' #-}

366foldlM' :: 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

368

369------------------------------------------------------------------------------

370-- Running a sink

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

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

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

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

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

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

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

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

41

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

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

207--

208{-# INLINE newFoldSVar #-}

209newFoldSVar :: 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)

211newFoldSVar stt f = do

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

213 sv <- newParallelVar StopAny (adaptState stt)

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

369--

370{-# INLINE teeToSVar #-}

371teeToSVar :: 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

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

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

375

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

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

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

39import Data.Maybe (isNothing)

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

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

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

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

43import System.Mem (performMajorGC)

44

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

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

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

91-- 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 ()

93toSVar sv m = do

94 runIn <- askRunInIO

95 liftIO $ enqueue sv (runIn, m)

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

186-- combinators.

187--

188{-# 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

190fromSVar sv =

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

192 ref <- liftIO $ newIORef ()

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

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

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

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

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

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

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

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

136 (IsStream, mkStream, fromStreamD)

137#if !defined(mingw32_HOST_OS)

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

353--

354-- @since 0.7.0

355{-# 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 ()

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

358

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

364-- @since 0.7.0

365{-# INLINE writeArraysPackedUpto #-}

366writeArraysPackedUpto :: (MonadIO m, Storable a)

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

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

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

369

370#if !defined(mingw32_HOST_OS)

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

374--

375-- @since 0.7.0

376{-# INLINE writev #-}

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

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

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

379

380-- XXX this is incomplete

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

386-- @since 0.7.0

387{-# INLINE _writevArraysPackedUpto #-}

388_writevArraysPackedUpto :: MonadIO m

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

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

390_writevArraysPackedUpto n h xs =

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

392-}

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

406--

407-- @since 0.7.0

408{-# INLINE writeInChunksOf #-}

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

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

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

411

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

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

417--

418-- @since 0.7.0

419{-# INLINE write #-}

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

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

421write = writeInChunksOf defaultChunkSize

422

423{-

424{-# INLINE write #-}

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

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

426write = toHandleWith A.defaultChunkSize

427-}

428

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

449--

450-- @since 0.7.0

451{-# INLINE writeUtf8 #-}

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

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

453writeUtf8 h s = write h $ encodeUtf8 s

454

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

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

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

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

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

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

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

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

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

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

324

325{-# INLINE fromChunksMode #-}

326fromChunksMode :: (MonadAsync m, MonadCatch m)

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

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

328fromChunksMode mode file xs = S.drain $

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

330

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

333-- @since 0.7.0

334{-# INLINE fromChunks #-}

335fromChunks :: (MonadAsync m, MonadCatch m)

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

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

337fromChunks = fromChunksMode WriteMode

338

339-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

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

351-- @since 0.9.0

352{-# INLINE fromBytesWith #-}

353fromBytesWith :: (MonadAsync m, MonadCatch m)

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

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

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

356

357-- > write = 'writeWith' defaultChunkSize

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

363--

364-- /Pre-release/

365{-# INLINE fromBytes #-}

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

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

367fromBytes = fromBytesWith defaultChunkSize

368

369{-

370{-# INLINE write #-}

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

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

372write = toHandleWith A.defaultChunkSize

373-}

374

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

424-- @since 0.7.0

425{-# INLINE appendChunks #-}

426appendChunks :: (MonadAsync m, MonadCatch m)

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

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

428appendChunks = fromChunksMode AppendMode

429

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

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

434-- @since 0.9.0

435{-# INLINE appendWith #-}

436appendWith :: (MonadAsync m, MonadCatch m)

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

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

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

439

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

444--

445-- @since 0.7.0

446{-# INLINE append #-}

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

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

448append = appendWith defaultChunkSize

449

450{-

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

452--

453-- @since 0.7.0

454{-# INLINE appendShared #-}

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

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

456appendShared = undefined

457-}

458

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

479--

480-- @since 0.7.0

481{-# INLINE writeUtf8 #-}

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

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

483writeUtf8 h s = write h $ encodeUtf8 s

484

485-- | 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(..))

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

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

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

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

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

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

128 (IsStream, mkStream, fromStreamD)

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

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

424--

425-- @since 0.7.0

426{-# INLINE putChunks #-}

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

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

428putChunks h = S.mapM_ (putChunk h)

429

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

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

439-- @since 0.9.0

440{-# INLINE putChunksWith #-}

441putChunksWith :: (MonadIO m, Storable a)

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

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

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

444

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

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

450--

451-- @since 0.9.0

452{-# INLINE putBytesWith #-}

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

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

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

455

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

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

465--

466-- @since 0.7.0

467{-# INLINE putBytes #-}

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

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

469putBytes = putBytesWith defaultChunkSize

470

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

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

578

579{-

580{-# INLINE write #-}

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

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

582write = toHandleWith A.defaultChunkSize

583-}

584

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

609--

610-- @since 0.7.0

611{-# INLINE writeUtf8 #-}

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

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

613writeUtf8 h s = write h $ encodeUtf8 s

614

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

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

662-- gets exceeded.

663{-# INLINE writeByChunksOrSessionsOf #-}

664writeByChunksOrSessionsOf :: MonadIO m

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

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

666writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined

667

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

669-- gets exceeded.

670{-# INLINE writeBySessionsOf #-}

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

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

672writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n

673

674-------------------------------------------------------------------------------

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

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

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

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

98 (IsStream, mkStream, fromStreamD)

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

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

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

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

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

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

224

225{-# INLINE recvConnectionTuplesWith #-}

226recvConnectionTuplesWith :: MonadAsync m

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

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

228recvConnectionTuplesWith tcpListenQ spec addr = S.unfoldrM step Nothing

229 where

230 step Nothing = do

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

243--

244-- /Pre-release/

245{-# INLINE connections #-}

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

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

247connections tcpListenQ spec addr =

248 fst <$> recvConnectionTuplesWith tcpListenQ spec addr

249

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

468-- @since 0.7.0

469{-# INLINE putChunks #-}

470putChunks :: (MonadIO m, Storable a)

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

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

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

473

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

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

514--

515-- @since 0.9.0

516{-# INLINE putBytesWith #-}

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

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

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

519

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

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

554--

555-- @since 0.7.0

556{-# INLINE putBytes #-}

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

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

558putBytes = putBytesWith defaultChunkSize

559

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

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

571

572{-

573{-# INLINE write #-}

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

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

575write = toHandleWith defaultChunkSize

576-}

577

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

598--

599-- @since 0.7.0

600{-# INLINE writeUtf8 #-}

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

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

602writeUtf8 h s = write h $ encodeUtf8 s

603

604-- | 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 $

110 it "Show instance" $ do

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

112#ifdef USE_STREAMLY_LIST

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

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

114#else

115 "[1,2,3]"

116#endif

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

118 it "Read instance" $ do

119 (read

120#ifdef USE_STREAMLY_LIST

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

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

122#else

123 "[1,2,3]"

124#endif