183325e · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

factor out the StreamK type in a separate file

183325eHarendra Kumar8 years ago

src/Streamly/Streams/StreamDType.hs+3−3

compare with master

unifiedsplit

@@ -60,12 +60,12 @@ instance Functor (Step s) where

60data Stream m a =

61 forall s. UnStream (State K.Stream m a -> s -> m (Step s a)) s

63−detachStream :: Stream m a -> Stream m a

63+unShare :: Stream m a -> Stream m a

64−detachStream (UnStream step state) = UnStream step' state

64+unShare (UnStream step state) = UnStream step' state

65 where step' gst = step (adaptState gst)

67pattern Stream :: (State K.Stream m a -> s -> m (Step s a)) -> s -> Stream m a

68−pattern Stream step state <- (detachStream -> UnStream step state)

68+pattern Stream step state <- (unShare -> UnStream step state)

69 where Stream = UnStream

71#if __GLASGOW_HASKELL__ >= 802

src/Streamly/Streams/StreamK.hs+33−285

@@ -161,10 +161,7 @@ module Streamly.Streams.StreamK

161where

162

163import Control.Monad (void)

164−import Control.Monad.IO.Class (MonadIO(liftIO))

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

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

166−import Control.Monad.Trans.Class (MonadTrans(lift))

167−import Data.Semigroup (Semigroup(..))

168import Prelude

165import Prelude

169 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

