src/Streamly/Streams/Serial.hs · streamly

git.iko.soy/streamly

on c042050 help sign in

streamly/src/Streamly/Streams/Serial.hs

349 lines 10 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

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

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

154

155instance IsStream SerialT where

156 toStream = getSerialT

157 fromStream = SerialT

158

159 {-# INLINE consM #-}

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

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

162 consM = consMSerial

163

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

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

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

167 (|:) = consMSerial

168

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

170-- Monad

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

172

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

174 return = pure

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

176

177------------------------------------------------------------------------------

178-- Other instances

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

180

181{-# INLINE_EARLY mapM #-}

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

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

184

185-- |

186-- @

187-- map = fmap

188-- @

189--

190-- Same as 'fmap'.

191--

192-- @

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

194-- [2,3,4]

195-- @

196--

197-- @since 0.4.0

198{-# INLINE map #-}

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

200map f = mapM (return . f)

201

202MONAD_APPLICATIVE_INSTANCE(SerialT,)

203MONAD_COMMON_INSTANCES(SerialT,)

204

205------------------------------------------------------------------------------

206-- WSerialT

207------------------------------------------------------------------------------

208

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

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

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

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

213--

214-- @

215-- import Streamly

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

217--

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

219-- @

220-- @

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

222-- @

223--

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

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

226--

227--

228-- @

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

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

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

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

233-- @

234-- @

235-- (1,3)

236-- (2,3)

237-- (1,4)

238-- (2,4)

239-- @

240--

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

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

243-- stream.

244--

245-- @since 0.2.0

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

247 deriving (MonadTrans)

248

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

250-- documentation for more details.

251--

252-- @since 0.2.0

253type WSerial = WSerialT IO

254

255-- |

256-- @since 0.1.0

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

258type InterleavedT = WSerialT

259

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

261--

262-- @since 0.2.0

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

264wSerially = adapt

265

266-- | Same as 'wSerially'.

267--

268-- @since 0.1.0

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

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

271interleaving = wSerially

272

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

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

275

276instance IsStream WSerialT where

277 toStream = getWSerialT

278 fromStream = WSerialT

279

280 {-# INLINE consM #-}

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

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

283 consM = consMWSerial

284

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

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

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

288 (|:) = consMWSerial

289

290------------------------------------------------------------------------------

291-- Semigroup

292------------------------------------------------------------------------------

293

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

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

296--

297-- @since 0.2.0

298{-# INLINE wSerial #-}

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

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

301 let stop = foldStream st yld sng stp m2

302 single a = yld a m2

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

304 foldStream st yieldk single stop m1

305

306instance Semigroup (WSerialT m a) where

307 (<>) = wSerial

308

309infixr 5 <=>

310

311-- | Same as 'wSerial'.

312--

313-- @since 0.1.0

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

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

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

317(<=>) = wSerial

318

319------------------------------------------------------------------------------

320-- Monoid

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

322

323instance Monoid (WSerialT m a) where

324 mempty = K.nil

325 mappend = (<>)

326

327------------------------------------------------------------------------------

328-- Monad

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

330

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

332 return = pure

333 (>>=) = K.bindWith wSerial

334

335------------------------------------------------------------------------------

336-- Other instances

337------------------------------------------------------------------------------

338

339MONAD_APPLICATIVE_INSTANCE(WSerialT,)

340MONAD_COMMON_INSTANCES(WSerialT,)

341

342LIST_INSTANCES(SerialT)

343LIST_INSTANCES(WSerialT)

344

345FOLDABLE_INSTANCE(SerialT)

346FOLDABLE_INSTANCE(WSerialT)

347

348TRAVERSABLE_INSTANCE(SerialT)

349TRAVERSABLE_INSTANCE(WSerialT)