src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 3ce061d help sign in

streamly/src/Streamly/Streams/Serial.hs

345 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.yieldM $ 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.yieldM $ 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 $ Stream $ \st stp sng yld ->

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

171 (rstState st) stp sng yld

172

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

174-- Monad

175------------------------------------------------------------------------------

176

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

178 return = pure

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

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

181 single a = run $ toStream (f a)

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

183 in m (rstState st) stp single yieldk

184

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

186-- Other instances

187------------------------------------------------------------------------------

188

189{-# INLINE_EARLY mapM #-}

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

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

192

193-- | Same as 'fmap'.

194--

195-- @since 0.4.0

196{-# INLINE map #-}

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

198map f = mapM (return . f)

199

200MONAD_APPLICATIVE_INSTANCE(SerialT,)

201MONAD_COMMON_INSTANCES(SerialT,)

202

203------------------------------------------------------------------------------

204-- WSerialT

205------------------------------------------------------------------------------

206

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

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

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

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

211--

212-- @

213-- import Streamly

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

215--

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

217-- @

218-- @

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

220-- @

221--

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

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

224--

225--

226-- @

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

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

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

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

231-- @

232-- @

233-- (1,3)

234-- (2,3)

235-- (1,4)

236-- (2,4)

237-- @

238--

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

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

241-- stream.

242--

243-- @since 0.2.0

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

245 deriving (MonadTrans)

246

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

248-- documentation for more details.

249--

250-- @since 0.2.0

251type WSerial a = WSerialT IO a

252

253-- |

254-- @since 0.1.0

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

256type InterleavedT = WSerialT

257

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

259--

260-- @since 0.2.0

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

262wSerially = adapt

263

264-- | Same as 'wSerially'.

265--

266-- @since 0.1.0

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

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

269interleaving = wSerially

270

271instance IsStream WSerialT where

272 toStream = getWSerialT

273 fromStream = WSerialT

274

275 {-# INLINE consM #-}

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

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

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

279

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

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

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

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

284

285------------------------------------------------------------------------------

286-- Semigroup

287------------------------------------------------------------------------------

288

289{-# INLINE interleave #-}

290interleave :: Stream m a -> Stream m a -> Stream m a

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

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

293 single a = yld a m2

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

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

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 = fromStream $ Stream $ \st stp sng yld ->

304 unStream (interleave (toStream m1) (toStream m2))

305 (rstState st) stp sng yld

306

307instance Semigroup (WSerialT m a) where

308 (<>) = wSerial

309

310infixr 5 <=>

311

312-- | Same as 'wSerial'.

313--

314-- @since 0.1.0

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

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

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

318(<=>) = wSerial

319

320------------------------------------------------------------------------------

321-- Monoid

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

323

324instance Monoid (WSerialT m a) where

325 mempty = K.nil

326 mappend = (<>)

327

328------------------------------------------------------------------------------

329-- Monad

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

331

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

333 return = pure

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

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

336 single a = run $ toStream (f a)

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

338 in m (rstState st) stp single yieldk

339

340------------------------------------------------------------------------------

341-- Other instances

342------------------------------------------------------------------------------

343

344MONAD_APPLICATIVE_INSTANCE(WSerialT,)

345MONAD_COMMON_INSTANCES(WSerialT,)