7a8227e · streamly · git.iko.soy

git.iko.soy/streamly

implement Array intercalate in terms of stream op

We do not need a custom Array unlines function anymore. We can just use stream operations to do that job in general.

commit7a8227eae45bb4f9dd62a9cb5ebda228391bc1cd authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 12 Sep 2019, 23:40 (7 years ago) parent7cf3784

7 files changed+163−23

compare with master

unifiedsplit

src/Streamly/Data/String.hs+16−3 src/Streamly/Internal.hs+1 src/Streamly/Memory/Array.hs−1 src/Streamly/Memory/ArrayStream.hs+13−8 src/Streamly/Prelude/Internal.hs+18−11 src/Streamly/Streams/StreamD.hs+84 src/Streamly/Unfold.hs+31

src/Streamly/Data/String.hs+16−3

@@ -70,6 +70,7 @@ module Streamly.Data.String

7070 , stripStart

7171 , foldLines

7272 , foldWords

73+ , unfoldLines

7374

7475 -- * Streams of Strings

7576 , lines

@@ -87,10 +88,13 @@ import Streamly (IsStream, MonadAsync)

8788import Prelude hiding (String, lines, words, unlines, unwords)

8889import Streamly.Fold (Fold)

8990import Streamly.Memory.Array (Array)

91+import Streamly.Unfold (Unfold)

9092

91−import qualified Streamly.Prelude as S

93+import qualified Streamly.Prelude.Internal as S

9294import qualified Streamly.Memory.Array as A

95+import qualified Streamly.Memory.Array.Types as A

9396import qualified Streamly.Streams.StreamD as D

97+import qualified Streamly.Unfold as UF

9498

9599-- type String = List Char

96100

@@ -242,6 +246,15 @@ lines = foldLines A.write

242246words :: (MonadIO m, IsStream t) => t m Char -> t m (Array Char)

243247words = foldWords A.write

244248

249+-- | Unfold a stream to character streams using the supplied 'Unfold'

250+-- and concat the results suffixing a newline character @\\n@ to each stream.

251+--

252+-- > unfoldLines = S.intercalateSuffix UF.singleton '\n'

253+--