166 hiding (foldl, foldr, last, map, mapM, mapM_, repeat, sequence,

170 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

167 take, filter, all, any, takeWhile, drop, dropWhile, minimum,

@@ -173,180 +170,18 @@ import Prelude

173import qualified Prelude

170import qualified Prelude

174

171

175import Streamly.SVar

172import Streamly.SVar

176−

173+import Streamly.Streams.StreamKType

177−------------------------------------------------------------------------------

178−-- The basic stream type

179−------------------------------------------------------------------------------

180−

181−-- | The type @Stream m a@ represents a monadic stream of values of type 'a'

182−-- constructed using actions in monad 'm'. It uses stop, singleton and yield

183−-- continuations equivalent to the following direct style type:

184−--

185−-- @

186−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

187−-- @

188−--

189−-- To facilitate parallel composition we maintain a local state in an 'SVar'

190−-- that is shared across and is used for synchronization of the streams being

191−-- composed.

192−--

193−-- The singleton case can be expressed in terms of stop and yield but we have

194−-- it as a separate case to optimize composition operations for streams with

195−-- single element. We build singleton streams in the implementation of 'pure'

196−-- for Applicative and Monad, and in 'lift' for MonadTrans.

197−--

198−newtype Stream m a =

199− Stream (forall r.

200− State Stream m a -- state

201− -> m r -- stop

202− -> (a -> m r) -- singleton

203− -> (a -> Stream m a -> m r) -- yield

204− -> m r

205− )

206−

207−-- Run a stream in detached mode i.e. not joining an SVar

208−{-# INLINE unStream #-}

209−unStream ::

210− Stream m a

211− -> State Stream m b -- state

212− -> m r -- stop

213− -> (a -> m r) -- singleton

214− -> (a -> Stream m a -> m r) -- yield

215− -> m r

216−unStream (Stream runner) st = runner (adaptState st)

217

174

218-- | Detach a stream from an SVar

175-- | Detach a stream from an SVar

219{-# INLINE unShare #-}

176{-# INLINE unShare #-}

220unShare :: Stream m a -> Stream m a

177unShare :: Stream m a -> Stream m a

221−unShare x = Stream $ \st stp sng yld ->

178+unShare x = mkStream $ \st stp sng yld ->

222 unStream x st stp sng yld

179 unStream x st stp sng yld

223

180

224−-- | Like unstream, but passes a shared SVar across continuations.

225−{-# INLINE unStreamShared #-}

226−unStreamShared ::

227− Stream m a

228− -> State Stream m a -- state

229− -> m r -- stop

230− -> (a -> m r) -- singleton

231− -> (a -> Stream m a -> m r) -- yield

232− -> m r

233−unStreamShared (Stream runner) st stp sng yld = runner st stp sng yld

234−

235−-- Run the stream using a run function associated with the SVar that runs the

236−-- streams with a captured snapshot of the monadic state.

237−{-# INLINE runStreamSVar #-}

238−runStreamSVar

239− :: MonadIO m

240− => SVar Stream m a

241− -> Stream m a

242− -> State Stream m a -- state

243− -> m r -- stop

244− -> (a -> m r) -- singleton

245− -> (a -> Stream m a -> m r) -- yield

246− -> m ()

247−runStreamSVar sv m st stp sng yld =

248− let mrun = runInIO $ svarMrun sv

249− in void $ liftIO $ mrun $ unStreamShared m st stp sng yld

250−

251−------------------------------------------------------------------------------

252−-- Types that can behave as a Stream

253−------------------------------------------------------------------------------

254−

255−infixr 5 `consM`

256−infixr 5 |:

257−

258−-- | Class of types that can represent a stream of elements of some type 'a' in

259−-- some monad 'm'.

260−--

261−-- @since 0.2.0

262−class IsStream t where

263− toStream :: t m a -> Stream m a

264− fromStream :: Stream m a -> t m a

265− -- | Constructs a stream by adding a monadic action at the head of an

266− -- existing stream. For example:

267− --

268− -- @

269− -- > toList $ getLine \`consM` getLine \`consM` nil

270− -- hello

271− -- world

272− -- ["hello","world"]

273− -- @

274− --

275− -- /Concurrent (do not use 'parallely' to construct infinite streams)/

276− --

277− -- @since 0.2.0

278− consM :: MonadAsync m => m a -> t m a -> t m a

279− -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"

280− -- to remember that @|@ comes before ':'.

281− --

282− -- @

283− -- > toList $ getLine |: getLine |: nil

284− -- hello

285− -- world

286− -- ["hello","world"]

287− -- @

288− --

289− -- @

290− -- let delay = threadDelay 1000000 >> print 1

291− -- runStream $ serially $ delay |: delay |: delay |: nil

292− -- runStream $ parallely $ delay |: delay |: delay |: nil

293− -- @

294− --

295− -- /Concurrent (do not use 'parallely' to construct infinite streams)/

296− --

297− -- @since 0.2.0

298− (|:) :: MonadAsync m => m a -> t m a -> t m a

299− -- We can define (|:) just as 'consM' but it is defined explicitly for each

300− -- type because we want to use SPECIALIZE pragma on the definition.

301−

302−-- | Same as 'IsStream'.

303−--

304−-- @since 0.1.0

305−{-# DEPRECATED Streaming "Please use IsStream instead." #-}

306−type Streaming = IsStream

307−

308−-------------------------------------------------------------------------------

309−-- Type adapting combinators

310−-------------------------------------------------------------------------------

311−

312−-- | Adapt any specific stream type to any other specific stream type.

313−--

314−-- @since 0.1.0

315−adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

316−adapt = fromStream . toStream

317−

318−------------------------------------------------------------------------------

319−-- Building a stream

320−------------------------------------------------------------------------------

321−

322−-- | Build a stream from an 'SVar', a stop continuation, a singleton stream

323−-- continuation and a yield continuation.

324−mkStream:: IsStream t

325− => (forall r. State Stream m a

326− -> m r

327− -> (a -> m r)

328− -> (a -> t m a -> m r)

329− -> m r)

330− -> t m a

331−mkStream k = fromStream $ Stream $ \st stp sng yld ->

332− let yieldk a r = yld a (toStream r)

333− in k st stp sng yieldk

334−

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

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

336-- Construction

182-- Construction

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

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

338

184

339−-- | An empty stream.

340−--

341−-- @

342−-- > toList nil

343−-- []

344−-- @

345−--

346−-- @since 0.1.0

347−nil :: IsStream t => t m a

348−nil = fromStream $ Stream $ \_ stp _ _ -> stp

349−

350infixr 5 `cons`

185infixr 5 `cons`

351

186

352-- faster than consM because there is no bind.

187-- faster than consM because there is no bind.

@@ -362,7 +197,7 @@ infixr 5 `cons`

362--

197--

363-- @since 0.1.0

198-- @since 0.1.0

364cons :: IsStream t => a -> t m a -> t m a

199cons :: IsStream t => a -> t m a -> t m a

365−cons a r = fromStream $ Stream $ \_ _ _ yld -> yld a (toStream r)

200+cons a r = fromStream $ mkStream $ \_ _ _ yld -> yld a (toStream r)

366

201

367infixr 5 .:

202infixr 5 .:

368

203

@@ -377,47 +212,25 @@ infixr 5 .:

377(.:) :: IsStream t => a -> t m a -> t m a

212(.:) :: IsStream t => a -> t m a -> t m a

378(.:) = cons

213(.:) = cons

379

214

380−{-# INLINE consMSerial #-}

381−consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

382−consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r

383−

384------------------------------------------------------------------------------

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

385-- Asynchronous construction

216-- Asynchronous construction

386------------------------------------------------------------------------------

217------------------------------------------------------------------------------

387

218

388-- | Make an empty stream from a callback function.

219-- | Make an empty stream from a callback function.

389nilK :: IsStream t => (forall r. m r -> m r) -> t m a

220nilK :: IsStream t => (forall r. m r -> m r) -> t m a

390−nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp

221+nilK k = fromStream $ mkStream $ \_ stp _ _ -> k stp

391

222

392-- | Make a singleton stream from a one shot callback function.

223-- | Make a singleton stream from a one shot callback function.

393yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

224yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

394−yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng

225+yieldK k = fromStream $ mkStream $ \_ _ sng _ -> k sng

395

226

396-- | Construct a stream from a callback function.

227-- | Construct a stream from a callback function.

397consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a

228consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a

398−consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))

229+consK k r = fromStream $ mkStream $ \_ _ _ yld -> k (\x -> yld x (toStream r))

399

230

400-- XXX consK with concurrent callbacks

231-- XXX consK with concurrent callbacks

401-- XXX Build a stream from a repeating callback function.

232-- XXX Build a stream from a repeating callback function.

402

233

403−-------------------------------------------------------------------------------

404−-- IsStream Stream

405−-------------------------------------------------------------------------------

406−

407−instance IsStream Stream where

408− toStream = id

409− fromStream = id

410−

411− {-# INLINE consM #-}

412− {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}

413− consM :: Monad m => m a -> Stream m a -> Stream m a

414− consM = consMSerial

415−

416− {-# INLINE (|:) #-}

417− {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}

418− (|:) :: Monad m => m a -> Stream m a -> Stream m a

419− (|:) = consMSerial

420−

421-------------------------------------------------------------------------------

234-------------------------------------------------------------------------------

422-- Deconstruction

235-- Deconstruction

423-------------------------------------------------------------------------------

236-------------------------------------------------------------------------------

@@ -438,7 +251,7 @@ uncons m =

438unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

251unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

439unfoldr step = fromStream . go

252unfoldr step = fromStream . go

440 where

253 where

441− go s = Stream $ \_ stp _ yld ->

254+ go s = mkStream $ \_ stp _ yld ->

442 case step s of

255 case step s of

443 Nothing -> stp

256 Nothing -> stp

444 Just (a, b) -> yld a (go b)

257 Just (a, b) -> yld a (go b)

@@ -447,7 +260,7 @@ unfoldr step = fromStream . go

447unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

260unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

448unfoldrM step = go

261unfoldrM step = go

449 where

262 where

450− go s = fromStream $ Stream $ \svr stp sng yld -> do

263+ go s = fromStream $ mkStream $ \svr stp sng yld -> do

451 mayb <- step s

264 mayb <- step s

452 case mayb of

265 case mayb of

453 Nothing -> stp

266 Nothing -> stp

@@ -459,11 +272,7 @@ unfoldrM step = go

459-------------------------------------------------------------------------------

272-------------------------------------------------------------------------------

460

273

461yield :: IsStream t => a -> t m a

274yield :: IsStream t => a -> t m a

462−yield a = fromStream $ Stream $ \_ _ single _ -> single a

275+yield a = fromStream $ mkStream $ \_ _ single _ -> single a

463−

464−{-# INLINE yieldM #-}

465−yieldM :: (Monad m, IsStream t) => m a -> t m a

466−yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single

467

276

468-- | Same as yieldM

277-- | Same as yieldM

469--

278--

@@ -523,20 +332,6 @@ fromStreamK = id

523-- Elimination by Folding

332-- Elimination by Folding

524-------------------------------------------------------------------------------

333-------------------------------------------------------------------------------

525

334

526−-- | Fold a stream by providing an SVar, a stop continuation, a singleton

527−-- continuation and a yield continuation.

528−foldStream

529− :: IsStream t

530− => State Stream m a

531− -> m r

532− -> (a -> m r)

533− -> (a -> t m a -> m r)

534− -> t m a

535− -> m r

536−foldStream st blank single step m =

537− let yieldk a x = step a (fromStream x)

538− in unStreamShared (toStream m) st blank single yieldk

539−

540-- | Lazy right associative fold.

335-- | Lazy right associative fold.

541foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

336foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

542foldr step acc m = go (toStream m)

337foldr step acc m = go (toStream m)

@@ -594,7 +389,7 @@ foldx step begin done m = get $ go (toStream m) begin

594 -- however that is more expensive because of unnecessary recursion

389 -- however that is more expensive because of unnecessary recursion

595 -- that cannot be tail call optimized. Unfolding recursion explicitly via

390 -- that cannot be tail call optimized. Unfolding recursion explicitly via

596 -- continuations is much more efficient.

391 -- continuations is much more efficient.

597− go m1 !acc = Stream $ \_ _ sng yld ->

392+ go m1 !acc = mkStream $ \_ _ sng yld ->

598 let stop = sng acc

393 let stop = sng acc

599 single a = sng $ step acc a

394 single a = sng $ step acc a

600 yieldk a r =

395 yieldk a r =

@@ -670,7 +465,7 @@ init m = go1 (toStream m)

670 case r of

465 case r of

671 Nothing -> return Nothing

466 Nothing -> return Nothing

672 Just (h, t) -> return . Just . fromStream $ go h t

467 Just (h, t) -> return . Just . fromStream $ go h t

673− go p m1 = Stream $ \_ stp sng yld ->

468+ go p m1 = mkStream $ \_ stp sng yld ->

674 let single _ = sng p

469 let single _ = sng p

675 yieldk a x = yld p $ go a x

470 yieldk a x = yld p $ go a x

676 in unStream m1 defState stp single yieldk

471 in unStream m1 defState stp single yieldk

@@ -850,7 +645,7 @@ find p = findM (return . p)

850findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

645findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

851findIndices p = fromStream . go 0 . toStream

646findIndices p = fromStream . go 0 . toStream

852 where

647 where

853− go offset m1 = Stream $ \st stp sng yld ->

648+ go offset m1 = mkStream $ \st stp sng yld ->

854 let single a | p a = sng offset

649 let single a | p a = sng offset

855 | otherwise = stp

650 | otherwise = stp

856 yieldk a x | p a = yld offset $ go (offset + 1) x

651 yieldk a x | p a = yld offset $ go (offset + 1) x

@@ -893,7 +688,7 @@ scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

893scanx step begin done m =

688scanx step begin done m =

894 cons (done begin) $ fromStream $ go (toStream m) begin

689 cons (done begin) $ fromStream $ go (toStream m) begin

895 where

690 where

896− go m1 !acc = Stream $ \st stp sng yld ->

691+ go m1 !acc = mkStream $ \st stp sng yld ->

897 let single a = sng (done $ step acc a)

692 let single a = sng (done $ step acc a)

898 yieldk a r =

693 yieldk a r =

899 let s = step acc a

694 let s = step acc a

@@ -912,7 +707,7 @@ scanl' step begin = scanx step begin id

912filter :: IsStream t => (a -> Bool) -> t m a -> t m a

707filter :: IsStream t => (a -> Bool) -> t m a -> t m a

913filter p m = fromStream $ go (toStream m)

708filter p m = fromStream $ go (toStream m)

914 where

709 where

915− go m1 = Stream $ \st stp sng yld ->

710+ go m1 = mkStream $ \st stp sng yld ->

916 let single a | p a = sng a

711 let single a | p a = sng a

917 | otherwise = stp

712 | otherwise = stp

918 yieldk a r | p a = yld a (go r)

713 yieldk a r | p a = yld a (go r)

@@ -923,7 +718,7 @@ filter p m = fromStream $ go (toStream m)

923take :: IsStream t => Int -> t m a -> t m a

718take :: IsStream t => Int -> t m a -> t m a

924take n m = fromStream $ go n (toStream m)

719take n m = fromStream $ go n (toStream m)

925 where

720 where

926− go n1 m1 = Stream $ \st stp sng yld ->

721+ go n1 m1 = mkStream $ \st stp sng yld ->

927 let yieldk a r = yld a (go (n1 - 1) r)

722 let yieldk a r = yld a (go (n1 - 1) r)

928 in if n1 <= 0

723 in if n1 <= 0

929 then stp

724 then stp

@@ -933,7 +728,7 @@ take n m = fromStream $ go n (toStream m)

933takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

728takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

934takeWhile p m = fromStream $ go (toStream m)

729takeWhile p m = fromStream $ go (toStream m)

935 where

730 where

936− go m1 = Stream $ \st stp sng yld ->

731+ go m1 = mkStream $ \st stp sng yld ->

937 let single a | p a = sng a

732 let single a | p a = sng a

938 | otherwise = stp

733 | otherwise = stp

939 yieldk a r | p a = yld a (go r)

734 yieldk a r | p a = yld a (go r)

@@ -941,10 +736,10 @@ takeWhile p m = fromStream $ go (toStream m)

941 in unStream m1 st stp single yieldk

736 in unStream m1 st stp single yieldk

942

737

943drop :: IsStream t => Int -> t m a -> t m a

738drop :: IsStream t => Int -> t m a -> t m a

944−drop n m = fromStream $ Stream $ \st stp sng yld ->

739+drop n m = fromStream $ mkStream $ \st stp sng yld ->

945 unStream (go n (toStream m)) st stp sng yld

740 unStream (go n (toStream m)) st stp sng yld

946 where

741 where

947− go n1 m1 = Stream $ \st stp sng yld ->

742+ go n1 m1 = mkStream $ \st stp sng yld ->

948 let single _ = stp

743 let single _ = stp

949 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld

744 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld

950 -- Somehow "<=" check performs better than a ">"

745 -- Somehow "<=" check performs better than a ">"

@@ -956,7 +751,7 @@ drop n m = fromStream $ Stream $ \st stp sng yld ->

956dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

751dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

957dropWhile p m = fromStream $ go (toStream m)

752dropWhile p m = fromStream $ go (toStream m)

958 where

753 where

959− go m1 = Stream $ \st stp sng yld ->

754+ go m1 = mkStream $ \st stp sng yld ->

960 let single a | p a = stp

755 let single a | p a = stp

961 | otherwise = sng a

756 | otherwise = sng a

962 yieldk a r | p a = unStream r st stp single yieldk

757 yieldk a r | p a = unStream r st stp single yieldk

@@ -967,20 +762,13 @@ dropWhile p m = fromStream $ go (toStream m)

967-- Mapping

762-- Mapping

968-------------------------------------------------------------------------------

763-------------------------------------------------------------------------------

969

764

970−{-# INLINE map #-}

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

972−map f m = fromStream $ Stream $ \st stp sng yld ->

973− let single = sng . f

974− yieldk a r = yld (f a) (fmap f r)

975− in unStream (toStream m) st stp single yieldk

976−

977-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

765-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

978-- other stream types to overload it.

766-- other stream types to overload it.

979{-# INLINE mapM #-}

767{-# INLINE mapM #-}

980mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b

768mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b

981mapM f m = go (toStream m)

769mapM f m = go (toStream m)

982 where

770 where

983− go m1 = fromStream $ Stream $ \st stp sng yld ->

771+ go m1 = fromStream $ mkStream $ \st stp sng yld ->

984 let single a = f a >>= sng

772 let single a = f a >>= sng

985 yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld

773 yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld

986 in unStream m1 st stp single yieldk

774 in unStream m1 st stp single yieldk

@@ -991,7 +779,7 @@ mapM f m = go (toStream m)

991sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

779sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

992sequence m = go (toStream m)

780sequence m = go (toStream m)

993 where

781 where

994− go m1 = fromStream $ Stream $ \st stp sng yld ->

782+ go m1 = fromStream $ mkStream $ \st stp sng yld ->

995 let single ma = ma >>= sng

783 let single ma = ma >>= sng

996 yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld

784 yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld

997 in unStream m1 st stp single yieldk

785 in unStream m1 st stp single yieldk

@@ -1004,10 +792,10 @@ sequence m = go (toStream m)

1004intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

792intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

1005intersperseM a m = fromStream $ prependingStart (toStream m)

793intersperseM a m = fromStream $ prependingStart (toStream m)

1006 where

794 where

1007− prependingStart m1 = Stream $ \st stp sng yld ->

795+ prependingStart m1 = mkStream $ \st stp sng yld ->

1008 let yieldk i x = unStreamShared (return i |: go x) st stp sng yld

796 let yieldk i x = unStreamShared (return i |: go x) st stp sng yld

1009 in unStream m1 st stp sng yieldk

797 in unStream m1 st stp sng yieldk

1010− go m2 = fromStream $ Stream $ \st stp sng yld ->

798+ go m2 = fromStream $ mkStream $ \st stp sng yld ->

1011 let single i = unStreamShared (a |: yield i) st stp sng yld

799 let single i = unStreamShared (a |: yield i) st stp sng yld

1012 yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld

800 yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld

1013 in unStream m2 st stp single yieldk

801 in unStream m2 st stp single yieldk

@@ -1016,7 +804,7 @@ intersperseM a m = fromStream $ prependingStart (toStream m)

1016insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

804insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

1017insertBy cmp x m = fromStream $ go (toStream m)

805insertBy cmp x m = fromStream $ go (toStream m)

1018 where

806 where

1019− go m1 = Stream $ \st _ _ yld ->

807+ go m1 = mkStream $ \st _ _ yld ->

1020 let single a = case cmp x a of

808 let single a = case cmp x a of

1021 GT -> yld a (yield x)

809 GT -> yld a (yield x)

1022 _ -> yld x (yield a)

810 _ -> yld x (yield a)

@@ -1034,7 +822,7 @@ insertBy cmp x m = fromStream $ go (toStream m)

1034deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

822deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

1035deleteBy eq x m = fromStream $ go (toStream m)

823deleteBy eq x m = fromStream $ go (toStream m)

1036 where

824 where

1037− go m1 = Stream $ \st stp sng yld ->

825+ go m1 = mkStream $ \st stp sng yld ->

1038 let single a = if eq x a then stp else sng a

826 let single a = if eq x a then stp else sng a

1039 yieldk a r = if eq x a

827 yieldk a r = if eq x a

1040 then unStream r st stp sng yld

828 then unStream r st stp sng yld

@@ -1049,7 +837,7 @@ deleteBy eq x m = fromStream $ go (toStream m)

1049mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b

837mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b

1050mapMaybe f m = go (toStream m)

838mapMaybe f m = go (toStream m)

1051 where

839 where

1052− go m1 = fromStream $ Stream $ \st stp sng yld ->

840+ go m1 = fromStream $ mkStream $ \st stp sng yld ->

1053 let single a = case f a of

841 let single a = case f a of

1054 Just b -> sng b

842 Just b -> sng b

1055 Nothing -> stp

843 Nothing -> stp

@@ -1066,7 +854,7 @@ mapMaybe f m = go (toStream m)

1066zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

854zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

1067zipWithS f = go

855zipWithS f = go

1068 where

856 where

1069− go mx my = Stream $ \st stp sng yld -> do

857+ go mx my = mkStream $ \st stp sng yld -> do

1070 let merge a ra =

858 let merge a ra =

1071 let single2 b = sng (f a b)

859 let single2 b = sng (f a b)

1072 yield2 b rb = yld (f a b) (go ra rb)

860 yield2 b rb = yld (f a b) (go ra rb)

@@ -1088,7 +876,7 @@ zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

1088zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

876zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

1089zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

877zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

1090 where

878 where

1091− go mx my = Stream $ \st stp sng yld -> do

879+ go mx my = mkStream $ \st stp sng yld -> do

1092 let merge a ra =

880 let merge a ra =

1093 let runIt x = unStream x st stp sng yld

881 let runIt x = unStream x st stp sng yld

1094 single2 b = f a b >>= sng

882 single2 b = f a b >>= sng

@@ -1108,7 +896,7 @@ mergeByS

1108 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a

896 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a

1109mergeByS cmp = go

897mergeByS cmp = go

1110 where

898 where

1111− go mx my = Stream $ \st stp sng yld -> do

899+ go mx my = mkStream $ \st stp sng yld -> do

1112 let mergeWithY a ra =

900 let mergeWithY a ra =

1113 let stop2 = unStream mx st stp sng yld

901 let stop2 = unStream mx st stp sng yld

1114 single2 b = do

902 single2 b = do

@@ -1158,40 +946,6 @@ the m = do

1158 | otherwise = return Nothing

946 | otherwise = return Nothing

1159 in unStream m1 defState (return $ Just h) single yieldk

947 in unStream m1 defState (return $ Just h) single yieldk

1160

948

1161−------------------------------------------------------------------------------

1162−-- Semigroup

1163−------------------------------------------------------------------------------

1164−

1165−-- | Concatenates two streams sequentially i.e. the first stream is

1166−-- exhausted completely before yielding any element from the second stream.

1167−{-# INLINE serial #-}

1168−serial :: Stream m a -> Stream m a -> Stream m a

1169−serial m1 m2 = go m1

1170− where

1171− go m = Stream $ \st stp sng yld ->

1172− let stop = unStream m2 st stp sng yld

1173− single a = yld a m2

1174− yieldk a r = yld a (go r)

1175− in unStream m st stop single yieldk

1176−

1177−instance Semigroup (Stream m a) where

1178− (<>) = serial

1179−

1180−------------------------------------------------------------------------------

1181−-- Monoid

1182−------------------------------------------------------------------------------

1183−

1184−instance Monoid (Stream m a) where

1185− mempty = nil

1186− mappend = (<>)

1187−

1188−-------------------------------------------------------------------------------

1189−-- Functor

1190−-------------------------------------------------------------------------------

1191−

1192−instance Monad m => Functor (Stream m) where

1193− fmap = map

1194−

1195-------------------------------------------------------------------------------

949-------------------------------------------------------------------------------

1196-- Bind utility

950-- Bind utility

1197-------------------------------------------------------------------------------

951-------------------------------------------------------------------------------

@@ -1205,7 +959,7 @@ bindWith

1205bindWith par m1 f = go m1

959bindWith par m1 f = go m1

1206 where

960 where

1207 go m =

961 go m =

1208− Stream $ \st stp sng yld ->

962+ mkStream $ \st stp sng yld ->

1209 let runShared x = unStreamShared x st stp sng yld

963 let runShared x = unStreamShared x st stp sng yld

1210 runIsolated x = unStream x st stp sng yld

964 runIsolated x = unStream x st stp sng yld

1211

965

@@ -1218,7 +972,7 @@ bindWith par m1 f = go m1

1218------------------------------------------------------------------------------

972------------------------------------------------------------------------------

1219

973

1220_alt :: Stream m a -> Stream m a -> Stream m a

974_alt :: Stream m a -> Stream m a -> Stream m a

1221−_alt m1 m2 = Stream $ \st stp sng yld ->

975+_alt m1 m2 = mkStream $ \st stp sng yld ->

1222 let stop = unStream m2 st stp sng yld

976 let stop = unStream m2 st stp sng yld

1223 in unStream m1 st stop sng yld

977 in unStream m1 st stop sng yld

1224

978

@@ -1228,7 +982,7 @@ _alt m1 m2 = Stream $ \st stp sng yld ->

1228

982

1229withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

983withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

1230withLocal f m =

984withLocal f m =

1231− Stream $ \st stp sng yld ->

985+ mkStream $ \st stp sng yld ->

1232 let single = local f . sng

986 let single = local f . sng

1233 yieldk a r = local f $ yld a (withLocal f r)

987 yieldk a r = local f $ yld a (withLocal f r)

1234 in unStream m st (local f stp) single yieldk

988 in unStream m st (local f stp) single yieldk

@@ -1243,16 +997,10 @@ withCatchError

1243 :: MonadError e m

997 :: MonadError e m

1244 => Stream m a -> (e -> Stream m a) -> Stream m a

998 => Stream m a -> (e -> Stream m a) -> Stream m a

1245withCatchError m h =

999withCatchError m h =

1246− Stream $ \_ stp sng yld ->

1000+ mkStream $ \_ stp sng yld ->

1247 let run x = unStream x Nothing stp sng yieldk

1001 let run x = unStream x Nothing stp sng yieldk

1248 handle r = r `catchError` \e -> run $ h e

1002 handle r = r `catchError` \e -> run $ h e

1249 yieldk a r = yld a (withCatchError r h)

1003 yieldk a r = yld a (withCatchError r h)

1250 in handle $ run m

1004 in handle $ run m

1251-}

1005-}

1252

1006

1253−-------------------------------------------------------------------------------

1254−-- Transformers

1255−-------------------------------------------------------------------------------

1256−

1257−instance MonadTrans Stream where

1258− lift = yieldM

src/Streamly/Streams/StreamKType.hsnew file+319

@@ -0,0 +1,319 @@

1+{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

3+{-# LANGUAGE ConstraintKinds #-}

4+{-# LANGUAGE FlexibleContexts #-}

5+{-# LANGUAGE FlexibleInstances #-}

6+{-# LANGUAGE InstanceSigs #-}

7+{-# LANGUAGE MultiParamTypeClasses #-}

8+{-# LANGUAGE PatternSynonyms #-}

9+{-# LANGUAGE ViewPatterns #-}

10+{-# LANGUAGE RankNTypes #-}

11+{-# LANGUAGE UndecidableInstances #-} -- XXX

12+

13+-- |

14+-- Module : Streamly.Streams.StreamKType

16+--

17+-- License : BSD3

18+-- Maintainer : harendra.kumar@gmail.com

19+-- Stability : experimental

20+-- Portability : GHC

21+--

22+--

23+-- Continuation passing style (CPS) stream implementation. The symbol 'K' below

24+-- denotes a function as well as a Kontinuation.

25+--

26+module Streamly.Streams.StreamKType

27+ (

28+ -- * A class for streams

29+ IsStream (..)

30+ , adapt

31+

32+ -- * The stream type

33+ , Stream ()

34+ , Streaming

35+

36+ -- * Construction

37+ , consMSerial

38+ , mkStream

39+

40+ -- * Elimination

41+ , unStream

42+ , unStreamShared

43+ , foldStream

44+ , runStreamSVar

45+

46+ , nil

47+ , serial

48+ , map

49+ , yieldM

50+ )

51+where

52+

53+import Control.Monad (void)

54+import Control.Monad.IO.Class (MonadIO(liftIO))

55+import Control.Monad.Trans.Class (MonadTrans(lift))

56+import Data.Semigroup (Semigroup(..))

57+import Prelude hiding (map)

58+

59+import Streamly.SVar

60+

61+------------------------------------------------------------------------------

62+-- The basic stream type

63+------------------------------------------------------------------------------

64+

65+-- | The type @Stream m a@ represents a monadic stream of values of type 'a'

66+-- constructed using actions in monad 'm'. It uses stop, singleton and yield

67+-- continuations equivalent to the following direct style type:

68+--

69+-- @

70+-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

71+-- @

72+--

73+-- To facilitate parallel composition we maintain a local state in an 'SVar'

74+-- that is shared across and is used for synchronization of the streams being

75+-- composed.

76+--

77+-- The singleton case can be expressed in terms of stop and yield but we have

78+-- it as a separate case to optimize composition operations for streams with

79+-- single element. We build singleton streams in the implementation of 'pure'

80+-- for Applicative and Monad, and in 'lift' for MonadTrans.

81+--

82+newtype Stream m a =

83+ MkStream (forall r.

84+ State Stream m a -- state

85+ -> m r -- stop

86+ -> (a -> m r) -- singleton

87+ -> (a -> Stream m a -> m r) -- yield

88+ -> m r

89+ )

90+

91+------------------------------------------------------------------------------

92+-- Types that can behave as a Stream

93+------------------------------------------------------------------------------

94+

95+infixr 5 `consM`

96+infixr 5 |:

97+

98+-- | Class of types that can represent a stream of elements of some type 'a' in

99+-- some monad 'm'.

100+--

101+-- @since 0.2.0

102+class IsStream t where

103+ toStream :: t m a -> Stream m a

104+ fromStream :: Stream m a -> t m a

105+ -- | Constructs a stream by adding a monadic action at the head of an

106+ -- existing stream. For example:

107+ --

108+ -- @

109+ -- > toList $ getLine \`consM` getLine \`consM` nil

110+ -- hello

111+ -- world

112+ -- ["hello","world"]

113+ -- @

114+ --

115+ -- /Concurrent (do not use 'parallely' to construct infinite streams)/

116+ --

117+ -- @since 0.2.0

118+ consM :: MonadAsync m => m a -> t m a -> t m a

119+ -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"

120+ -- to remember that @|@ comes before ':'.

121+ --

122+ -- @

123+ -- > toList $ getLine |: getLine |: nil

124+ -- hello

125+ -- world

126+ -- ["hello","world"]

127+ -- @

128+ --

129+ -- @

130+ -- let delay = threadDelay 1000000 >> print 1

131+ -- runStream $ serially $ delay |: delay |: delay |: nil

132+ -- runStream $ parallely $ delay |: delay |: delay |: nil

133+ -- @

134+ --

135+ -- /Concurrent (do not use 'parallely' to construct infinite streams)/

136+ --

137+ -- @since 0.2.0

138+ (|:) :: MonadAsync m => m a -> t m a -> t m a

139+ -- We can define (|:) just as 'consM' but it is defined explicitly for each

140+ -- type because we want to use SPECIALIZE pragma on the definition.

141+

142+-- | Same as 'IsStream'.

143+--

144+-- @since 0.1.0

145+{-# DEPRECATED Streaming "Please use IsStream instead." #-}

146+type Streaming = IsStream

147+

148+-------------------------------------------------------------------------------

149+-- Type adapting combinators

150+-------------------------------------------------------------------------------

151+

152+-- | Adapt any specific stream type to any other specific stream type.

153+--

154+-- @since 0.1.0

155+adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

156+adapt = fromStream . toStream

157+

158+-------------------------------------------------------------------------------

159+-- Instances

160+-------------------------------------------------------------------------------

161+

162+{-# INLINE consMSerial #-}

163+consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

164+consMSerial m r = MkStream $ \_ _ _ yld -> m >>= \a -> yld a r

165+

166+-------------------------------------------------------------------------------

167+-- IsStream Stream

168+-------------------------------------------------------------------------------

169+

170+instance IsStream Stream where

171+ toStream = id

172+ fromStream = id

173+

174+ {-# INLINE consM #-}

175+ {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}

176+ consM :: Monad m => m a -> Stream m a -> Stream m a

177+ consM = consMSerial

178+

179+ {-# INLINE (|:) #-}

180+ {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}

181+ (|:) :: Monad m => m a -> Stream m a -> Stream m a

182+ (|:) = consMSerial

183+

184+------------------------------------------------------------------------------

185+-- Building a stream

186+------------------------------------------------------------------------------

187+

188+-- | Build a stream from an 'SVar', a stop continuation, a singleton stream

189+-- continuation and a yield continuation.

190+mkStream:: IsStream t

191+ => (forall r. State Stream m a

192+ -> m r

193+ -> (a -> m r)

194+ -> (a -> t m a -> m r)

195+ -> m r)

196+ -> t m a

197+mkStream k = fromStream $ MkStream $ \st stp sng yld ->

198+ let yieldk a r = yld a (toStream r)

199+ in k st stp sng yieldk

200+

201+-- | An empty stream.

202+--

203+-- @

204+-- > toList nil

205+-- []

206+-- @

207+--

208+-- @since 0.1.0

209+nil :: IsStream t => t m a

210+nil = fromStream $ mkStream $ \_ stp _ _ -> stp

211+

212+------------------------------------------------------------------------------

213+-- Folding a stream

214+------------------------------------------------------------------------------

215+

216+-- Run a stream in detached mode i.e. not joining an SVar

217+{-# INLINE unStream #-}

218+unStream ::

219+ Stream m a

220+ -> State Stream m b -- state

221+ -> m r -- stop

222+ -> (a -> m r) -- singleton

223+ -> (a -> Stream m a -> m r) -- yield

224+ -> m r

225+unStream (MkStream runner) st = runner (adaptState st)

226+

227+-- | Like unstream, but passes a shared SVar across continuations.

228+{-# INLINE unStreamShared #-}

229+unStreamShared ::

230+ Stream m a

231+ -> State Stream m a -- state

232+ -> m r -- stop

233+ -> (a -> m r) -- singleton

234+ -> (a -> Stream m a -> m r) -- yield

235+ -> m r

236+unStreamShared (MkStream runner) st stp sng yld = runner st stp sng yld

237+

238+-- | Fold a stream by providing an SVar, a stop continuation, a singleton

239+-- continuation and a yield continuation.

240+foldStream

241+ :: IsStream t

242+ => State Stream m a

243+ -> m r

244+ -> (a -> m r)

245+ -> (a -> t m a -> m r)

246+ -> t m a

247+ -> m r

248+foldStream st blank single step m =

249+ let yieldk a x = step a (fromStream x)

250+ in unStreamShared (toStream m) st blank single yieldk

251+

252+-- Run the stream using a run function associated with the SVar that runs the

253+-- streams with a captured snapshot of the monadic state.

254+{-# INLINE runStreamSVar #-}

255+runStreamSVar

256+ :: MonadIO m

257+ => SVar Stream m a

258+ -> Stream m a

259+ -> State Stream m a -- state

260+ -> m r -- stop

261+ -> (a -> m r) -- singleton

262+ -> (a -> Stream m a -> m r) -- yield

263+ -> m ()

264+runStreamSVar sv m st stp sng yld =

265+ let mrun = runInIO $ svarMrun sv

266+ in void $ liftIO $ mrun $ unStreamShared m st stp sng yld

267+

268+------------------------------------------------------------------------------

269+-- Semigroup

270+------------------------------------------------------------------------------

271+

272+-- | Concatenates two streams sequentially i.e. the first stream is

273+-- exhausted completely before yielding any element from the second stream.

274+{-# INLINE serial #-}

275+serial :: Stream m a -> Stream m a -> Stream m a

276+serial m1 m2 = go m1

277+ where

278+ go m = mkStream $ \st stp sng yld ->

279+ let stop = unStream m2 st stp sng yld

280+ single a = yld a m2

281+ yieldk a r = yld a (go r)

282+ in unStream m st stop single yieldk

283+

284+instance Semigroup (Stream m a) where

285+ (<>) = serial

286+

287+------------------------------------------------------------------------------

288+-- Monoid

289+------------------------------------------------------------------------------

290+

291+instance Monoid (Stream m a) where

292+ mempty = nil

293+ mappend = (<>)

294+

295+-------------------------------------------------------------------------------

296+-- Functor

297+-------------------------------------------------------------------------------

298+

299+

300+{-# INLINE map #-}

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

302+map f m = fromStream $ mkStream $ \st stp sng yld ->

303+ let single = sng . f

304+ yieldk a r = yld (f a) (fmap f r)

305+ in unStream (toStream m) st stp single yieldk

306+

307+instance Monad m => Functor (Stream m) where

308+ fmap = map

309+

310+-------------------------------------------------------------------------------

311+-- Transformers

312+-------------------------------------------------------------------------------

313+

314+{-# INLINE yieldM #-}

315+yieldM :: (Monad m, IsStream t) => m a -> t m a

316+yieldM m = fromStream $ mkStream $ \_ _ single _ -> m >>= single

317+

318+instance MonadTrans Stream where

319+ lift = yieldM

streamly.cabal+1

@@ -128,6 +128,7 @@ library

128 other-modules: Streamly.SVar

129

130 -- Base streams

131+ , Streamly.Streams.StreamKType

131 , Streamly.Streams.StreamK

132 , Streamly.Streams.StreamK

132 , Streamly.Streams.StreamDType

133 , Streamly.Streams.StreamDType

133 , Streamly.Streams.StreamD

134 , Streamly.Streams.StreamD