src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 0258d77 help sign in

streamly/src/Streamly/Streams/Serial.hs

360 lines 11 kB history raw

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE ConstraintKinds #-}

3{-# LANGUAGE FlexibleContexts #-}

4{-# LANGUAGE FlexibleInstances #-}

5{-# LANGUAGE GeneralizedNewtypeDeriving#-}

6{-# LANGUAGE InstanceSigs #-}

7{-# LANGUAGE MultiParamTypeClasses #-}

8{-# LANGUAGE TypeFamilies #-}

9{-# LANGUAGE UndecidableInstances #-} -- XXX

11-- |

12-- Module : Streamly.Streams.Serial

14--

15-- License : BSD3

16-- Maintainer : harendra.kumar@gmail.com

17-- Stability : experimental

18-- Portability : GHC

19--

20--

21module Streamly.Streams.Serial

22 (

23 -- * Serial appending stream

24 SerialT

25 , StreamT -- deprecated

26 , Serial

27 , K.serial

28 , serially

30 -- * Serial interleaving stream

31 , WSerialT

32 , InterleavedT -- deprecated

33 , WSerial

34 , wSerial

35 , (<=>) -- deprecated

36 , wSerially

37 , interleaving -- deprecated

39 -- * Transformation

40 , map

41 , mapM

42 )

43where

45import Control.Applicative (liftA2)

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

47import Control.Monad (ap)

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

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

50-- import Control.Monad.Error.Class (MonadError(..))

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

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

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

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

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

56import Data.Foldable (fold)

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

58import GHC.Exts (IsList(..), IsString(..))

59import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

60 readListPrecDefault)

61import Prelude hiding (map, mapM)

63import Streamly.SVar (adaptState)

64import Streamly.Streams.StreamK (IsStream(..), adapt, Stream, mkStream,

65 foldStream)

66import qualified Streamly.Streams.Prelude as P

67import qualified Streamly.Streams.StreamK as K

68import qualified Streamly.Streams.StreamD as D

70#include "Instances.hs"

71#include "inline.hs"

73------------------------------------------------------------------------------

74-- SerialT

75------------------------------------------------------------------------------

77-- | Deep serial composition or serial composition with depth first traversal.

78-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a

79-- depth first manner, yielding all elements from the first stream, and then

80-- all elements from the second stream.

81--

82-- @

83-- import Streamly

84-- import qualified "Streamly.Prelude" as S

85--

86-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print

87-- @

88-- @

89-- [1,2,3,4]

90-- @

91--

92-- The 'Monad' instance runs the /monadic continuation/ for each

93-- element of the stream, serially.

94--

95-- @

96-- main = 'runStream' . 'serially' $ do

97-- x <- return 1 \<\> return 2

98-- S.yieldM $ print x

99-- @

100-- @

101-- 1

102-- 2

103-- @

104--

105-- 'SerialT' nests streams serially in a depth first manner.

106--

107-- @

108-- main = 'runStream' . 'serially' $ do

109-- x <- return 1 \<\> return 2

110-- y <- return 3 \<\> return 4

111-- S.yieldM $ print (x, y)

112-- @

113-- @

114-- (1,3)

115-- (1,4)

116-- (2,3)

117-- (2,4)

118-- @

119--

120-- This behavior of 'SerialT' is exactly like a list transformer. We call the

121-- monadic code being run for each element of the stream a monadic

122-- continuation. In imperative paradigm we can think of this composition as

123-- nested @for@ loops and the monadic continuation is the body of the loop. The

124-- loop iterates for all elements of the stream.

125--

126-- The 'serially' combinator can be omitted as the default stream type is

127-- 'SerialT'.

128-- Note that serial composition with depth first traversal can be used to

129-- combine an infinite number of streams as it explores only one stream at a

130-- time.

131--

132-- @since 0.2.0

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

134 deriving (Semigroup, Monoid, MonadTrans)

135

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

137-- for more details.

138--

139-- @since 0.2.0

140type Serial = SerialT IO

141

142-- |

143-- @since 0.1.0

144{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}

145type StreamT = SerialT

146

147-- | Fix the type of a polymorphic stream as 'SerialT'.

148--

149-- @since 0.1.0

150serially :: IsStream t => SerialT m a -> t m a

151serially = adapt

152

153instance IsStream SerialT where

154 toStream = getSerialT

155 fromStream = SerialT

156

157 {-# INLINE consM #-}

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

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

160 consM = K.consMSerial

161

162 {-# INLINE (|:) #-}

163 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}

164 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a

165 (|:) = K.consMSerial

166

167------------------------------------------------------------------------------

168-- Monad

169------------------------------------------------------------------------------

170

171-- XXX if foldStreamShared/unShare/adaptState are not efficient in serial case

172-- we can use a rewrite rule for that for the serial case and then just use a

173-- bindWith serial here.

174--

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

176 return = pure

177 m >>= f = mkStream $ \st stp sng yld ->

178 let run x = foldStream st stp sng yld x

179 single a = run $ f a

180 yieldk a r = run $ f a <> (r >>= f)

181 in foldStream (adaptState st) stp single yieldk m

182

183------------------------------------------------------------------------------

184-- Other instances

185------------------------------------------------------------------------------

186

187{-# INLINE_EARLY mapM #-}

188mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

189mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)

190

191-- |

192-- @

193-- map = fmap

194-- @

195--

196-- Same as 'fmap'.

197--

198-- @

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

200-- [2,3,4]

201-- @

202--

203-- @since 0.4.0

204{-# INLINE map #-}

205map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

206map f = mapM (return . f)

207

208MONAD_APPLICATIVE_INSTANCE(SerialT,)

209MONAD_COMMON_INSTANCES(SerialT,)

210

211------------------------------------------------------------------------------

212-- WSerialT

213------------------------------------------------------------------------------

214

215-- | Wide serial composition or serial composition with a breadth first

216-- traversal. The 'Semigroup' instance of 'WSerialT' traverses

217-- the two streams in a breadth first manner. In other words, it interleaves

218-- two streams, yielding one element from each stream alternately.

219--

220-- @

221-- import Streamly

222-- import qualified "Streamly.Prelude" as S

223--

224-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print

225-- @

226-- @

227-- [1,3,2,4]

228-- @

229--

230-- Similarly, the 'Monad' instance interleaves the iterations of the

231-- inner and the outer loop, nesting loops in a breadth first manner.

232--

233--

234-- @

235-- main = 'runStream' . 'wSerially' $ do

236-- x <- return 1 \<\> return 2

237-- y <- return 3 \<\> return 4

238-- S.yieldM $ print (x, y)

239-- @

240-- @

241-- (1,3)

242-- (2,3)

243-- (1,4)

244-- (2,4)

245-- @

246--

247-- Note that a serial composition with breadth first traversal can only combine

248-- a finite number of streams as it needs to retain state for each unfinished

249-- stream.

250--

251-- @since 0.2.0

252newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}

253 deriving (MonadTrans)

254

255-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

256-- documentation for more details.

257--

258-- @since 0.2.0

259type WSerial = WSerialT IO

260

261-- |

262-- @since 0.1.0

263{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}

264type InterleavedT = WSerialT

265

266-- | Fix the type of a polymorphic stream as 'WSerialT'.

267--

268-- @since 0.2.0

269wSerially :: IsStream t => WSerialT m a -> t m a

270wSerially = adapt

271

272-- | Same as 'wSerially'.

273--

274-- @since 0.1.0

275{-# DEPRECATED interleaving "Please use wSerially instead." #-}

276interleaving :: IsStream t => WSerialT m a -> t m a

277interleaving = wSerially

278

279instance IsStream WSerialT where

280 toStream = getWSerialT

281 fromStream = WSerialT

282

283 {-# INLINE consM #-}

284 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}

285 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a

286 consM = K.consMSerial

287

288 {-# INLINE (|:) #-}

289 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}

290 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a

291 (|:) = K.consMSerial

292

293------------------------------------------------------------------------------

294-- Semigroup

295------------------------------------------------------------------------------

296

297-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.

298-- Interleaves two streams, yielding one element from each stream alternately.

299--

300-- @since 0.2.0

301{-# INLINE wSerial #-}

302wSerial :: IsStream t => t m a -> t m a -> t m a

303wSerial m1 m2 = mkStream $ \st stp sng yld -> do

304 let stop = foldStream st stp sng yld m2

305 single a = yld a m2

306 yieldk a r = yld a (wSerial m2 r)

307 foldStream st stop single yieldk m1

308

309instance Semigroup (WSerialT m a) where

310 (<>) = wSerial

311

312infixr 5 <=>

313

314-- | Same as 'wSerial'.

315--

316-- @since 0.1.0

317{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}

318{-# INLINE (<=>) #-}

319(<=>) :: IsStream t => t m a -> t m a -> t m a

320(<=>) = wSerial

321

322------------------------------------------------------------------------------

323-- Monoid

324------------------------------------------------------------------------------

325

326instance Monoid (WSerialT m a) where

327 mempty = K.nil

328 mappend = (<>)

329

330------------------------------------------------------------------------------

331-- Monad

332------------------------------------------------------------------------------

333

334-- XXX if foldStreamShared/unShare/adaptState are not efficient in serial case

335-- we can use a rewrite rule for that for the serial case and then just use a

336-- bindWith interleave here.

337--

338instance Monad m => Monad (WSerialT m) where

339 return = pure

340 m >>= f = mkStream $ \st stp sng yld ->

341 let run x = foldStream st stp sng yld x

342 single a = run $ f a

343 yieldk a r = run $ f a <> (r >>= f)

344 in foldStream (adaptState st) stp single yieldk m

345

346------------------------------------------------------------------------------

347-- Other instances

348------------------------------------------------------------------------------

349

350MONAD_APPLICATIVE_INSTANCE(WSerialT,)

351MONAD_COMMON_INSTANCES(WSerialT,)

352

353LIST_INSTANCES(SerialT)

354LIST_INSTANCES(WSerialT)

355

356FOLDABLE_INSTANCE(SerialT)

357FOLDABLE_INSTANCE(WSerialT)

358

359TRAVERSABLE_INSTANCE(SerialT)

360TRAVERSABLE_INSTANCE(WSerialT)