src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 364af08 help sign in

streamly/src/Streamly/Streams/Serial.hs

379 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 , wSerialEndByFirst

36 , wSerialEndByAny

37 , (<=>) -- deprecated

38 , wSerially

39 , interleaving -- deprecated

41 -- * Transformation

42 , map

43 , mapM

44 )

45where

47import Control.Applicative (liftA2)

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

49import Control.Monad (ap)

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

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

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

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

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

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

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

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

58import Data.Foldable (fold)

59import Data.Semigroup (Semigroup(..))

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

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

62 readListPrecDefault)

63import Prelude hiding (map, mapM)

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

66 foldStream)

67import qualified Streamly.Streams.Prelude as P

68import qualified Streamly.Streams.StreamK as K

69import qualified Streamly.Streams.StreamD as D

71#include "Instances.hs"

72#include "inline.hs"

74------------------------------------------------------------------------------

75-- SerialT

76------------------------------------------------------------------------------

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

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

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

81-- all elements from the second stream.

82--

83-- @

84-- import Streamly

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

86--

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

88-- @

89-- @

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

91-- @

92--

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

94-- element of the stream, serially.

95--

96-- @

97-- main = 'drain' . 'serially' $ do

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

99-- S.yieldM $ print x

100-- @

101-- @

102-- 1

103-- 2

104-- @

105--

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

107--

108-- @

109-- main = 'drain' . 'serially' $ do

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

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

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

113-- @

114-- @

115-- (1,3)

116-- (1,4)

117-- (2,3)

118-- (2,4)

119-- @

120--

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

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

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

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

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

126--

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

128-- 'SerialT'.

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

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

131-- time.

132--

133-- @since 0.2.0

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

135 deriving (Semigroup, Monoid, MonadTrans)

136

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

138-- for more details.

139--

140-- @since 0.2.0

141type Serial = SerialT IO

142

143-- |

144-- @since 0.1.0

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

146type StreamT = SerialT

147

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

149--

150-- @since 0.1.0

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

152serially = adapt

153

154{-# INLINE consMSerial #-}

155{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}

156consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a

157consMSerial m ms = fromStream $ K.consMStream m (toStream ms)

158

159instance IsStream SerialT where

160 toStream = getSerialT

161 fromStream = SerialT

162 consM = consMSerial

163 (|:) = consMSerial

164

165------------------------------------------------------------------------------

166-- Monad

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

168

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

170 return = pure

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

172 (>>=) = K.bindWith K.serial

173

174 -- StreamD based implementation

175 -- return = SerialT . D.fromStreamD . D.yield

176 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)

177

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

179-- Other instances

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

181

182{-# INLINE_EARLY mapM #-}

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

184mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream 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 :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

201map f = mapM (return . f)

202

203MONAD_APPLICATIVE_INSTANCE(SerialT,)

204MONAD_COMMON_INSTANCES(SerialT,)

205LIST_INSTANCES(SerialT)

206FOLDABLE_INSTANCE(SerialT)

207TRAVERSABLE_INSTANCE(SerialT)

208

209------------------------------------------------------------------------------

210-- WSerialT

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

212

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

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

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

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

217--

218-- @

219-- import Streamly

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

221--

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

223-- @

224-- @

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

226-- @

227--

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

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

230--

231--

232-- @

233-- main = 'drain' . 'wSerially' $ do

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

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

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

237-- @

238-- @

239-- (1,3)

240-- (2,3)

241-- (1,4)

242-- (2,4)

243-- @

244--

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

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

247-- stream.

248--

249-- @since 0.2.0

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

251 deriving (MonadTrans)

252

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

254-- documentation for more details.

255--

256-- @since 0.2.0

257type WSerial = WSerialT IO

258

259-- |

260-- @since 0.1.0

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

262type InterleavedT = WSerialT

263

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

265--

266-- @since 0.2.0

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

268wSerially = adapt

269

270-- | Same as 'wSerially'.

271--

272-- @since 0.1.0

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

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

275interleaving = wSerially

276

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

278consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)

279

280instance IsStream WSerialT where

281 toStream = getWSerialT

282 fromStream = WSerialT

283

284 {-# INLINE consM #-}

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

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

287 consM = consMWSerial

288

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

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

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

292 (|:) = consMWSerial

293

294------------------------------------------------------------------------------

295-- Semigroup

296------------------------------------------------------------------------------

297

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

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

300-- When one stream stops the rest of the other stream is used in the output

301-- stream.

302--

303-- @since 0.2.0

304{-# INLINE wSerial #-}

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

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

307 let stop = foldStream st yld sng stp m2

308 single a = yld a m2

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

310 foldStream st yieldk single stop m1

311

312-- | Like `wSerial` but stops interleaving as soon as the first stream stops.

313--

314-- @since 0.7.0

315{-# INLINE wSerialEndByFirst #-}

316wSerialEndByFirst :: IsStream t => t m a -> t m a -> t m a

317wSerialEndByFirst m1 m2 = mkStream $ \st yld sng stp -> do

318 let yieldFirst a r = yld a (yieldSecond r m2)

319 in foldStream st yieldFirst sng stp m1

320

321 where

322

323 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do

324 let stop = foldStream st yld sng stp s1

325 single a = yld a s1

326 yieldk a r = yld a (wSerial s1 r)

327 in foldStream st yieldk single stop s2

328

329-- | Like `wSerial` but stops interleaving as soon as any of the two streams

330-- stops.

331--

332-- @since 0.7.0

333{-# INLINE wSerialEndByAny #-}

334wSerialEndByAny :: IsStream t => t m a -> t m a -> t m a

335wSerialEndByAny m1 m2 = mkStream $ \st yld sng stp -> do

336 let stop = stp

337 single a = sng a

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

339 foldStream st yieldk single stop m1

340

341instance Semigroup (WSerialT m a) where

342 (<>) = wSerial

343

344infixr 5 <=>

345

346-- | Same as 'wSerial'.

347--

348-- @since 0.1.0

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

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

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

352(<=>) = wSerial

353

354------------------------------------------------------------------------------

355-- Monoid

356------------------------------------------------------------------------------

357

358instance Monoid (WSerialT m a) where

359 mempty = K.nil

360 mappend = (<>)

361

362------------------------------------------------------------------------------

363-- Monad

364------------------------------------------------------------------------------

365

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

367 return = pure

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

369 (>>=) = K.bindWith wSerial

370

371------------------------------------------------------------------------------

372-- Other instances

373------------------------------------------------------------------------------

374

375MONAD_APPLICATIVE_INSTANCE(WSerialT,)

376MONAD_COMMON_INSTANCES(WSerialT,)

377LIST_INSTANCES(WSerialT)

378FOLDABLE_INSTANCE(WSerialT)

379TRAVERSABLE_INSTANCE(WSerialT)