src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 103705f help sign in

streamly/src/Streamly/Streams/Serial.hs

399 lines 12 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 , wSerialFst

36 , wSerialMin

37 , (<=>) -- deprecated

38 , wSerially

39 , interleaving -- deprecated

41 -- * Transformation

42 , map

43 , mapM

44 )

45where

47import Control.Applicative (liftA2)

48import Control.DeepSeq (NFData(..))

49#if MIN_VERSION_deepseq(1,4,3)

50import Control.DeepSeq (NFData1(..))

51#endif

52import Control.Monad (ap)

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

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

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

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

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

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

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

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

61import Data.Foldable (fold)

62#if __GLASGOW_HASKELL__ < 808

63import Data.Semigroup (Semigroup(..))

64#endif

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

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

67 readListPrecDefault)

68import Prelude hiding (map, mapM)

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

71 foldStream)

72import qualified Streamly.Streams.Prelude as P

73import qualified Streamly.Streams.StreamK as K

74import qualified Streamly.Streams.StreamD as D

76#include "Instances.hs"

77#include "inline.hs"

79------------------------------------------------------------------------------

80-- SerialT

81------------------------------------------------------------------------------

83-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append

84-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated

85-- first until it exhausts and then stream @b@ is evaluated. In other words,

86-- the elements of stream @b@ are appended to the elements of stream @a@. This

87-- operation can be used to fold an infinite lazy container of streams.

88--

89-- @

90-- import Streamly

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

92--

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

94-- @

95-- @

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

97-- @

98--

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

100-- element of the stream, serially.

101--

102-- @

103-- main = S.drain . 'serially' $ do

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

105-- S.yieldM $ print x

106-- @

107-- @

108-- 1

109-- 2

110-- @

111--

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

113--

114-- @

115-- main = S.drain . 'serially' $ do

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

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

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

119-- @

120-- @

121-- (1,3)

122-- (1,4)

123-- (2,3)

124-- (2,4)

125-- @

126--

127-- We call the monadic code being run for each element of the stream a monadic

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

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

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

131--

132-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'

133-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'

134-- can contain effectful actions while lists are pure.

135--

136-- In the code above, the 'serially' combinator can be omitted as the default

137-- stream type is 'SerialT'.

138--

139-- @since 0.2.0

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

141 deriving (Semigroup, Monoid, MonadTrans)

142

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

144-- for more details.

145--

146-- @since 0.2.0

147type Serial = SerialT IO

148

149-- |

150-- @since 0.1.0

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

152type StreamT = SerialT

153

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

155--

156-- @since 0.1.0

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

158serially = adapt

159

