src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 516cab5 help sign in

streamly/src/Streamly/Streams/Serial.hs

338 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 StandaloneDeriving #-}

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 , 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.Monad (ap)

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

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

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

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

54import Prelude hiding (map, mapM)

56import Streamly.SVar (rstState)

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

58import qualified Streamly.Streams.StreamK as K

59import qualified Streamly.Streams.StreamD as D

61#include "Instances.hs"

62#include "inline.h"

64------------------------------------------------------------------------------

65-- SerialT

66------------------------------------------------------------------------------

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

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

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

71-- all elements from the second stream.

72--

73-- @

74-- import Streamly

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

76--

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

78-- @

79-- @

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

81-- @

82--

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

84-- element of the stream, serially.

85--

86-- @

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

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

89-- S.once $ print x

90-- @

91-- @

92-- 1

93-- 2

94-- @

95--

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

97--

98-- @

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

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

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

102-- S.once $ print (x, y)

103-- @

104-- @

105-- (1,3)

106-- (1,4)

107-- (2,3)

108-- (2,4)

109-- @

110--

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

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

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

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

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

116--

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

118-- 'SerialT'.

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

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

121-- time.

122--

123-- @since 0.2.0

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

125 deriving (Semigroup, Monoid, MonadTrans)

126

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

128-- for more details.

129--

130-- @since 0.2.0

131type Serial a = SerialT IO a

132

133-- |

134-- @since 0.1.0

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

136type StreamT = SerialT

137

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

139--

140-- @since 0.1.0

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

142serially = adapt

143

144instance IsStream SerialT where

145 toStream = getSerialT

146 fromStream = SerialT

147

148 {-# INLINE consM #-}

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

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

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

152

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

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

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

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

157

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

159-- Semigroup

160------------------------------------------------------------------------------

161

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

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

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

165--

166-- @since 0.2.0

167{-# INLINE serial #-}

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

169serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)

170

171------------------------------------------------------------------------------

172-- Monad

173------------------------------------------------------------------------------

174

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

176 return = pure

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

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

179 single a = run $ toStream (f a)

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

181 in m (rstState st) stp single yieldk

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{-# INLINE map #-}

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

193map f = mapM (return . f)

194

195MONAD_APPLICATIVE_INSTANCE(SerialT,)

196MONAD_COMMON_INSTANCES(SerialT,)

197

198------------------------------------------------------------------------------

199-- WSerialT

200------------------------------------------------------------------------------

201

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

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

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

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

206--

207-- @

208-- import Streamly

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

210--

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

212-- @

213-- @

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

215-- @

216--

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

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

219--

220--

221-- @

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

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

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

225-- S.once $ print (x, y)

226-- @

227-- @

228-- (1,3)

229-- (2,3)

230-- (1,4)

231-- (2,4)

232-- @

233--

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

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

236-- stream.

237--

238-- @since 0.2.0

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

240 deriving (MonadTrans)

241

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

243-- documentation for more details.

244--

245-- @since 0.2.0

246type WSerial a = WSerialT IO a

247

248-- |

249-- @since 0.1.0

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

251type InterleavedT = WSerialT

252

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

254--

255-- @since 0.2.0

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

257wSerially = adapt

258

259-- | Same as 'wSerially'.

260--

261-- @since 0.1.0

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

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

264interleaving = wSerially

265

266instance IsStream WSerialT where

267 toStream = getWSerialT

268 fromStream = WSerialT

269

270 {-# INLINE consM #-}

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

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

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

274

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

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

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

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

279

280------------------------------------------------------------------------------

281-- Semigroup

282------------------------------------------------------------------------------

283

284{-# INLINE interleave #-}

285interleave :: Stream m a -> Stream m a -> Stream m a

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

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

288 single a = yld a m2

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

290 (unStream m1) (rstState st) stop single yieldk

291

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

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

294--

295-- @since 0.2.0

296{-# INLINE wSerial #-}

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

298wSerial m1 m2 = fromStream $ interleave (toStream m1) (toStream m2)

299

300instance Semigroup (WSerialT m a) where

301 (<>) = wSerial

302

303infixr 5 <=>

304

305-- | Same as 'wSerial'.

306--

307-- @since 0.1.0

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

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

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

311(<=>) = wSerial

312

313------------------------------------------------------------------------------

314-- Monoid

315------------------------------------------------------------------------------

316

317instance Monoid (WSerialT m a) where

318 mempty = K.nil

319 mappend = (<>)

320

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

322-- Monad

323------------------------------------------------------------------------------

324

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

326 return = pure

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

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

329 single a = run $ toStream (f a)

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

331 in m (rstState st) stp single yieldk

332

333------------------------------------------------------------------------------

334-- Other instances

335------------------------------------------------------------------------------

336

337MONAD_APPLICATIVE_INSTANCE(WSerialT,)

338MONAD_COMMON_INSTANCES(WSerialT,)