src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 9268208 help sign in

streamly/src/Streamly/Streams/Serial.hs

340 lines 9.9 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 , (<=>) -- deprecated

36 , wSerially

37 , interleaving -- deprecated

39 -- * Transformation

40 , map

41 , mapM

42 )

43where

45import Control.Applicative (liftA2)

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

47import Control.Monad (ap)

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

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

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

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

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

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

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

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

56import Data.Foldable (fold)

57import Data.Semigroup (Semigroup(..))

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

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

60 readListPrecDefault)

61import Prelude hiding (map, mapM)

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

64 foldStream)

65import qualified Streamly.Streams.Prelude as P

66import qualified Streamly.Streams.StreamK as K

67import qualified Streamly.Streams.StreamD as D

69#include "Instances.hs"

70#include "inline.hs"

72------------------------------------------------------------------------------

73-- SerialT

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

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

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

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

79-- all elements from the second stream.

80--

81-- @

82-- import Streamly

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

84--

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

86-- @

87-- @

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

89-- @

90--

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

92-- element of the stream, serially.

93--

94-- @

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

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

97-- S.yieldM $ print x

98-- @

99-- @

100-- 1

101-- 2

102-- @

103--

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

105--

106-- @

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

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

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

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

111-- @

112-- @

113-- (1,3)

114-- (1,4)

115-- (2,3)

116-- (2,4)

117-- @

118--

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

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

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

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

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

124--

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

126-- 'SerialT'.

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

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

129-- time.

130--

131-- @since 0.2.0

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

133 deriving (Semigroup, Monoid, MonadTrans)

134

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

136-- for more details.

137--

138-- @since 0.2.0

139type Serial = SerialT IO

140

141-- |

142-- @since 0.1.0

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

144type StreamT = SerialT

145

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

147--

148-- @since 0.1.0

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

150serially = adapt

151

152{-# INLINE consMSerial #-}

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

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

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

156

157instance IsStream SerialT where

158 toStream = getSerialT

159 fromStream = SerialT

160 consM = consMSerial

161 (|:) = consMSerial

162

163------------------------------------------------------------------------------

164-- Monad

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

166

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

168 return = pure

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

170

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

172-- Other instances

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

174

175{-# INLINE_EARLY mapM #-}

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

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

178

179-- |

180-- @

181-- map = fmap

182-- @

183--

184-- Same as 'fmap'.

185--

186-- @

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

188-- [2,3,4]

189-- @

190--

191-- @since 0.4.0

192{-# INLINE map #-}

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

194map f = mapM (return . f)

195

196MONAD_APPLICATIVE_INSTANCE(SerialT,)

197MONAD_COMMON_INSTANCES(SerialT,)

198LIST_INSTANCES(SerialT)

199FOLDABLE_INSTANCE(SerialT)

200TRAVERSABLE_INSTANCE(SerialT)

201

202------------------------------------------------------------------------------

203-- WSerialT

204------------------------------------------------------------------------------

205

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

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

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

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

210--

211-- @

212-- import Streamly

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

214--

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

216-- @

217-- @

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

219-- @

220--

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

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

223--

224--

225-- @

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

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

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

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

230-- @

231-- @

232-- (1,3)

233-- (2,3)

234-- (1,4)

235-- (2,4)

236-- @

237--

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

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

240-- stream.

241--

242-- @since 0.2.0

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

244 deriving (MonadTrans)

245

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

247-- documentation for more details.

248--

249-- @since 0.2.0

250type WSerial = WSerialT IO

251

252-- |

253-- @since 0.1.0

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

255type InterleavedT = WSerialT

256

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

258--

259-- @since 0.2.0

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

261wSerially = adapt

262

263-- | Same as 'wSerially'.

264--

265-- @since 0.1.0

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

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

268interleaving = wSerially

269

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

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

272

273instance IsStream WSerialT where

274 toStream = getWSerialT

275 fromStream = WSerialT

276

277 {-# INLINE consM #-}

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

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

280 consM = consMWSerial

281

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

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

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

285 (|:) = consMWSerial

286

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

288-- Semigroup

289------------------------------------------------------------------------------

290

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

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

293--

294-- @since 0.2.0

295{-# INLINE wSerial #-}

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

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

298 let stop = foldStream st yld sng stp m2

299 single a = yld a m2

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

301 foldStream st yieldk single stop m1

302

303instance Semigroup (WSerialT m a) where

304 (<>) = wSerial

305

306infixr 5 <=>

307

308-- | Same as 'wSerial'.

309--

310-- @since 0.1.0

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

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

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

314(<=>) = wSerial

315

316------------------------------------------------------------------------------

317-- Monoid

318------------------------------------------------------------------------------

319

320instance Monoid (WSerialT m a) where

321 mempty = K.nil

322 mappend = (<>)

323

324------------------------------------------------------------------------------

325-- Monad

326------------------------------------------------------------------------------

327

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

329 return = pure

330 (>>=) = K.bindWith wSerial

331

332------------------------------------------------------------------------------

333-- Other instances

334------------------------------------------------------------------------------

335

336MONAD_APPLICATIVE_INSTANCE(WSerialT,)

337MONAD_COMMON_INSTANCES(WSerialT,)

338LIST_INSTANCES(WSerialT)

339FOLDABLE_INSTANCE(WSerialT)

340TRAVERSABLE_INSTANCE(WSerialT)