160{-# INLINE consMSerial #-}

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

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

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

164

165instance IsStream SerialT where

166 toStream = getSerialT

167 fromStream = SerialT

168 consM = consMSerial

169 (|:) = consMSerial

170

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

172-- Monad

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

174

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

176 return = pure

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

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

179

180 -- StreamD based implementation

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

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

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-- |

193-- @

194-- map = fmap

195-- @

196--

197-- Same as 'fmap'.

198--

199-- @

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

201-- [2,3,4]

202-- @

203--

204-- @since 0.4.0

205{-# INLINE map #-}

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

207map f = mapM (return . f)

208

209MONAD_APPLICATIVE_INSTANCE(SerialT,)

210MONAD_COMMON_INSTANCES(SerialT,)

211LIST_INSTANCES(SerialT)

212NFDATA1_INSTANCE(SerialT)

213FOLDABLE_INSTANCE(SerialT)

214TRAVERSABLE_INSTANCE(SerialT)

215

216------------------------------------------------------------------------------

217-- WSerialT

218------------------------------------------------------------------------------

219

220-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the

221-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated

222-- first to produce the first element of the combined stream and then stream

223-- @b@ is evaluated to produce the next element of the combined stream, and

224-- then we go back to evaluating stream @a@ and so on. In other words, the

225-- elements of stream @a@ are interleaved with the elements of stream @b@.

226--

227-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we

228-- interleave them in a binary fashion i.e. @a@ and @b@ are interleaved with

229-- each other and the result is interleaved with @c@ and so on. This will not

230-- act as a true round-robin scheduling across all the streams. Note that this

231-- operation cannot be used to fold a container of infinite streams as the

232-- state that it needs to maintain is proportional to the number of streams.

233--

234-- @

235-- import Streamly

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

237--

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

239-- @

240-- @

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

242-- @

243--

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

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

246--

247--

248-- @

249-- main = S.drain . 'wSerially' $ do

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

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

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

253-- @

254-- @

255-- (1,3)

256-- (2,3)

257-- (1,4)

258-- (2,4)

259-- @

260--

261-- @since 0.2.0

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

263 deriving (MonadTrans)

264

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

266-- documentation for more details.

267--

268-- @since 0.2.0

269type WSerial = WSerialT IO

270

271-- |

272-- @since 0.1.0

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

274type InterleavedT = WSerialT

275

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

277--

278-- @since 0.2.0

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

280wSerially = adapt

281

282-- | Same as 'wSerially'.

283--

284-- @since 0.1.0

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

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

287interleaving = wSerially

288

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

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

291

292instance IsStream WSerialT where

293 toStream = getWSerialT

294 fromStream = WSerialT

295

296 {-# INLINE consM #-}

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

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

299 consM = consMWSerial

300

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

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

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

304 (|:) = consMWSerial

305

306------------------------------------------------------------------------------

307-- Semigroup

308------------------------------------------------------------------------------

309

310-- Additionally we can have m elements yield from the first stream and n

311-- elements yeilding from the second stream. We can also have time slicing

312-- variants of positional interleaving, e.g. run first stream for m seconds and

313-- run the second stream for n seconds.

314--

315-- Similar combinators can be implemented using WAhead style.

316

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

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

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

320-- stream.

321--

322-- @since 0.2.0

323{-# INLINE wSerial #-}

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

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

326 let stop = foldStream st yld sng stp m2

327 single a = yld a m2

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

329 foldStream st yieldk single stop m1

330

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

332--

333-- @since 0.7.0

334{-# INLINE wSerialFst #-}

335wSerialFst :: IsStream t => t m a -> t m a -> t m a

336wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do

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

338 in foldStream st yieldFirst sng stp m1

339

340 where

341

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

343 let stop = foldStream st yld sng stp s1

344 single a = yld a s1

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

346 in foldStream st yieldk single stop s2

347

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

349-- stops.

350--

351-- @since 0.7.0

352{-# INLINE wSerialMin #-}

353wSerialMin :: IsStream t => t m a -> t m a -> t m a

354wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do

355 let stop = stp

356 single a = sng a

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

358 foldStream st yieldk single stop m1

359

360instance Semigroup (WSerialT m a) where

361 (<>) = wSerial

362

363infixr 5 <=>

364

365-- | Same as 'wSerial'.

366--

367-- @since 0.1.0

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

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

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

371(<=>) = wSerial

372

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

374-- Monoid

375------------------------------------------------------------------------------

376

377instance Monoid (WSerialT m a) where

378 mempty = K.nil

379 mappend = (<>)

380

381------------------------------------------------------------------------------

382-- Monad

383------------------------------------------------------------------------------

384

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

386 return = pure

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

388 (>>=) = K.bindWith wSerial

389

390------------------------------------------------------------------------------

391-- Other instances

392------------------------------------------------------------------------------

393

394MONAD_APPLICATIVE_INSTANCE(WSerialT,)

395MONAD_COMMON_INSTANCES(WSerialT,)

396LIST_INSTANCES(WSerialT)

397NFDATA1_INSTANCE(WSerialT)

398FOLDABLE_INSTANCE(WSerialT)

399TRAVERSABLE_INSTANCE(WSerialT)