5ee02ff · streamly · git.iko.soy

141import qualified Prelude

142import qualified System.IO as IO

143

144−import Streamly.Streams.CPS

144+import Streamly.Streams.StreamK

145−import qualified Streamly.Streams.CPS as C

145+import qualified Streamly.Streams.StreamK as K

146import Streamly.Streams.Zip

147import Streamly.Streams.Serial

148import Streamly.SVar (MonadAsync)

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

173unfoldr step = fromStream . go

174 where

175− go s = C.Stream $ \_ stp _ yld ->

175+ go s = K.Stream $ \_ stp _ yld ->

176 case step s of

177 Nothing -> stp

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

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

213unfoldrM step = go

214 where

215− go s = fromStream $ C.Stream $ \svr stp sng yld -> do

215+ go s = fromStream $ K.Stream $ \svr stp sng yld -> do

216 mayb <- step s

217 case mayb of

218 Nothing -> stp

219 Just (a, b) ->

220− C.runStream (toStream (return a |: go b)) svr stp sng yld

220+ K.runStream (toStream (return a |: go b)) svr stp sng yld

221

222-- | Construct a stream from a 'Foldable' containing pure values.

223--

284iterate :: IsStream t => (a -> a) -> a -> t m a

285iterate step = fromStream . go

286 where

287− go s = C.cons s (go (step s))

287+ go s = K.cons s (go (step s))

288

289-- | Iterate a monadic function from a seed value, streaming the results

290-- forever.

308iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a

309iterateM step = go

310 where

311− go s = fromStream $ C.Stream $ \svr stp sng yld -> do

311+ go s = fromStream $ K.Stream $ \svr stp sng yld -> do

312 next <- step s

313− C.runStream (toStream (return s |: go next)) svr stp sng yld

313+ K.runStream (toStream (return s |: go next)) svr stp sng yld

314

315-- | Read lines from an IO Handle into a stream of Strings.

316--

318fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String

319fromHandle h = fromStream go

320 where

321− go = C.Stream $ \_ stp _ yld -> do

321+ go = K.Stream $ \_ stp _ yld -> do

322 eof <- liftIO $ IO.hIsEOF h

323 if eof

324 then stp

345 let stop = return acc

346 single a = return (step a acc)

347 yield a r = go r >>= \b -> return (step a b)

348− in (C.runStream m1) Nothing stop single yield

348+ in (K.runStream m1) Nothing stop single yield

349

350-- | Lazy right fold with a monadic step function. For example, to fold a

351-- stream into a list:

364 let stop = return acc

365 single a = step a acc

366 yield a r = go r >>= step a

367− in (C.runStream m1) Nothing stop single yield

367+ in (K.runStream m1) Nothing stop single yield

368

369-- | Strict left scan with an extraction function. Like 'scanl'', but applies a

370-- user supplied extraction function (the third argument) at each step. This is

377scanx step begin done m =

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

379 where

380− go m1 !acc = C.Stream $ \_ stp sng yld ->

380+ go m1 !acc = K.Stream $ \_ stp sng yld ->

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

382 yield a r =

383 let s = step acc a

384 in yld (done s) (go r s)

385− in C.runStream m1 Nothing stp single yield

385+ in K.runStream m1 Nothing stp single yield

386

387-- |

388-- @since 0.1.1