254+{-# INLINE unfoldLines #-}

255+unfoldLines :: (MonadIO m, IsStream t) => Unfold m a Char -> t m a -> t m Char

256+unfoldLines unf = S.intercalateSuffix UF.singleton '\n' unf

257+

245258-- | Flattens the stream of @Array Char@, after appending a terminating

246259-- newline to each string.

247260--

@@ -250,14 +263,14 @@ words = foldWords A.write

250263-- >>> S.toList $ unlines $ S.fromList ["lines", "this", "string"]

251264-- "lines\nthis\nstring\n"

252265--

253−-- > unlines = A.concatWithSuffix '\n'

266+-- > unlines = unfoldLines A.readU

254267--

255268-- Note that, in general

256269--

257270-- > unlines . lines /= id

258271{-# INLINE unlines #-}

259272unlines :: (MonadIO m, IsStream t) => t m (Array Char) -> t m Char

260−unlines = A.concatWithSuffix '\n'

273+unlines = unfoldLines A.readU

261274

262275-- | Flattens the stream of @Array Char@, after appending a separating

263276-- space to each string.

src/Streamly/Internal.hs+1

@@ -66,6 +66,7 @@ module Streamly.Internal

6666 , concatMapU

6767 , concatUnfoldInterleave

6868 , concatUnfoldRoundrobin

69+ , intercalateSuffix

6970

7071 -- * Streamly.Fold Experimental Exports

7172 , Fold (..)

src/Streamly/Memory/Array.hs−1

@@ -128,7 +128,6 @@ module Streamly.Memory.Array

128128 -- ** Elimination

129129 , AS.concat

130130 -- , AS.concatRev

131− , AS.concatWithSuffix

132131 , AS.toArray

133132

134133 -- ** Transformation

src/Streamly/Memory/ArrayStream.hs+13−8

@@ -26,7 +26,8 @@ module Streamly.Memory.ArrayStream

2626 -- * Flattening to elements

2727 , concat

2828 , concatRev

29− , concatWithSuffix

29+ , interposeSuffix

30+ , intercalateSuffix

3031

3132 -- * Transformation

3233 , splitOn

@@ -49,8 +50,9 @@ import Streamly.Memory.Array.Types (Array(..), length)

4950import Streamly.Streams.Serial (SerialT)

5051import Streamly.Streams.StreamK.Type (IsStream)

5152

53+import qualified Streamly.Unfold as UF

5254import qualified Streamly.Memory.Array.Types as A

53−import qualified Streamly.Prelude as S

55+import qualified Streamly.Prelude.Internal as S

5456import qualified Streamly.Streams.StreamD as D

5557import qualified Streamly.Streams.Prelude as P

5658

@@ -82,17 +84,20 @@ concat m = D.fromStreamD $ D.concatMapU A.readU (D.toStreamD m)

8284concatRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

8385concatRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)

8486

85−-- XXX use an Array instead as separator? Or use a separate unlinesArraysBySeq

86−-- API for that?

87−--

87+{-# INLINE intercalateSuffix #-}

88+intercalateSuffix :: (MonadIO m, IsStream t, Storable a)

89+ => Array a -> t m (Array a) -> t m a

90+intercalateSuffix arr = S.intercalateSuffix A.readU arr A.readU

91+

8892-- | Flatten a stream of arrays appending the given element after each

8993-- array.

9094--

9195-- @since 0.7.0

92−{-# INLINE concatWithSuffix #-}

93−concatWithSuffix :: (MonadIO m, IsStream t, Storable a)

96+{-# INLINE interposeSuffix #-}

97+interposeSuffix :: (MonadIO m, IsStream t, Storable a)

9498 => a -> t m (Array a) -> t m a

95−concatWithSuffix x = D.fromStreamD . A.unlines x . D.toStreamD

99+-- interposeSuffix x = D.fromStreamD . A.unlines x . D.toStreamD

100+interposeSuffix x = S.intercalateSuffix UF.singleton x A.readU

96101

97102-- | Split a stream of arrays on a given separator byte, dropping the separator

98103-- and coalescing all the arrays between two separators into a single array.

src/Streamly/Prelude/Internal.hs+18−11

@@ -245,8 +245,8 @@ module Streamly.Prelude.Internal

245245 , concatUnfoldRoundrobin

246246 , concatMap

247247 , concatMapWith

248− -- , interposeBy

249248 -- , intercalate

249+ , intercalateSuffix

250250

251251 -- -- ** Breaking

252252

@@ -2210,22 +2210,29 @@ concatUnfoldRoundrobin u m =

22102210 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)

22112211

22122212{-

2213−-- | A generalization of insertBy to inserting sequences.

2214−interposeBy :: (IsStream t, Monad m)

2215− => (a -> a -> Ordering) -> t m a -> t m a -> t m a

2216−interposeBy = undefined

2217−

22182213-- | Insert a stream between segements of streams and flatten.

22192214intercalate :: (IsStream t, MonadAsync m)

22202215 => SerialT Identity b -> (a -> t m b) -> t m a -> t m b

22212216intercalate = undefined

2222−

2223−-- | Insert a stream after segements of streams and flatten.

2224−intercalatePost :: (IsStream t, MonadAsync m)

2225− => SerialT Identity b -> (a -> t m b) -> t m a -> t m b

2226−intercalatePost = undefined

22272217-}

22282218

2219+-- | @intercalateSuffix genSuffix seed gen stream@ generates streams from seeds

2220+-- in the input @stream@ using the generator @gen@ and concatenates them

2221+-- after suffixing the stream generated by @genSuffix@ using @seed@.

2222+--

2223+-- For example to insert CRLF in a stream of character strings:

2224+--

2225+-- > unlines = intercalateSuffix UF.fromList "\r\n" UF.fromList

2226+--

2227+{-# INLINE intercalateSuffix #-}

2228+intercalateSuffix

2229+ :: (IsStream t, Monad m)

2230+ => Unfold m a c -> a -> Unfold m b c -> t m b -> t m c

2231+intercalateSuffix suffix seed unf str =

2232+ D.fromStreamD $ D.interleaveFstThenConcat

2233+ unf (D.toStreamD str)

2234+ suffix (D.repeat seed)

2235+

22292236------------------------------------------------------------------------------

22302237-- Grouping/Splitting

22312238------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs+84

@@ -147,6 +147,7 @@ module Streamly.Streams.StreamD

147147 , interleaveMin

148148 , interleaveFst

149149 , roundRobin -- interleaveFair?/ParallelFair

150+ , interleaveFstThenConcat

150151

151152 -- ** Grouping

152153 , groupsOf

@@ -2025,6 +2026,89 @@ roundRobin (Stream step1 state1) (Stream step2 state2) =

20252026 Skip s -> Skip (InterleaveFirstOnly s)

20262027 Stop -> Stop

20272028

2029+data ICUState s1 s2 i1 i2 =

2030+ ICUFirst s1 s2

2031+ | ICUSecond s1 s2

2032+ | ICUSecondOnly s2

2033+ | ICUFirstOnly s1

2034+ | ICUFirstInner s1 s2 i1

2035+ | ICUSecondInner s1 s2 i2

2036+ | ICUFirstOnlyInner s1 i1

2037+ | ICUSecondOnlyInner s2 i2

2038+

2039+-- | Interleave streams (full streams, not the elements) unfolded from two

2040+-- input streams and concat. Stop when the first stream stops. If the second

2041+-- stream ends before the first one then first stream still keeps running alone

2042+-- without any interleaving with the second stream.

2043+--

2044+-- [a1, a2, ... an] [b1, b2 ...]

2045+-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]

2046+-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]

2047+-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]

2048+--

2049+{-# INLINE_NORMAL interleaveFstThenConcat #-}

2050+interleaveFstThenConcat

2051+ :: Monad m

2052+ => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c

2053+interleaveFstThenConcat

2054+ (Unfold istep1 inject1) (Stream step1 state1)

2055+ (Unfold istep2 inject2) (Stream step2 state2) =

2056+ Stream step (ICUFirst state1 state2)

2057+

2058+ where

2059+

2060+ {-# INLINE_LATE step #-}

2061+ step gst (ICUFirst s1 s2) = do

2062+ r <- step1 (adaptState gst) s1

2063+ case r of

2064+ Yield a s -> do

2065+ i <- inject1 a

2066+ i `seq` return (Skip (ICUFirstInner s s2 i))

2067+ Skip s -> return $ Skip (ICUFirst s s2)

2068+ Stop -> return Stop

2069+

2070+ step gst (ICUFirstOnly s1) = do

2071+ r <- step1 (adaptState gst) s1

2072+ case r of

2073+ Yield a s -> do

2074+ i <- inject1 a

2075+ i `seq` return (Skip (ICUFirstOnlyInner s i))

2076+ Skip s -> return $ Skip (ICUFirstOnly s)

2077+ Stop -> return Stop

2078+

2079+ step _ (ICUFirstInner s1 s2 i1) = do

2080+ r <- istep1 i1

2081+ return $ case r of

2082+ Yield x i' -> Yield x (ICUFirstInner s1 s2 i')

2083+ Skip i' -> Skip (ICUFirstInner s1 s2 i')

2084+ Stop -> Skip (ICUSecond s1 s2)

2085+

2086+ step _ (ICUFirstOnlyInner s1 i1) = do

2087+ r <- istep1 i1

2088+ return $ case r of

2089+ Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')

2090+ Skip i' -> Skip (ICUFirstOnlyInner s1 i')

