src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on bcd787e help sign in

streamly/src/Streamly/Streams/Serial.hs

343 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 = 'runStream' . '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 = 'runStream' . '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,)

198

199------------------------------------------------------------------------------

200-- WSerialT

201------------------------------------------------------------------------------

202

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

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

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

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

207--

208-- @

209-- import Streamly

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

211--

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

213-- @

214-- @

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

216-- @

217--

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

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

220--

221--

222-- @

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

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

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

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

227-- @

228-- @

229-- (1,3)

230-- (2,3)

231-- (1,4)

232-- (2,4)

233-- @

234--

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

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

237-- stream.

238--

239-- @since 0.2.0

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

241 deriving (MonadTrans)

242

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

244-- documentation for more details.

245--

246-- @since 0.2.0

247type WSerial = WSerialT IO

248

249-- |

250-- @since 0.1.0

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

252type InterleavedT = WSerialT

253

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

255--

256-- @since 0.2.0

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

258wSerially = adapt

259

260-- | Same as 'wSerially'.

261--

262-- @since 0.1.0

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

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

265interleaving = wSerially

266

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

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

269

270instance IsStream WSerialT where

271 toStream = getWSerialT

272 fromStream = WSerialT

273

274 {-# INLINE consM #-}

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

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

277 consM = consMWSerial

278

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

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

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

282 (|:) = consMWSerial

283

284------------------------------------------------------------------------------

285-- Semigroup

286------------------------------------------------------------------------------

287

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

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

290--

291-- @since 0.2.0

292{-# INLINE wSerial #-}

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

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

295 let stop = foldStream st yld sng stp m2

296 single a = yld a m2

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

298 foldStream st yieldk single stop m1

299

300instance Semigroup (WSerialT m a) where

301 (<>) = wSerial

302

303infixr 5 <=>

304

305-- | Same as 'wSerial'.

306--

307-- @since 0.1.0

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

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

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

311(<=>) = wSerial

312

313------------------------------------------------------------------------------

314-- Monoid

315------------------------------------------------------------------------------

316

317instance Monoid (WSerialT m a) where

318 mempty = K.nil

319 mappend = (<>)

320

321------------------------------------------------------------------------------

322-- Monad

323------------------------------------------------------------------------------

324

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

326 return = pure

327 (>>=) = K.bindWith wSerial

328

329------------------------------------------------------------------------------

330-- Other instances

331------------------------------------------------------------------------------

332

333MONAD_APPLICATIVE_INSTANCE(WSerialT,)

334MONAD_COMMON_INSTANCES(WSerialT,)

335

336LIST_INSTANCES(SerialT)

337LIST_INSTANCES(WSerialT)

338

339FOLDABLE_INSTANCE(SerialT)

340FOLDABLE_INSTANCE(WSerialT)

341

342TRAVERSABLE_INSTANCE(SerialT)

343TRAVERSABLE_INSTANCE(WSerialT)