@@ -314,12 +314,10 @@ import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_, control)
314314import Data.Functor.Identity (Identity(..))
315315import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)
316316import Data.Maybe (fromJust, isJust)
317−import Fusion.Plugin.Types (Fuse(..))
318317import Streamly.Internal.Data.Time.Units
319318 (TimeUnit64, toRelTime64, diffAbsTime64)
320319import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))
321320import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
322−import Streamly.Internal.Data.Unfold.Types (Unfold(..))
323321
324322import qualified Streamly.Internal.Data.Pipe.Types as Pipe
325323import qualified Streamly.Internal.Data.Fold as FL
@@ -337,6 +335,7 @@ import Prelude hiding
337335import Streamly.Internal.Data.Stream.StreamD.Type
338336import Streamly.Internal.Data.Stream.StreamD.Generate
339337import Streamly.Internal.Data.Stream.StreamD.Eliminate
338+import Streamly.Internal.Data.Stream.StreamD.Flatten
340339import Streamly.Internal.Data.Stream.StreamD.SplitGroup
341340import Streamly.Internal.Data.SVar
342341
@@ -600,721 +599,6 @@ stripPrefix (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
600599mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()
601600mapM_ m = drain . mapM m
602601
603−-------------------------------------------------------------------------------
604−-- Stream transformations using Unfolds
605−-------------------------------------------------------------------------------
606−
607−-- Define a unique structure to use in inspection testing
608−data ConcatMapUState o i =
609− ConcatMapUOuter o
610− | ConcatMapUInner o i
611−
612−-- | @concatMapU unfold stream@ uses @unfold@ to map the input stream elements
613−-- to streams and then flattens the generated streams into a single output
614−-- stream.
615−
616−-- This is like 'concatMap' but uses an unfold with an explicit state to
617−-- generate the stream instead of a 'Stream' type generator. This allows better
618−-- optimization via fusion. This can be many times more efficient than
619−-- 'concatMap'.
620−
621−{-# INLINE_NORMAL concatMapU #-}
622−concatMapU :: Monad m => Unfold m a b -> Stream m a -> Stream m b
623−concatMapU (Unfold istep inject) (Stream ostep ost) =
624− Stream step (ConcatMapUOuter ost)
625− where
626− {-# INLINE_LATE step #-}
627− step gst (ConcatMapUOuter o) = do
628− r <- ostep (adaptState gst) o
629− case r of
630− Yield a o' -> do
631− i <- inject a
632− i `seq` return (Skip (ConcatMapUInner o' i))
633− Skip o' -> return $ Skip (ConcatMapUOuter o')
634− Stop -> return $ Stop
635−
636− step _ (ConcatMapUInner o i) = do
637− r <- istep i
638− return $ case r of
639− Yield x i' -> Yield x (ConcatMapUInner o i')
640− Skip i' -> Skip (ConcatMapUInner o i')
641− Stop -> Skip (ConcatMapUOuter o)
642−
643−data ConcatUnfoldInterleaveState o i =
644− ConcatUnfoldInterleaveOuter o [i]
645− | ConcatUnfoldInterleaveInner o [i]
646− | ConcatUnfoldInterleaveInnerL [i] [i]
647− | ConcatUnfoldInterleaveInnerR [i] [i]
648−
649−-- XXX use arrays to store state instead of lists.
650−-- XXX In general we can use different scheduling strategies e.g. how to
651−-- schedule the outer vs inner loop or assigning weights to different streams
652−-- or outer and inner loops.
653−
654−-- After a yield, switch to the next stream. Do not switch streams on Skip.
655−-- Yield from outer stream switches to the inner stream.
656−--
657−-- There are two choices here, (1) exhaust the outer stream first and then
658−-- start yielding from the inner streams, this is much simpler to implement,
659−-- (2) yield at least one element from an inner stream before going back to
660−-- outer stream and opening the next stream from it.
661−--
662−-- Ideally, we need some scheduling bias to inner streams vs outer stream.
663−-- Maybe we can configure the behavior.
664−--
665−{-# INLINE_NORMAL concatUnfoldInterleave #-}
666−concatUnfoldInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
667−concatUnfoldInterleave (Unfold istep inject) (Stream ostep ost) =
668− Stream step (ConcatUnfoldInterleaveOuter ost [])
669− where
670− {-# INLINE_LATE step #-}
671− step gst (ConcatUnfoldInterleaveOuter o ls) = do
672− r <- ostep (adaptState gst) o
673− case r of
674− Yield a o' -> do
675− i <- inject a
676− i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
677− Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
678− Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
679−
680− step _ (ConcatUnfoldInterleaveInner _ []) = undefined
681− step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
682− r <- istep st
683− return $ case r of
684− Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
685− Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
686− Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
687−
688− step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
689− step _ (ConcatUnfoldInterleaveInnerL [] rs) =
690− return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
691−
692− step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
693− r <- istep st
694− return $ case r of
695− Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
696− Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
697− Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
698−
699− step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
700− step _ (ConcatUnfoldInterleaveInnerR ls []) =
701− return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
702−
703− step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
704− r <- istep st
705− return $ case r of
706− Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
707− Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
708− Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
709−
710−-- XXX In general we can use different scheduling strategies e.g. how to
711−-- schedule the outer vs inner loop or assigning weights to different streams
712−-- or outer and inner loops.
713−--
714−-- This could be inefficient if the tasks are too small.
715−--
716−-- Compared to concatUnfoldInterleave this one switches streams on Skips.
717−--
718−{-# INLINE_NORMAL concatUnfoldRoundrobin #-}
719−concatUnfoldRoundrobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
720−concatUnfoldRoundrobin (Unfold istep inject) (Stream ostep ost) =
721− Stream step (ConcatUnfoldInterleaveOuter ost [])
722− where
723− {-# INLINE_LATE step #-}
724− step gst (ConcatUnfoldInterleaveOuter o ls) = do
725− r <- ostep (adaptState gst) o
726− case r of
727− Yield a o' -> do
728− i <- inject a
729− i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
730− Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
731− Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
732−
733− step _ (ConcatUnfoldInterleaveInner o []) =
734− return $ Skip (ConcatUnfoldInterleaveOuter o [])
735−
736− step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
737− r <- istep st
738− return $ case r of
739− Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
740− Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
741− Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
742−
743− step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
744− step _ (ConcatUnfoldInterleaveInnerL [] rs) =
745− return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
746−
747− step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
748− r <- istep st
749− return $ case r of
750− Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
751− Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
752− Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
753−
754− step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
755− step _ (ConcatUnfoldInterleaveInnerR ls []) =
756− return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
757−
758− step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
759− r <- istep st
760− return $ case r of
761− Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
762− Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
763− Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
764−
765−data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
766−
767−-- Note that this could be much faster compared to the CPS stream. However, as
768−-- the number of streams being composed increases this may become expensive.
769−-- Need to see where the breaking point is between the two.
770−--
771−{-# INLINE_NORMAL append #-}
772−append :: Monad m => Stream m a -> Stream m a -> Stream m a
773−append (Stream step1 state1) (Stream step2 state2) =
774− Stream step (AppendFirst state1)
775−
776− where
777−
778− {-# INLINE_LATE step #-}
779− step gst (AppendFirst st) = do
780− r <- step1 gst st
781− return $ case r of
782− Yield a s -> Yield a (AppendFirst s)
783− Skip s -> Skip (AppendFirst s)
784− Stop -> Skip (AppendSecond state2)
785−
786− step gst (AppendSecond st) = do
787− r <- step2 gst st
788− return $ case r of
789− Yield a s -> Yield a (AppendSecond s)
790− Skip s -> Skip (AppendSecond s)
791− Stop -> Stop
792−
793−data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2
794− | InterleaveSecondOnly s2 | InterleaveFirstOnly s1
795−
796−{-# INLINE_NORMAL interleave #-}
797−interleave :: Monad m => Stream m a -> Stream m a -> Stream m a
798−interleave (Stream step1 state1) (Stream step2 state2) =
799− Stream step (InterleaveFirst state1 state2)
800−
801− where
802−
803− {-# INLINE_LATE step #-}
804− step gst (InterleaveFirst st1 st2) = do
805− r <- step1 gst st1
806− return $ case r of
807− Yield a s -> Yield a (InterleaveSecond s st2)
808− Skip s -> Skip (InterleaveFirst s st2)
809− Stop -> Skip (InterleaveSecondOnly st2)
810−
811− step gst (InterleaveSecond st1 st2) = do
812− r <- step2 gst st2
813− return $ case r of
814− Yield a s -> Yield a (InterleaveFirst st1 s)
815− Skip s -> Skip (InterleaveSecond st1 s)
816− Stop -> Skip (InterleaveFirstOnly st1)
817−
818− step gst (InterleaveFirstOnly st1) = do
819− r <- step1 gst st1
820− return $ case r of
821− Yield a s -> Yield a (InterleaveFirstOnly s)
822− Skip s -> Skip (InterleaveFirstOnly s)
823− Stop -> Stop
824−
825− step gst (InterleaveSecondOnly st2) = do
826− r <- step2 gst st2
827− return $ case r of
828− Yield a s -> Yield a (InterleaveSecondOnly s)
829− Skip s -> Skip (InterleaveSecondOnly s)
830− Stop -> Stop
831−
832−{-# INLINE_NORMAL interleaveMin #-}
833−interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a
834−interleaveMin (Stream step1 state1) (Stream step2 state2) =
835− Stream step (InterleaveFirst state1 state2)
836−
837− where
838−
839− {-# INLINE_LATE step #-}
840− step gst (InterleaveFirst st1 st2) = do
841− r <- step1 gst st1
842− return $ case r of
843− Yield a s -> Yield a (InterleaveSecond s st2)
844− Skip s -> Skip (InterleaveFirst s st2)
845− Stop -> Stop
846−
847− step gst (InterleaveSecond st1 st2) = do
848− r <- step2 gst st2
849− return $ case r of
850− Yield a s -> Yield a (InterleaveFirst st1 s)
851− Skip s -> Skip (InterleaveSecond st1 s)
852− Stop -> Stop
853−
854− step _ (InterleaveFirstOnly _) = undefined
855− step _ (InterleaveSecondOnly _) = undefined
856−
857−{-# INLINE_NORMAL interleaveSuffix #-}
858−interleaveSuffix :: Monad m => Stream m a -> Stream m a -> Stream m a
859−interleaveSuffix (Stream step1 state1) (Stream step2 state2) =
860− Stream step (InterleaveFirst state1 state2)
861−
862− where
863−
864− {-# INLINE_LATE step #-}
865− step gst (InterleaveFirst st1 st2) = do
866− r <- step1 gst st1
867− return $ case r of
868− Yield a s -> Yield a (InterleaveSecond s st2)
869− Skip s -> Skip (InterleaveFirst s st2)
870− Stop -> Stop
871−
872− step gst (InterleaveSecond st1 st2) = do
873− r <- step2 gst st2
874− return $ case r of
875− Yield a s -> Yield a (InterleaveFirst st1 s)
876− Skip s -> Skip (InterleaveSecond st1 s)
877− Stop -> Skip (InterleaveFirstOnly st1)
878−
879− step gst (InterleaveFirstOnly st1) = do
880− r <- step1 gst st1
881− return $ case r of
882− Yield a s -> Yield a (InterleaveFirstOnly s)
883− Skip s -> Skip (InterleaveFirstOnly s)
884− Stop -> Stop
885−
886− step _ (InterleaveSecondOnly _) = undefined
887−
888−data InterleaveInfixState s1 s2 a
889− = InterleaveInfixFirst s1 s2
890− | InterleaveInfixSecondBuf s1 s2
891− | InterleaveInfixSecondYield s1 s2 a
892− | InterleaveInfixFirstYield s1 s2 a
893− | InterleaveInfixFirstOnly s1
894−
895−{-# INLINE_NORMAL interleaveInfix #-}
896−interleaveInfix :: Monad m => Stream m a -> Stream m a -> Stream m a
897−interleaveInfix (Stream step1 state1) (Stream step2 state2) =
898− Stream step (InterleaveInfixFirst state1 state2)
899−
900− where
901−
902− {-# INLINE_LATE step #-}
903− step gst (InterleaveInfixFirst st1 st2) = do
904− r <- step1 gst st1
905− return $ case r of
906− Yield a s -> Yield a (InterleaveInfixSecondBuf s st2)
907− Skip s -> Skip (InterleaveInfixFirst s st2)
908− Stop -> Stop
909−
910− step gst (InterleaveInfixSecondBuf st1 st2) = do
911− r <- step2 gst st2
912− return $ case r of
913− Yield a s -> Skip (InterleaveInfixSecondYield st1 s a)
914− Skip s -> Skip (InterleaveInfixSecondBuf st1 s)
915− Stop -> Skip (InterleaveInfixFirstOnly st1)
916−
917− step gst (InterleaveInfixSecondYield st1 st2 x) = do
918− r <- step1 gst st1
919− return $ case r of
920− Yield a s -> Yield x (InterleaveInfixFirstYield s st2 a)
921− Skip s -> Skip (InterleaveInfixSecondYield s st2 x)
922− Stop -> Stop
923−
924− step _ (InterleaveInfixFirstYield st1 st2 x) = do
925− return $ Yield x (InterleaveInfixSecondBuf st1 st2)
926−
927− step gst (InterleaveInfixFirstOnly st1) = do
928− r <- step1 gst st1
929− return $ case r of
930− Yield a s -> Yield a (InterleaveInfixFirstOnly s)
931− Skip s -> Skip (InterleaveInfixFirstOnly s)
932− Stop -> Stop
933−
934−{-# INLINE_NORMAL roundRobin #-}
935−roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a
936−roundRobin (Stream step1 state1) (Stream step2 state2) =
937− Stream step (InterleaveFirst state1 state2)
938−
939− where
940−
941− {-# INLINE_LATE step #-}
942− step gst (InterleaveFirst st1 st2) = do
943− r <- step1 gst st1
944− return $ case r of
945− Yield a s -> Yield a (InterleaveSecond s st2)
946− Skip s -> Skip (InterleaveSecond s st2)
947− Stop -> Skip (InterleaveSecondOnly st2)
948−
949− step gst (InterleaveSecond st1 st2) = do
950− r <- step2 gst st2
951− return $ case r of
952− Yield a s -> Yield a (InterleaveFirst st1 s)
953− Skip s -> Skip (InterleaveFirst st1 s)
954− Stop -> Skip (InterleaveFirstOnly st1)
955−
956− step gst (InterleaveSecondOnly st2) = do
957− r <- step2 gst st2
958− return $ case r of
959− Yield a s -> Yield a (InterleaveSecondOnly s)
960− Skip s -> Skip (InterleaveSecondOnly s)
961− Stop -> Stop
962−
963− step gst (InterleaveFirstOnly st1) = do
964− r <- step1 gst st1
965− return $ case r of
966− Yield a s -> Yield a (InterleaveFirstOnly s)
967− Skip s -> Skip (InterleaveFirstOnly s)
968− Stop -> Stop
969−
970−data ICUState s1 s2 i1 i2 =
971− ICUFirst s1 s2
972− | ICUSecond s1 s2
973− | ICUSecondOnly s2
974− | ICUFirstOnly s1
975− | ICUFirstInner s1 s2 i1
976− | ICUSecondInner s1 s2 i2
977− | ICUFirstOnlyInner s1 i1
978− | ICUSecondOnlyInner s2 i2
979−
980−-- | Interleave streams (full streams, not the elements) unfolded from two
981−-- input streams and concat. Stop when the first stream stops. If the second
982−-- stream ends before the first one then first stream still keeps running alone
983−-- without any interleaving with the second stream.
984−--
985−-- [a1, a2, ... an] [b1, b2 ...]
986−-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
987−-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
988−-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
989−--
990−{-# INLINE_NORMAL gintercalateSuffix #-}
991−gintercalateSuffix
992− :: Monad m
993− => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
994−gintercalateSuffix
995− (Unfold istep1 inject1) (Stream step1 state1)
996− (Unfold istep2 inject2) (Stream step2 state2) =
997− Stream step (ICUFirst state1 state2)
998−
999− where
1000−
1001− {-# INLINE_LATE step #-}
1002− step gst (ICUFirst s1 s2) = do
1003− r <- step1 (adaptState gst) s1
1004− case r of
1005− Yield a s -> do
1006− i <- inject1 a
1007− i `seq` return (Skip (ICUFirstInner s s2 i))
1008− Skip s -> return $ Skip (ICUFirst s s2)
1009− Stop -> return Stop
1010−
1011− step gst (ICUFirstOnly s1) = do
1012− r <- step1 (adaptState gst) s1
1013− case r of
1014− Yield a s -> do
1015− i <- inject1 a
1016− i `seq` return (Skip (ICUFirstOnlyInner s i))
1017− Skip s -> return $ Skip (ICUFirstOnly s)
1018− Stop -> return Stop
1019−
1020− step _ (ICUFirstInner s1 s2 i1) = do
1021− r <- istep1 i1
1022− return $ case r of
1023− Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
1024− Skip i' -> Skip (ICUFirstInner s1 s2 i')
1025− Stop -> Skip (ICUSecond s1 s2)
1026−
1027− step _ (ICUFirstOnlyInner s1 i1) = do
1028− r <- istep1 i1
1029− return $ case r of
1030− Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
1031− Skip i' -> Skip (ICUFirstOnlyInner s1 i')
1032− Stop -> Skip (ICUFirstOnly s1)
1033−
1034− step gst (ICUSecond s1 s2) = do
1035− r <- step2 (adaptState gst) s2
1036− case r of
1037− Yield a s -> do
1038− i <- inject2 a
1039− i `seq` return (Skip (ICUSecondInner s1 s i))
1040− Skip s -> return $ Skip (ICUSecond s1 s)
1041− Stop -> return $ Skip (ICUFirstOnly s1)
1042−
1043− step _ (ICUSecondInner s1 s2 i2) = do
1044− r <- istep2 i2
1045− return $ case r of
1046− Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
1047− Skip i' -> Skip (ICUSecondInner s1 s2 i')
1048− Stop -> Skip (ICUFirst s1 s2)
1049−
1050− step _ (ICUSecondOnly _s2) = undefined
1051− step _ (ICUSecondOnlyInner _s2 _i2) = undefined
1052−
1053−{-# ANN type InterposeSuffixState Fuse #-}
1054−data InterposeSuffixState s1 i1 =
1055− InterposeSuffixFirst s1
1056− -- | InterposeSuffixFirstYield s1 i1
1057− | InterposeSuffixFirstInner s1 i1
1058− | InterposeSuffixSecond s1
1059−
1060−-- Note that if an unfolded layer turns out to be nil we still emit the
1061−-- separator effect. An alternate behavior could be to emit the separator
1062−-- effect only if at least one element has been yielded by the unfolding.
1063−-- However, that becomes a bit complicated, so we have chosen the former
1064−-- behvaior for now.
1065−{-# INLINE_NORMAL interposeSuffix #-}
1066−interposeSuffix
1067− :: Monad m
1068− => m c -> Unfold m b c -> Stream m b -> Stream m c
1069−interposeSuffix
1070− action
1071− (Unfold istep1 inject1) (Stream step1 state1) =
1072− Stream step (InterposeSuffixFirst state1)
1073−
1074− where
1075−
1076− {-# INLINE_LATE step #-}
1077− step gst (InterposeSuffixFirst s1) = do
1078− r <- step1 (adaptState gst) s1
1079− case r of
1080− Yield a s -> do
1081− i <- inject1 a
1082− i `seq` return (Skip (InterposeSuffixFirstInner s i))
1083− -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
1084− Skip s -> return $ Skip (InterposeSuffixFirst s)
1085− Stop -> return Stop
1086−
1087− {-
1088− step _ (InterposeSuffixFirstYield s1 i1) = do
1089− r <- istep1 i1
1090− return $ case r of
1091− Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
1092− Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
1093− Stop -> Skip (InterposeSuffixFirst s1)
1094− -}
1095−
1096− step _ (InterposeSuffixFirstInner s1 i1) = do
1097− r <- istep1 i1
1098− return $ case r of
1099− Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
1100− Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
1101− Stop -> Skip (InterposeSuffixSecond s1)
1102−
1103− step _ (InterposeSuffixSecond s1) = do
1104− r <- action
1105− return $ Yield r (InterposeSuffixFirst s1)
1106−
1107−data ICALState s1 s2 i1 i2 a =
1108− ICALFirst s1 s2
1109− -- | ICALFirstYield s1 s2 i1
1110− | ICALFirstInner s1 s2 i1
1111− | ICALFirstOnly s1
1112− | ICALFirstOnlyInner s1 i1
1113− | ICALSecondInject s1 s2
1114− | ICALFirstInject s1 s2 i2
1115− -- | ICALFirstBuf s1 s2 i1 i2
1116− | ICALSecondInner s1 s2 i1 i2
1117− -- -- | ICALSecondInner s1 s2 i1 i2 a
1118− -- -- | ICALFirstResume s1 s2 i1 i2 a
1119−
1120−-- | Interleave streams (full streams, not the elements) unfolded from two
1121−-- input streams and concat. Stop when the first stream stops. If the second
1122−-- stream ends before the first one then first stream still keeps running alone
1123−-- without any interleaving with the second stream.
1124−--
1125−-- [a1, a2, ... an] [b1, b2 ...]
1126−-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
1127−-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
1128−-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
1129−--
1130−{-# INLINE_NORMAL gintercalate #-}
1131−gintercalate
1132− :: Monad m
1133− => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
1134−gintercalate
1135− (Unfold istep1 inject1) (Stream step1 state1)
1136− (Unfold istep2 inject2) (Stream step2 state2) =
1137− Stream step (ICALFirst state1 state2)
1138−
1139− where
1140−
1141− {-# INLINE_LATE step #-}
1142− step gst (ICALFirst s1 s2) = do
1143− r <- step1 (adaptState gst) s1
1144− case r of
1145− Yield a s -> do
1146− i <- inject1 a
1147− i `seq` return (Skip (ICALFirstInner s s2 i))
1148− -- i `seq` return (Skip (ICALFirstYield s s2 i))
1149− Skip s -> return $ Skip (ICALFirst s s2)
1150− Stop -> return Stop
1151−
1152− {-
1153− step _ (ICALFirstYield s1 s2 i1) = do
1154− r <- istep1 i1
1155− return $ case r of
1156− Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
1157− Skip i' -> Skip (ICALFirstYield s1 s2 i')
1158− Stop -> Skip (ICALFirst s1 s2)
1159− -}
1160−
1161− step _ (ICALFirstInner s1 s2 i1) = do
1162− r <- istep1 i1
1163− return $ case r of
1164− Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
1165− Skip i' -> Skip (ICALFirstInner s1 s2 i')
1166− Stop -> Skip (ICALSecondInject s1 s2)
1167−
1168− step gst (ICALFirstOnly s1) = do
1169− r <- step1 (adaptState gst) s1
1170− case r of
1171− Yield a s -> do
1172− i <- inject1 a
1173− i `seq` return (Skip (ICALFirstOnlyInner s i))
1174− Skip s -> return $ Skip (ICALFirstOnly s)
1175− Stop -> return Stop
1176−
1177− step _ (ICALFirstOnlyInner s1 i1) = do
1178− r <- istep1 i1
1179− return $ case r of
1180− Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
1181− Skip i' -> Skip (ICALFirstOnlyInner s1 i')
1182− Stop -> Skip (ICALFirstOnly s1)
1183−
1184− -- We inject the second stream even before checking if the first stream
1185− -- would yield any more elements. There is no clear choice whether we
1186− -- should do this before or after that. Doing it after may make the state
1187− -- machine a bit simpler though.
1188− step gst (ICALSecondInject s1 s2) = do
1189− r <- step2 (adaptState gst) s2
1190− case r of
1191− Yield a s -> do
1192− i <- inject2 a
1193− i `seq` return (Skip (ICALFirstInject s1 s i))
1194− Skip s -> return $ Skip (ICALSecondInject s1 s)
1195− Stop -> return $ Skip (ICALFirstOnly s1)
1196−
1197− step gst (ICALFirstInject s1 s2 i2) = do
1198− r <- step1 (adaptState gst) s1
1199− case r of
1200− Yield a s -> do
1201− i <- inject1 a
1202− i `seq` return (Skip (ICALSecondInner s s2 i i2))
1203− -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
1204− Skip s -> return $ Skip (ICALFirstInject s s2 i2)
1205− Stop -> return Stop
1206−
1207− {-
1208− step _ (ICALFirstBuf s1 s2 i1 i2) = do
1209− r <- istep1 i1
1210− return $ case r of
1211− Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
1212− Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
1213− Stop -> Stop
1214−
1215− step _ (ICALSecondInner s1 s2 i1 i2 v) = do
1216− r <- istep2 i2
1217− return $ case r of
1218− Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
1219− Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
1220− Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
1221− -}
1222−
1223− step _ (ICALSecondInner s1 s2 i1 i2) = do
1224− r <- istep2 i2
1225− return $ case r of
1226− Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
1227− Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
1228− Stop -> Skip (ICALFirstInner s1 s2 i1)
1229− -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
1230−
1231− {-
1232− step _ (ICALFirstResume s1 s2 i1 i2 x) = do
1233− return $ Yield x (ICALFirstInner s1 s2 i1 i2)
1234− -}
1235−
1236−{-# ANN type InterposeState Fuse #-}
1237−data InterposeState s1 i1 a =
1238− InterposeFirst s1
1239− -- | InterposeFirstYield s1 i1
1240− | InterposeFirstInner s1 i1
1241− | InterposeFirstInject s1
1242− -- | InterposeFirstBuf s1 i1
1243− | InterposeSecondYield s1 i1
1244− -- -- | InterposeSecondYield s1 i1 a
1245− -- -- | InterposeFirstResume s1 i1 a
1246−
1247−-- Note that this only interposes the pure values, we may run many effects to
1248−-- generate those values as some effects may not generate anything (Skip).
1249−{-# INLINE_NORMAL interpose #-}
1250−interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
1251−interpose
1252− action
1253− (Unfold istep1 inject1) (Stream step1 state1) =
1254− Stream step (InterposeFirst state1)
1255−
1256− where
1257−
1258− {-# INLINE_LATE step #-}
1259− step gst (InterposeFirst s1) = do
1260− r <- step1 (adaptState gst) s1
1261− case r of
1262− Yield a s -> do
1263− i <- inject1 a
1264− i `seq` return (Skip (InterposeFirstInner s i))
1265− -- i `seq` return (Skip (InterposeFirstYield s i))
1266− Skip s -> return $ Skip (InterposeFirst s)
1267− Stop -> return Stop
1268−
1269− {-
1270− step _ (InterposeFirstYield s1 i1) = do
1271− r <- istep1 i1
1272− return $ case r of
1273− Yield x i' -> Yield x (InterposeFirstInner s1 i')
1274− Skip i' -> Skip (InterposeFirstYield s1 i')
1275− Stop -> Skip (InterposeFirst s1)
1276− -}
1277−
1278− step _ (InterposeFirstInner s1 i1) = do
1279− r <- istep1 i1
1280− return $ case r of
1281− Yield x i' -> Yield x (InterposeFirstInner s1 i')
1282− Skip i' -> Skip (InterposeFirstInner s1 i')
1283− Stop -> Skip (InterposeFirstInject s1)
1284−
1285− step gst (InterposeFirstInject s1) = do
1286− r <- step1 (adaptState gst) s1
1287− case r of
1288− Yield a s -> do
1289− i <- inject1 a
1290− -- i `seq` return (Skip (InterposeFirstBuf s i))
1291− i `seq` return (Skip (InterposeSecondYield s i))
1292− Skip s -> return $ Skip (InterposeFirstInject s)
1293− Stop -> return Stop
1294−
1295− {-
1296− step _ (InterposeFirstBuf s1 i1) = do
1297− r <- istep1 i1
1298− return $ case r of
1299− Yield x i' -> Skip (InterposeSecondYield s1 i' x)
1300− Skip i' -> Skip (InterposeFirstBuf s1 i')
1301− Stop -> Stop
1302− -}
1303−
1304− {-
1305− step _ (InterposeSecondYield s1 i1 v) = do
1306− r <- action
1307− return $ Yield r (InterposeFirstResume s1 i1 v)
1308− -}
1309− step _ (InterposeSecondYield s1 i1) = do
1310− r <- action
1311− return $ Yield r (InterposeFirstInner s1 i1)
1312−
1313− {-
1314− step _ (InterposeFirstResume s1 i1 v) = do
1315− return $ Yield v (InterposeFirstInner s1 i1)
1316− -}
1317−
1318602------------------------------------------------------------------------------
1319603-- Exceptions
1320604------------------------------------------------------------------------------