2091+ Stop -> Skip (ICUFirstOnly s1)

2092+

2093+ step gst (ICUSecond s1 s2) = do

2094+ r <- step2 (adaptState gst) s2

2095+ case r of

2096+ Yield a s -> do

2097+ i <- inject2 a

2098+ i `seq` return (Skip (ICUSecondInner s1 s i))

2099+ Skip s -> return $ Skip (ICUSecond s1 s)

2100+ Stop -> return $ Skip (ICUFirstOnly s1)

2101+

2102+ step _ (ICUSecondInner s1 s2 i2) = do

2103+ r <- istep2 i2

2104+ return $ case r of

2105+ Yield x i' -> Yield x (ICUSecondInner s1 s2 i')

2106+ Skip i' -> Skip (ICUSecondInner s1 s2 i')

2107+ Stop -> Skip (ICUFirst s1 s2)

2108+

2109+ step _ (ICUSecondOnly _s2) = undefined

2110+ step _ (ICUSecondOnlyInner _s2 _i2) = undefined

2111+

20282112------------------------------------------------------------------------------

20292113-- Exceptions

20302114------------------------------------------------------------------------------

src/Streamly/Unfold.hs+31

@@ -57,7 +57,9 @@ module Streamly.Unfold

5757 , unfold

5858

5959 -- ** Unfolds

60+ , singleton

6061 , replicateM

62+ , fromList

6163 )

6264where

6365

@@ -71,6 +73,8 @@ import Streamly.Unfold.Types (Unfold(..))

7173-- Running unfolds

7274-------------------------------------------------------------------------------

7375

76+-- | Convert an 'Unfold' into a 'Stream' by supplying it a seed.

77+--

7478{-# INLINE_NORMAL unfold #-}

7579unfold :: Monad m => a -> Unfold m a b -> Stream m b

7680unfold seed (Unfold ustep inject) = Stream step Nothing

@@ -88,12 +92,39 @@ unfold seed (Unfold ustep inject) = Stream step Nothing

8892-- Unfolds

8993-------------------------------------------------------------------------------

9094

95+-- | Identity unfold. Generates a singleton stream with the seed as the only

96+-- element in the stream.

97+--

98+-- > singleton = replicateM 1

99+--

100+{-# INLINE singleton #-}

101+singleton :: Monad m => Unfold m a a

102+singleton = Unfold step inject

103+ where

104+ inject x = return $ Just x

105+ {-# INLINE_LATE step #-}

106+ step (Just x) = return $ Yield x Nothing

107+ step Nothing = return Stop

108+

109+-- | Generates a stream replicating the seed @n@ times.

110+--

91111{-# INLINE replicateM #-}

92112replicateM :: Monad m => Int -> Unfold m a a

93113replicateM n = Unfold step inject

94114 where

95115 inject x = return (x, n)

116+ {-# INLINE_LATE step #-}

96117 step (x, i) = return $

97118 if i <= 0

98119 then Stop

99120 else Yield x (x, (i - 1))

121+

122+-- | Convert a list of pure values to a 'Stream'

123+{-# INLINE_LATE fromList #-}

124+fromList :: Monad m => Unfold m [a] a

125+fromList = Unfold step inject

126+ where

127+ inject x = return x

128+ {-# INLINE_LATE step #-}

129+ step (x:xs) = return $ Yield x xs

130+ step [] = return Stop