src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on 266148c help sign in

streamly/src/Streamly/Streams/Serial.hs

344 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 StandaloneDeriving #-}

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

28 , serially

30 -- * Serial interleaving stream

31 , WSerialT

32 , InterleavedT -- deprecated

33 , WSerial

34 , wSerial

35 , (<=>) -- deprecated

36 , wSerially

37 , interleaving -- deprecated

39 -- * Running Streams

40 , runStreamT -- deprecated

41 , runInterleavedT -- deprecated

42 )

43where

45import Control.Monad (ap)

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

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

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

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

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

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

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

53import Data.Semigroup (Semigroup(..))

55import Streamly.Streams.StreamK hiding (serial)

56import qualified Streamly.Streams.StreamK as C

58#include "Instances.hs"

60------------------------------------------------------------------------------

61-- SerialT

62------------------------------------------------------------------------------

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

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

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

67-- all elements from the second stream.

68--

69-- @

70-- import Streamly

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

72--

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

74-- @

75-- @

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

77-- @

78--

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

80-- element of the stream, serially.

81--

82-- @

83-- main = 'runStream' . 'serially' $ do

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

85-- S.once $ print x

86-- @

87-- @

88-- 1

89-- 2

90-- @

91--

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

93--

94-- @

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

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

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

98-- S.once $ print (x, y)

99-- @

100-- @

101-- (1,3)

102-- (1,4)

103-- (2,3)

104-- (2,4)

105-- @

106--

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

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

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

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

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

112--

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

114-- 'SerialT'.

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

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

117-- time.

118--

119-- @since 0.2.0

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

121 deriving (Semigroup, Monoid, Functor, MonadTrans)

122

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

124-- for more details.

125--

126-- @since 0.2.0

127type Serial a = SerialT IO a

128

129-- |

130-- @since 0.1.0

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

132type StreamT = SerialT

133

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

135--

136-- @since 0.1.0

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

138serially = adapt

139

140instance IsStream SerialT where

141 toStream = getSerialT

142 fromStream = SerialT

143

144 {-# INLINE consM #-}

145 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

146 consM :: Monad m => m a -> SerialT m a -> SerialT m a

147 consM m r = fromStream $ consMSerial m (toStream r)

148

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

150 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}

151 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a

152 m |: r = fromStream $ consMSerial m (toStream r)

153

154------------------------------------------------------------------------------

155-- Semigroup

156------------------------------------------------------------------------------

157

158-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.

159-- Appends two streams sequentially, yielding all elements from the first

160-- stream, and then all elements from the second stream.

161--

162-- @since 0.2.0

163{-# INLINE serial #-}

164serial :: IsStream t => t m a -> t m a -> t m a

165serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)

166

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

168-- Monad

169------------------------------------------------------------------------------

170

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

172 return = pure

173 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp sng yld ->

174 let run x = (unStream x) Nothing stp sng yld

175 single a = run $ toStream (f a)

176 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

177 in m Nothing stp single yieldk

178

179------------------------------------------------------------------------------

180-- Other instances

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

182

183MONAD_APPLICATIVE_INSTANCE(SerialT,)

184MONAD_COMMON_INSTANCES(SerialT,)

185

186------------------------------------------------------------------------------

187-- WSerialT

188------------------------------------------------------------------------------

189

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

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

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

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

194--

195-- @

196-- import Streamly

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

198--

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

200-- @

201-- @

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

203-- @

204--

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

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

207--

208--

209-- @

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

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

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

213-- S.once $ print (x, y)

214-- @

215-- @

216-- (1,3)

217-- (2,3)

218-- (1,4)

219-- (2,4)

220-- @

221--

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

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

224-- stream.

225--

226-- @since 0.2.0

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

228 deriving (Functor, MonadTrans)

229

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

231-- documentation for more details.

232--

233-- @since 0.2.0

234type WSerial a = WSerialT IO a

235

236-- |

237-- @since 0.1.0

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

239type InterleavedT = WSerialT

240

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

242--

243-- @since 0.2.0

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

245wSerially = adapt

246

247-- | Same as 'wSerially'.

248--

249-- @since 0.1.0

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

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

252interleaving = wSerially

253

254instance IsStream WSerialT where

255 toStream = getWSerialT

256 fromStream = WSerialT

257

258 {-# INLINE consM #-}

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

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

261 consM m r = fromStream $ consMSerial m (toStream r)

262

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

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

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

266 m |: r = fromStream $ consMSerial m (toStream r)

267

268------------------------------------------------------------------------------

269-- Semigroup

270------------------------------------------------------------------------------

271

272{-# INLINE interleave #-}

273interleave :: Stream m a -> Stream m a -> Stream m a

274interleave m1 m2 = Stream $ \_ stp sng yld -> do

275 let stop = (unStream m2) Nothing stp sng yld

276 single a = yld a m2

277 yieldk a r = yld a (interleave m2 r)

278 (unStream m1) Nothing stop single yieldk

279

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

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

282--

283-- @since 0.2.0

284{-# INLINE wSerial #-}

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

286wSerial m1 m2 = fromStream $ interleave (toStream m1) (toStream m2)

287

288instance Semigroup (WSerialT m a) where

289 (<>) = wSerial

290

291infixr 5 <=>

292

293-- | Same as 'wSerial'.

294--

295-- @since 0.1.0

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

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

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

299(<=>) = wSerial

300

301------------------------------------------------------------------------------

302-- Monoid

303------------------------------------------------------------------------------

304

305instance Monoid (WSerialT m a) where

306 mempty = nil

307 mappend = (<>)

308

309------------------------------------------------------------------------------

310-- Monad

311------------------------------------------------------------------------------

312

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

314 return = pure

315 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \_ stp sng yld ->

316 let run x = (unStream x) Nothing stp sng yld

317 single a = run $ toStream (f a)

318 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

319 in m Nothing stp single yieldk

320

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

322-- Other instances

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

324

325MONAD_APPLICATIVE_INSTANCE(WSerialT,)

326MONAD_COMMON_INSTANCES(WSerialT,)

327

328-------------------------------------------------------------------------------

329-- Running Streams

330-------------------------------------------------------------------------------

331

332-- | Same as @runStream@.

333--

334-- @since 0.1.0

335{-# DEPRECATED runStreamT "Please use runStream instead." #-}

336runStreamT :: Monad m => SerialT m a -> m ()

337runStreamT = runStream

338

339-- | Same as @runStream . wSerially@.

340--

341-- @since 0.1.0

342{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}

343runInterleavedT :: Monad m => InterleavedT m a -> m ()

344runInterleavedT = runStream