src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 6bde698 help sign in

streamly/src/Streamly/Streams/Serial.hs

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

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)

62import Data.Semigroup (Semigroup(..))

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

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

65 readListPrecDefault)

66import Prelude hiding (map, mapM)

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

69 foldStream)

70import qualified Streamly.Streams.Prelude as P

71import qualified Streamly.Streams.StreamK as K

72import qualified Streamly.Streams.StreamD as D

74#include "Instances.hs"

75#include "inline.hs"

77------------------------------------------------------------------------------

78-- SerialT

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

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

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

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

84-- all elements from the second stream.

85--

86-- @

87-- import Streamly

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

89--

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

91-- @

92-- @

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

94-- @

95--

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

97-- element of the stream, serially.

98--

99-- @

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

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

102-- S.yieldM $ print x

103-- @

104-- @

105-- 1

106-- 2

107-- @

108--

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

110--

111-- @

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

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

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

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

116-- @

117-- @

118-- (1,3)

119-- (1,4)

120-- (2,3)

121-- (2,4)

122-- @

123--

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

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

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

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

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

129--

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

131-- 'SerialT'.

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

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

134-- time.

135--

136-- @since 0.2.0

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

138 deriving (Semigroup, Monoid, MonadTrans)

139

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

141-- for more details.

142--

143-- @since 0.2.0

144type Serial = SerialT IO

145

146-- |

147-- @since 0.1.0

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

149type StreamT = SerialT

150

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

152--

153-- @since 0.1.0

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

155serially = adapt

156

157{-# INLINE consMSerial #-}

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

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

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

161

162instance IsStream SerialT where

163 toStream = getSerialT

164 fromStream = SerialT

165 consM = consMSerial

166 (|:) = consMSerial

167

168------------------------------------------------------------------------------

169-- Monad

170------------------------------------------------------------------------------

171

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

173 return = pure

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

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

176

177 -- StreamD based implementation

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

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

180

181------------------------------------------------------------------------------

182-- Other instances

183------------------------------------------------------------------------------

184

185{-# INLINE_EARLY mapM #-}

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

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

188

189-- |

190-- @

191-- map = fmap

192-- @

193--

194-- Same as 'fmap'.

195--

196-- @

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

198-- [2,3,4]

199-- @

200--

201-- @since 0.4.0

202{-# INLINE map #-}

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

204map f = mapM (return . f)

205

206MONAD_APPLICATIVE_INSTANCE(SerialT,)

207MONAD_COMMON_INSTANCES(SerialT,)

208LIST_INSTANCES(SerialT)

209NFDATA1_INSTANCE(SerialT)

210FOLDABLE_INSTANCE(SerialT)

211TRAVERSABLE_INSTANCE(SerialT)

212

213------------------------------------------------------------------------------

214-- WSerialT

215------------------------------------------------------------------------------

216

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

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

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

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

221--

222-- @

223-- import Streamly

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

225--

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

227-- @

228-- @

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

230-- @

231--

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

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

234--

235--

236-- @

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

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

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

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

241-- @

242-- @

243-- (1,3)

244-- (2,3)

245-- (1,4)

246-- (2,4)

247-- @

248--

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

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

251-- stream.

252--

253-- @since 0.2.0

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

255 deriving (MonadTrans)

256

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

258-- documentation for more details.

259--

260-- @since 0.2.0

261type WSerial = WSerialT IO

262

263-- |

264-- @since 0.1.0

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

266type InterleavedT = WSerialT

267

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

269--

270-- @since 0.2.0

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

272wSerially = adapt

273

274-- | Same as 'wSerially'.

275--

276-- @since 0.1.0

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

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

279interleaving = wSerially

280

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

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

283

284instance IsStream WSerialT where

285 toStream = getWSerialT

286 fromStream = WSerialT

287

288 {-# INLINE consM #-}

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

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

291 consM = consMWSerial

292

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

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

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

296 (|:) = consMWSerial

297

298------------------------------------------------------------------------------

299-- Semigroup

300------------------------------------------------------------------------------

301

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

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

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

305-- stream.

306--

307-- @since 0.2.0

308{-# INLINE wSerial #-}

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

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

311 let stop = foldStream st yld sng stp m2

312 single a = yld a m2

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

314 foldStream st yieldk single stop m1

315

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

317--

318-- @since 0.7.0

319{-# INLINE wSerialEndByFirst #-}

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

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

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

323 in foldStream st yieldFirst sng stp m1

324

325 where

326

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

328 let stop = foldStream st yld sng stp s1

329 single a = yld a s1

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

331 in foldStream st yieldk single stop s2

332

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

334-- stops.

335--

336-- @since 0.7.0

337{-# INLINE wSerialEndByAny #-}

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

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

340 let stop = stp

341 single a = sng a

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

343 foldStream st yieldk single stop m1

344

345instance Semigroup (WSerialT m a) where

346 (<>) = wSerial

347

348infixr 5 <=>

349

350-- | Same as 'wSerial'.

351--

352-- @since 0.1.0

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

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

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

356(<=>) = wSerial

357

358------------------------------------------------------------------------------

359-- Monoid

360------------------------------------------------------------------------------

361

362instance Monoid (WSerialT m a) where

363 mempty = K.nil

364 mappend = (<>)

365

366------------------------------------------------------------------------------

367-- Monad

368------------------------------------------------------------------------------

369

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

371 return = pure

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

373 (>>=) = K.bindWith wSerial

374

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

376-- Other instances

377------------------------------------------------------------------------------

378

379MONAD_APPLICATIVE_INSTANCE(WSerialT,)

380MONAD_COMMON_INSTANCES(WSerialT,)

381LIST_INSTANCES(WSerialT)

382NFDATA1_INSTANCE(WSerialT)

383FOLDABLE_INSTANCE(WSerialT)

384TRAVERSABLE_INSTANCE(WSerialT)