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

git.iko.soy/streamly

on aaa775b help sign in

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

343 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

36 -- * Transformation

37 , map

38 , mapM

39 , filter

40 , foldFilter

41 )

42where

44import Control.Applicative (liftA2)

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

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

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

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

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

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

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

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

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

54import Data.Maybe (fromMaybe)

55import Data.Semigroup (Endo(..))

56#if __GLASGOW_HASKELL__ < 808

57import Data.Semigroup (Semigroup(..))

58#endif

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

60import Text.Read

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

62 , readListPrecDefault)

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

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

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

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

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

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

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

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

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

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

76#include "Instances.hs"

77#include "inline.hs"

79-- $setup

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

82------------------------------------------------------------------------------

83-- SerialT

84------------------------------------------------------------------------------

86-- | For 'SerialT' streams:

87--

88-- @

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

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

91-- @

92--

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

94--

95-- >>> :{

96-- Stream.toList $ do

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

98-- return x

99-- :}

100-- [1,2]

101--

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

103--

104-- >>> :{

105-- Stream.toList $ do

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

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

108-- return (x, y)

109-- :}

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

111--

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

113--

114-- @since 0.8.0

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

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

117 deriving (Semigroup, Monoid, MonadTrans)

118

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

120-- for more details.

121--

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

123--

124-- @since 0.8.0

125type Serial = SerialT IO

126

127------------------------------------------------------------------------------

128-- Generation

129------------------------------------------------------------------------------

130

131{-# INLINE cons #-}

132cons :: a -> SerialT m a -> SerialT m a

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

134

135{-# INLINE consM #-}

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

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

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

139

140-- |

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

142--

143{-# INLINE_NORMAL repeat #-}

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

145repeat = SerialT . D.toStreamK . D.repeat

146

147------------------------------------------------------------------------------

148-- Combining

149------------------------------------------------------------------------------

150

151{-# INLINE serial #-}

152serial :: SerialT m a -> SerialT m a -> SerialT m a

153serial = (<>)

154

155------------------------------------------------------------------------------

156-- Monad

157------------------------------------------------------------------------------

158

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

160 return = pure

161

162 -- Benchmarks better with StreamD bind and pure:

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

164 --

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

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

167

168 -- Benchmarks better with CPS bind and pure:

169 -- Prime sieve (25x)

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

171 --

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

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

174

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

176 (>>) = (*>)

177

178------------------------------------------------------------------------------

179-- Other instances

180------------------------------------------------------------------------------

181

182{-# INLINE mapM #-}

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

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

185

186-- |

187-- @

188-- map = fmap

189-- @

190--

191-- Same as 'fmap'.

192--

193-- @

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

195-- [2,3,4]

196-- @

197--

198-- @since 0.4.0

199{-# INLINE map #-}

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

201map f = mapM (return . f)

202

203{-# INLINE apSerial #-}

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

205apSerial (SerialT m1) (SerialT m2) =

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

207

208{-# INLINE apSequence #-}

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

210apSequence (SerialT m1) (SerialT m2) =

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

212

213{-# INLINE apDiscardSnd #-}

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

215apDiscardSnd (SerialT m1) (SerialT m2) =

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

217

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

219-- INLINE them.

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

221 {-# INLINE pure #-}

222 pure = SerialT . K.fromPure

223

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

225 (<*>) = apSerial

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

227

228 {-# INLINE liftA2 #-}

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

230

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

232 (*>) = apSequence

233 -- (*>) = K.apSerialDiscardFst

234

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

236 (<*) = apDiscardSnd

237 -- (<*) = K.apSerialDiscardSnd

238

239MONAD_COMMON_INSTANCES(SerialT,)

240LIST_INSTANCES(SerialT)

241NFDATA1_INSTANCE(SerialT)

242FOLDABLE_INSTANCE(SerialT)

243TRAVERSABLE_INSTANCE(SerialT)

244

245{-# INLINE toStreamD #-}

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

247toStreamD (SerialT m) = D.fromStreamK m

248

249{-# INLINE fromStreamD #-}

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

251fromStreamD m = SerialT $ D.toStreamK m

252

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

254--

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

256--

257{-# INLINE filter #-}

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

259filter p = fromStreamD . D.filter p . toStreamD

260

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

262--

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

264-- 40

265--

266{-# INLINE foldFilter #-}

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

268foldFilter p = fromStreamD . D.foldFilter p . toStreamD

269

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

271-- method fold.

272--

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

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

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

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

277--

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

279-- 5050

280--

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

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

283-- result. For example:

284--

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

286-- 0

287--

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

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

290-- f@.

291--

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

293--

294-- /Pre-release/

295{-# INLINE foldWith #-}

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

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

298

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

300--

301-- |

302-- @

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

304-- @

305--

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

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

308--

309-- @since 0.4.0

310{-# INLINE_EARLY list #-}

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

312list = fromStreamD . D.fromList

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

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

315

316------------------------------------------------------------------------------

317-- Construction

318------------------------------------------------------------------------------

319

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

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

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

323-- example,

324--

325-- @

326-- let f b =

327-- if b > 3

328-- then return Nothing

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

330-- in drain $ unfoldrM f 0

331-- @

332-- @

333-- 0

334-- 1

335-- 2

336-- 3

337-- @

338--

339-- /Pre-release/

340--

341{-# INLINE unfoldrM #-}

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

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