src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on c8ae480 help sign in

streamly/src/Streamly/Streams/Serial.hs

344 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 UndecidableInstances #-} -- XXX

10-- |

11-- Module : Streamly.Streams.Serial

13--

14-- License : BSD3

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

16-- Stability : experimental

17-- Portability : GHC

18--

19--

20module Streamly.Streams.Serial

21 (

22 -- * Serial appending stream

23 SerialT

24 , StreamT -- deprecated

25 , Serial

26 , serial

27 , serially

29 -- * Serial interleaving stream

30 , WSerialT

31 , InterleavedT -- deprecated

32 , WSerial

33 , wSerial

34 , (<=>) -- deprecated

35 , wSerially

36 , interleaving -- deprecated

38 -- * Transformation

39 , map

40 , mapM

41 )

42where

44import Control.Monad (ap)

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

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

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

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

53import Prelude hiding (map, mapM)

55import Streamly.SVar (rstState)

56import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))

57import qualified Streamly.Streams.StreamK as K

58import qualified Streamly.Streams.StreamD as D

60#include "Instances.hs"

61#include "inline.hs"

63------------------------------------------------------------------------------

64-- SerialT

65------------------------------------------------------------------------------

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

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

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

70-- all elements from the second stream.

71--

72-- @

73-- import Streamly

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

75--

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

77-- @

78-- @

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

80-- @

81--

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

83-- element of the stream, serially.

84--

85-- @

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

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

88-- S.yieldM $ print x

89-- @

90-- @

91-- 1

92-- 2

93-- @

94--

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

96--

97-- @

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

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

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

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

102-- @

103-- @

104-- (1,3)

105-- (1,4)

106-- (2,3)

107-- (2,4)

108-- @

109--

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

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

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

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

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

115--

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

117-- 'SerialT'.

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

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

120-- time.

121--

122-- @since 0.2.0

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

124 deriving (Semigroup, Monoid, MonadTrans)

125

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

127-- for more details.

128--

129-- @since 0.2.0

130type Serial = SerialT IO

131

132-- |

133-- @since 0.1.0

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

135type StreamT = SerialT

136

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

138--

139-- @since 0.1.0

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

141serially = adapt

142

143instance IsStream SerialT where

144 toStream = getSerialT

145 fromStream = SerialT

146

147 {-# INLINE consM #-}

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

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

150 consM m r = fromStream $ K.consMSerial m (toStream r)

151

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

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

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

155 m |: r = fromStream $ K.consMSerial m (toStream r)

156

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

158-- Semigroup

159------------------------------------------------------------------------------

160

161-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.

162-- Appends two streams sequentially, yielding all elements from the first

163-- stream, and then all elements from the second stream.

164--

165-- @since 0.2.0

166{-# INLINE serial #-}

167serial :: IsStream t => t m a -> t m a -> t m a

168serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

169 unStream (K.serial (toStream m1) (toStream m2))

170 (rstState st) stp sng yld

171

172------------------------------------------------------------------------------

173-- Monad

174------------------------------------------------------------------------------

175

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

177 return = pure

178 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->

179 let run x = unStream x (rstState st) stp sng yld

180 single a = run $ toStream (f a)

181 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

182 in m (rstState st) stp single yieldk

183

184------------------------------------------------------------------------------

185-- Other instances

186------------------------------------------------------------------------------

187

188{-# INLINE_EARLY mapM #-}

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

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

191

192-- | Same as 'fmap'.

193--

194-- @since 0.4.0

195{-# INLINE map #-}

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

197map f = mapM (return . f)

198

199MONAD_APPLICATIVE_INSTANCE(SerialT,)

200MONAD_COMMON_INSTANCES(SerialT,)

201

202------------------------------------------------------------------------------

203-- WSerialT

204------------------------------------------------------------------------------

205

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

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

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

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

210--

211-- @

212-- import Streamly

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

214--

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

216-- @

217-- @

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

219-- @

220--

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

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

223--

224--

225-- @

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

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

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

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

230-- @

231-- @

232-- (1,3)

233-- (2,3)

234-- (1,4)

235-- (2,4)

236-- @

237--

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

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

240-- stream.

241--

242-- @since 0.2.0

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

244 deriving (MonadTrans)

245

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

247-- documentation for more details.

248--

249-- @since 0.2.0

250type WSerial = WSerialT IO

251

252-- |

253-- @since 0.1.0

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

255type InterleavedT = WSerialT

256

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

258--

259-- @since 0.2.0

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

261wSerially = adapt

262

263-- | Same as 'wSerially'.

264--

265-- @since 0.1.0

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

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

268interleaving = wSerially

269

270instance IsStream WSerialT where

271 toStream = getWSerialT

272 fromStream = WSerialT

273

274 {-# INLINE consM #-}

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

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

277 consM m r = fromStream $ K.consMSerial m (toStream r)

278

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

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

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

282 m |: r = fromStream $ K.consMSerial m (toStream r)

283

284------------------------------------------------------------------------------

285-- Semigroup

286------------------------------------------------------------------------------

287

288{-# INLINE interleave #-}

289interleave :: Stream m a -> Stream m a -> Stream m a

290interleave m1 m2 = Stream $ \st stp sng yld -> do

291 let stop = unStream m2 (rstState st) stp sng yld

292 single a = yld a m2

293 yieldk a r = yld a (interleave m2 r)

294 unStream m1 (rstState st) stop single yieldk

295

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

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

298--

299-- @since 0.2.0

300{-# INLINE wSerial #-}

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

302wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

303 unStream (interleave (toStream m1) (toStream m2))

304 (rstState st) stp sng yld

305

306instance Semigroup (WSerialT m a) where

307 (<>) = wSerial

308

309infixr 5 <=>

310

311-- | Same as 'wSerial'.

312--

313-- @since 0.1.0

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

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

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

317(<=>) = wSerial

318

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

320-- Monoid

321------------------------------------------------------------------------------

322

323instance Monoid (WSerialT m a) where

324 mempty = K.nil

325 mappend = (<>)

326

327------------------------------------------------------------------------------

328-- Monad

329------------------------------------------------------------------------------

330

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

332 return = pure

333 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->

334 let run x = unStream x (rstState st) stp sng yld

335 single a = run $ toStream (f a)

336 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

337 in m (rstState st) stp single yieldk

338

339------------------------------------------------------------------------------

340-- Other instances

341------------------------------------------------------------------------------

342

343MONAD_APPLICATIVE_INSTANCE(WSerialT,)

344MONAD_COMMON_INSTANCES(WSerialT,)