413 {-# NOINLINE get #-}

414 get m1 =

415 let single = return . done

416− in (C.runStream m1) Nothing undefined single undefined

416+ in (K.runStream m1) Nothing undefined single undefined

417

418 -- Note, this can be implemented by making a recursive call to "go",

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

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

421 -- continuations is much more efficient.

422− go m1 !acc = C.Stream $ \_ _ sng yld ->

422+ go m1 !acc = K.Stream $ \_ _ sng yld ->

423 let stop = sng acc

424 single a = sng $ step acc a

425 yield a r =

426 let stream = go r (step acc a)

427− in (C.runStream stream) Nothing undefined sng yld

427+ in (K.runStream stream) Nothing undefined sng yld

428− in (C.runStream m1) Nothing stop single yield

428+ in (K.runStream m1) Nothing stop single yield

429

430-- |

431-- @since 0.1.0

451 let stop = acc >>= done

452 single a = acc >>= \b -> step b a >>= done

453 yield a r = acc >>= \b -> go (step b a) r

454− in (C.runStream m1) Nothing stop single yield

454+ in (K.runStream m1) Nothing stop single yield

455

456-- |

457-- @since 0.1.0

475 let stop = return Nothing

476 single a = return (Just (a, nil))

477 yield a r = return (Just (a, fromStream r))

478− in (C.runStream (toStream m)) Nothing stop single yield

478+ in (K.runStream (toStream m)) Nothing stop single yield

479

480-- | Write a stream of Strings to an IO Handle.

481--

487 let stop = return ()

488 single a = liftIO (IO.hPutStrLn h a)

489 yield a r = liftIO (IO.hPutStrLn h a) >> go r

490− in (C.runStream m1) Nothing stop single yield

490+ in (K.runStream m1) Nothing stop single yield

491

492------------------------------------------------------------------------------

493-- Special folds

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

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

509 where

510− go n1 m1 = C.Stream $ \_ stp sng yld ->

510+ go n1 m1 = K.Stream $ \_ stp sng yld ->

511 let yield a r = yld a (go (n1 - 1) r)

512− in if n1 <= 0 then stp else (C.runStream m1) Nothing stp sng yield

512+ in if n1 <= 0 then stp else (K.runStream m1) Nothing stp sng yield

513

514-- | Include only those elements that pass a predicate.

515--

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

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

520 where

521− go m1 = C.Stream $ \_ stp sng yld ->

521+ go m1 = K.Stream $ \_ stp sng yld ->

522 let single a | p a = sng a

523 | otherwise = stp

524 yield a r | p a = yld a (go r)

525− | otherwise = (C.runStream r) Nothing stp single yield

525+ | otherwise = (K.runStream r) Nothing stp single yield

526− in (C.runStream m1) Nothing stp single yield

526+ in (K.runStream m1) Nothing stp single yield

527

528-- | End the stream as soon as the predicate fails on an element.

529--

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

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

534 where

535− go m1 = C.Stream $ \_ stp sng yld ->

535+ go m1 = K.Stream $ \_ stp sng yld ->

536 let single a | p a = sng a

537 | otherwise = stp

538 yield a r | p a = yld a (go r)

539 | otherwise = stp

540− in (C.runStream m1) Nothing stp single yield

540+ in (K.runStream m1) Nothing stp single yield

541

542-- | Discard first 'n' elements from the stream and take the rest.

543--

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

546drop n m = fromStream $ go n (toStream m)

547 where

548− go n1 m1 = C.Stream $ \_ stp sng yld ->

548+ go n1 m1 = K.Stream $ \_ stp sng yld ->

549 let single _ = stp

550− yield _ r = (C.runStream $ go (n1 - 1) r) Nothing stp sng yld

550+ yield _ r = (K.runStream $ go (n1 - 1) r) Nothing stp sng yld

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

552 in if n1 <= 0

553− then (C.runStream m1) Nothing stp sng yld

553+ then (K.runStream m1) Nothing stp sng yld

554− else (C.runStream m1) Nothing stp single yield

554+ else (K.runStream m1) Nothing stp single yield

555

556-- | Drop elements in the stream as long as the predicate succeeds and then

557-- take the rest of the stream.

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

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

563 where

564− go m1 = C.Stream $ \_ stp sng yld ->

564+ go m1 = K.Stream $ \_ stp sng yld ->

565 let single a | p a = stp

566 | otherwise = sng a

567− yield a r | p a = (C.runStream r) Nothing stp single yield

567+ yield a r | p a = (K.runStream r) Nothing stp single yield

568 | otherwise = yld a r

569− in (C.runStream m1) Nothing stp single yield

569+ in (K.runStream m1) Nothing stp single yield

570

571-- | Determine whether all elements of a stream satisfy a predicate.

572--

579 | otherwise = return False

580 yield a r | p a = go r

581 | otherwise = return False

582− in (C.runStream m1) Nothing (return True) single yield

582+ in (K.runStream m1) Nothing (return True) single yield

583

584-- | Determine whether any of the elements of a stream satisfy a predicate.

585--

592 | otherwise = return False

593 yield a r | p a = return True

594 | otherwise = go r

595− in (C.runStream m1) Nothing (return False) single yield

595+ in (K.runStream m1) Nothing (return False) single yield

596

597-- | Determine the sum of all elements of a stream of numbers

598--

614 let stop = return Nothing

615 single a = return (Just a)

616 yield a _ = return (Just a)

617− in (C.runStream (toStream m)) Nothing stop single yield

617+ in (K.runStream (toStream m)) Nothing stop single yield

618

619-- | Extract all but the first element of the stream, if any.

620--

624 let stop = return Nothing

625 single _ = return $ Just nil

626 yield _ r = return $ Just $ fromStream r

627− in (C.runStream (toStream m)) Nothing stop single yield

627+ in (K.runStream (toStream m)) Nothing stop single yield

628

629-- | Extract the last element of the stream, if any.

630--

641 let stop = return True

642 single _ = return False

643 yield _ _ = return False

644− in (C.runStream (toStream m)) Nothing stop single yield

644+ in (K.runStream (toStream m)) Nothing stop single yield

645

646-- | Determine whether an element is present in the stream.

647--

653 let stop = return False

654 single a = return (a == e)

655 yield a r = if a == e then return True else go r

656− in (C.runStream m1) Nothing stop single yield

656+ in (K.runStream m1) Nothing stop single yield

657

658-- | Determine whether an element is not present in the stream.

659--

665 let stop = return True

666 single a = return (a /= e)

667 yield a r = if a == e then return False else go r

668− in (C.runStream m1) Nothing stop single yield

668+ in (K.runStream m1) Nothing stop single yield

669

670-- | Determine the length of the stream.

671--

678--

679-- @since 0.1.1

680reverse :: (IsStream t) => t m a -> t m a

681−reverse m = fromStream $ go C.nil (toStream m)

681+reverse m = fromStream $ go K.nil (toStream m)

682 where

683− go rev rest = C.Stream $ \_ stp sng yld ->

683+ go rev rest = K.Stream $ \_ stp sng yld ->

684− let runIt x = C.runStream x Nothing stp sng yld

684+ let runIt x = K.runStream x Nothing stp sng yld

685 stop = runIt rev

686− single a = runIt $ a `C.cons` rev

686+ single a = runIt $ a `K.cons` rev

687− yield a r = runIt $ go (a `C.cons` rev) r

687+ yield a r = runIt $ go (a `K.cons` rev) r

688− in C.runStream rest Nothing stop single yield

688+ in K.runStream rest Nothing stop single yield

689

690-- XXX replace the recursive "go" with continuation

691-- | Determine the minimum element in a stream.

698 let stop = return res

699 single a = return $ min_ a res

700 yield a r = go (min_ a res) r

701− in (C.runStream m1) Nothing stop single yield

701+ in (K.runStream m1) Nothing stop single yield

702

703 min_ a res = case res of

704 Nothing -> Just a

715 let stop = return res

716 single a = return $ max_ a res

717 yield a r = go (max_ a res) r

718− in (C.runStream m1) Nothing stop single yield

718+ in (K.runStream m1) Nothing stop single yield

719

720 max_ a res = case res of

721 Nothing -> Just a

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

744mapM f m = go (toStream m)

745 where

746− go m1 = fromStream $ C.Stream $ \svr stp sng yld ->

746+ go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

747 let single a = f a >>= sng

748− yield a r = C.runStream (toStream (f a |: (go r))) svr stp sng yld

748+ yield a r = K.runStream (toStream (f a |: (go r))) svr stp sng yld

749− in (C.runStream m1) Nothing stp single yield

749+ in (K.runStream m1) Nothing stp single yield

750

751-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'

752-- elements, and return a stream of values extracted from 'Just'.

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

757mapMaybe f m = go (toStream m)

758 where

759− go m1 = fromStream $ C.Stream $ \_ stp sng yld ->

759+ go m1 = fromStream $ K.Stream $ \_ stp sng yld ->

760 let single a = case f a of

761 Just b -> sng b

762 Nothing -> stp

763 yield a r = case f a of

764 Just b -> yld b (toStream $ go r)

765− Nothing -> (C.runStream r) Nothing stp single yield

765+ Nothing -> (K.runStream r) Nothing stp single yield

766− in (C.runStream m1) Nothing stp single yield

766+ in (K.runStream m1) Nothing stp single yield

767

768-- | Like 'mapMaybe' but maps a monadic function.

769--

787 let stop = return ()

788 single a = void (f a)

789 yield a r = f a >> go r

790− in (C.runStream m1) Nothing stop single yield

790+ in (K.runStream m1) Nothing stop single yield

791

792-- | Reduce a stream of monadic actions to a stream of the output of those

793-- actions.

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

807sequence m = go (toStream m)

808 where

809− go m1 = fromStream $ C.Stream $ \svr stp sng yld ->

809+ go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

810 let single ma = ma >>= sng

811− yield ma r = C.runStream (toStream $ ma |: go r) svr stp sng yld

811+ yield ma r = K.runStream (toStream $ ma |: go r) svr stp sng yld

812− in (C.runStream m1) Nothing stp single yield

812+ in (K.runStream m1) Nothing stp single yield