core/src/Streamly/Internal/Data/Stream/Serial.hs · streamly

git.iko.soy/streamly

on cbccb77 help sign in

streamly/core/src/Streamly/Internal/Data/Stream/Serial.hs

348 lines 10 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

2{-# OPTIONS_GHC -Wno-orphans #-}

4-- |

5-- Module : Streamly.Internal.Data.Stream.Serial

7--

8-- License : BSD3

9-- Maintainer : streamly@composewell.com

10-- Stability : experimental

11-- Portability : GHC

12--

13-- To run examples in this module:

14--

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

16--

17module Streamly.Internal.Data.Stream.Serial

18 (

19 -- * Serial appending stream

20 SerialT(..)

21 , Serial

22 , serial

24 -- * Construction

25 , cons

26 , consM

27 , repeat

28 , unfoldrM

29 , fromList

30 , list

32 -- * Elimination

33 , toList

34 , foldWith

35 , drain

37 -- * Transformation

38 , map

39 , mapM

40 , filter

41 , foldFilter

42 )

43where

45import Control.Applicative (liftA2)

46import Control.DeepSeq (NFData(..), NFData1(..))

47import Control.Monad.Base (MonadBase(..), liftBaseDefault)

48import Control.Monad.Catch (MonadThrow, throwM)

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

50import Control.Monad.Reader.Class (MonadReader(..))

51import Control.Monad.State.Class (MonadState(..))

52import Control.Monad.Trans.Class (MonadTrans(lift))

53import Data.Foldable (Foldable(foldl'), fold)

54import Data.Functor.Identity (Identity(..), runIdentity)

55import Data.Maybe (fromMaybe)

56import Data.Semigroup (Endo(..))

57#if __GLASGOW_HASKELL__ < 808

58import Data.Semigroup (Semigroup(..))

59#endif

60import GHC.Exts (IsList(..), IsString(..), oneShot)

61import Text.Read

62 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec

63 , readListPrecDefault)

64import Streamly.Internal.BaseCompat ((#.))

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

66import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

67import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

69import qualified Streamly.Internal.Data.Stream.Common as P

70import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D

71import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D

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

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

75import Prelude hiding (map, mapM, repeat, filter)

77#include "Instances.hs"

78#include "inline.hs"

80-- $setup

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

83------------------------------------------------------------------------------

84-- SerialT

85------------------------------------------------------------------------------

87-- | For 'SerialT' streams:

88--

89-- @

90-- (<>) = 'Streamly.Prelude.serial' -- 'Semigroup'

91-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.serial' -- 'Monad'

92-- @

93--

94-- A single 'Monad' bind behaves like a @for@ loop:

95--

96-- >>> :{

97-- Stream.toList $ do

98-- x <- Stream.fromList [1,2] -- foreach x in stream

99-- return x

100-- :}

101-- [1,2]

102--

103-- Nested monad binds behave like nested @for@ loops:

104--

105-- >>> :{

106-- Stream.toList $ do

107-- x <- Stream.fromList [1,2] -- foreach x in stream

108-- y <- Stream.fromList [3,4] -- foreach y in stream

109-- return (x, y)

110-- :}

111-- [(1,3),(1,4),(2,3),(2,4)]

112--

113-- /Since: 0.2.0 ("Streamly")/

114--

115-- @since 0.8.0

116newtype SerialT m a = SerialT {getSerialT :: Stream m a}

117 -- XXX when deriving do we inherit an INLINE?

118 deriving (Semigroup, Monoid, MonadTrans)

119

120-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation

121-- for more details.

122--

123-- /Since: 0.2.0 ("Streamly")/

124--

125-- @since 0.8.0

126type Serial = SerialT IO

127

128------------------------------------------------------------------------------

129-- Generation

130------------------------------------------------------------------------------

131

132{-# INLINE cons #-}

133cons :: a -> SerialT m a -> SerialT m a

134cons x (SerialT ms) = SerialT $ K.cons x ms

135

136{-# INLINE consM #-}

137{-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

138consM :: Monad m => m a -> SerialT m a -> SerialT m a

139consM m (SerialT ms) = SerialT $ K.consM m ms

140

141-- |

142-- Generate an infinite stream by repeating a pure value.

143--

144{-# INLINE_NORMAL repeat #-}

145repeat :: Monad m => a -> SerialT m a

146repeat = SerialT . D.toStreamK . D.repeat

147

148------------------------------------------------------------------------------

149-- Combining

150------------------------------------------------------------------------------

151

152{-# INLINE serial #-}

153serial :: SerialT m a -> SerialT m a -> SerialT m a

154serial = (<>)

155

156------------------------------------------------------------------------------

157-- Monad

158------------------------------------------------------------------------------

159

160instance Monad m => Monad (SerialT m) where

161 return = pure

162

163 -- Benchmarks better with StreamD bind and pure:

164 -- toList, filterAllout, *>, *<, >> (~2x)

165 --

166 -- pure = SerialT . D.fromStreamD . D.fromPure

167 -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)

168

169 -- Benchmarks better with CPS bind and pure:

170 -- Prime sieve (25x)

171 -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)

172 --

173 {-# INLINE (>>=) #-}

174 (>>=) (SerialT m) f = SerialT $ K.bindWith K.serial m (getSerialT . f)

175

176 {-# INLINE (>>) #-}

177 (>>) = (*>)

178

179------------------------------------------------------------------------------

180-- Other instances

181------------------------------------------------------------------------------

182

183{-# INLINE mapM #-}

184mapM :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b

185mapM f (SerialT m) = SerialT $ D.toStreamK $ D.mapM f $ D.fromStreamK m

186

187-- |

188-- @

189-- map = fmap

190-- @

191--

192-- Same as 'fmap'.

193--

194-- @

195-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]

196-- [2,3,4]

197-- @

198--

199-- @since 0.4.0

200{-# INLINE map #-}

201map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b

202map f = mapM (return . f)

203

204{-# INLINE apSerial #-}

205apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b

206apSerial (SerialT m1) (SerialT m2) =

207 SerialT $ D.toStreamK $ D.fromStreamK m1 <*> D.fromStreamK m2

208

209{-# INLINE apSequence #-}

210apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b

211apSequence (SerialT m1) (SerialT m2) =

212 SerialT $ D.toStreamK $ D.fromStreamK m1 *> D.fromStreamK m2

213

214{-# INLINE apDiscardSnd #-}

215apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a

216apDiscardSnd (SerialT m1) (SerialT m2) =

217 SerialT $ D.toStreamK $ D.fromStreamK m1 <* D.fromStreamK m2

218

219-- Note: we need to define all the typeclass operations because we want to

220-- INLINE them.

221instance Monad m => Applicative (SerialT m) where

222 {-# INLINE pure #-}

223 pure = SerialT . K.fromPure

224

225 {-# INLINE (<*>) #-}

226 (<*>) = apSerial

227 -- (<*>) = K.apSerial

228

229 {-# INLINE liftA2 #-}

230 liftA2 f x = (<*>) (fmap f x)

231

232 {-# INLINE (*>) #-}

233 (*>) = apSequence

234 -- (*>) = K.apSerialDiscardFst

235

236 {-# INLINE (<*) #-}

237 (<*) = apDiscardSnd

238 -- (<*) = K.apSerialDiscardSnd

239

240MONAD_COMMON_INSTANCES(SerialT,)

241LIST_INSTANCES(SerialT)

242NFDATA1_INSTANCE(SerialT)

243FOLDABLE_INSTANCE(SerialT)

244TRAVERSABLE_INSTANCE(SerialT)

245

246{-# INLINE toStreamD #-}

247toStreamD :: Applicative m => SerialT m a -> D.Stream m a

248toStreamD (SerialT m) = D.fromStreamK m

249

250{-# INLINE fromStreamD #-}

251fromStreamD :: Monad m => D.Stream m a -> SerialT m a

252fromStreamD m = SerialT $ D.toStreamK m

253

254-- XXX We should only export generation and combinators from this module.

255--

256-- | Include only those elements that pass a predicate.

257--

258{-# INLINE filter #-}

259filter :: Monad m => (a -> Bool) -> SerialT m a -> SerialT m a

260filter p = fromStreamD . D.filter p . toStreamD

261

262-- | Use a filtering fold on a stream.

263--

264-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.fromList [1..10]

265-- 40

266--

267{-# INLINE foldFilter #-}

268foldFilter :: Monad m => Fold m a (Maybe b) -> SerialT m a -> SerialT m b

269foldFilter p = fromStreamD . D.foldFilter p . toStreamD

270

271-- XXX Renamed to foldWith because SerialT has a Foldable instance having

272-- method fold.

273--

274-- | Fold a stream using the supplied left 'Fold' and reducing the resulting

275-- expression strictly at each step. The behavior is similar to 'foldl''. A

276-- 'Fold' can terminate early without consuming the full stream. See the

277-- documentation of individual 'Fold's for termination behavior.

278--

279-- >>> Stream.foldWith Fold.sum (Stream.enumerateFromTo 1 100)

280-- 5050

281--

282-- Folds never fail, therefore, they produce a default value even when no input

283-- is provided. It means we can always fold an empty stream and get a valid

284-- result. For example:

285--

286-- >>> Stream.foldWith Fold.sum Stream.nil

287-- 0

288--

289-- However, 'foldMany' on an empty stream results in an empty stream.

290-- Therefore, @Stream.foldWith f@ is not the same as @Stream.head . Stream.foldMany

291-- f@.

292--

293-- @foldWith f = Stream.parse (Parser.fromFold f)@

294--

295-- /Pre-release/

296{-# INLINE foldWith #-}

297foldWith :: Monad m => Fold m a b -> SerialT m a -> m b

298foldWith fld m = D.fold fld $ toStreamD m

299

300-- XXX Renamed to "list" because fromList is present in IsList instance.

301--

302-- |

303-- @

304-- fromList = 'Prelude.foldr' 'K.cons' 'K.nil'

305-- @

306--

307-- Construct a stream from a list of pure values. This is more efficient than

308-- 'K.fromFoldable' for serial streams.

309--

310-- @since 0.4.0

311{-# INLINE_EARLY list #-}

312list :: Monad m => [a] -> SerialT m a

313list = fromStreamD . D.fromList

314{-# RULES "list fallback to StreamK" [1]

315 forall a. D.toStreamK (D.fromList a) = K.fromFoldable a #-}

316

317------------------------------------------------------------------------------

318-- Construction

319------------------------------------------------------------------------------

320

321-- | Build a stream by unfolding a /monadic/ step function starting from a

322-- seed. The step function returns the next element in the stream and the next

323-- seed value. When it is done it returns 'Nothing' and the stream ends. For

324-- example,

325--

326-- @

327-- let f b =

328-- if b > 3

329-- then return Nothing

330-- else print b >> return (Just (b, b + 1))

331-- in drain $ unfoldrM f 0

332-- @

333-- @

334-- 0

335-- 1

336-- 2

337-- 3

338-- @

339--

340-- /Pre-release/

341--

342{-# INLINE unfoldrM #-}

343unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> SerialT m a

344unfoldrM step seed = SerialT $ D.toStreamK (D.unfoldrM step seed)

345

346{-# INLINE drain #-}

347drain :: Monad m => SerialT m a -> m ()

348drain (SerialT m) = P.drain m