@@ -14,43 +14,100 @@ results together and consume the resulting stream concurrently.
1414First, distribute an item to multiple effectful consumer concurrently, for
1515example:
1616
17−```haskell
17+- Create a stream of actions.
18+- Sequence the actions concurrently.
19+ - Note that we respect the order of input. (`Config.ordered True`)
20+
21+```haskell ghci
22+{-# LANGUAGE FlexibleContexts #-}
23+
1824import Data.Function ((&))
19−import qualified Streamly.Prelude as Stream
25+import qualified Streamly.Data.Fold as Fold
26+import qualified Streamly.Data.Stream as Stream
27+import qualified Streamly.Data.Stream.Concurrent as Concur
2028
2129f1 x =
22− Prelude.map ($ x) [return . (+ 1), return . (+ 2)]
23− & Stream.fromListM
24− & Stream.fromAhead
25− & Stream.toList
30+ Prelude.map ($ x) [return . (+ 1), return . (+ 2)] -- List of actions
31+ & Stream.fromList -- Stream of actions
32+ & Concur.sequenceWith (Concur.ordered True) -- Sequence the actions concurrently
33+ & Stream.fold Fold.toList -- Fold the resulting stream into a list
2634```
2735
2836Alternatively, if you want to concurrently zip different types of outputs, you
29−can use `Applicative` composition:
37+can use `zipWith`:
38+
39+- Create 2 streams of actions.
40+- Sequence the actions concurrently.
41+- Zip the resulting streams.
42+
43+```haskell ghci
44+import Data.Maybe (fromJust)
45+
46+f2 x =
47+ let actionStream1 = Stream.fromPure (return $ show x)
48+ actionStream2 = Stream.fromPure (return $ x + 1)
49+ actionStream3 = Stream.fromPure (return $ fromIntegral x / 2)
50+ in Stream.zipWith (\a (b, c) -> (a, b, c))
51+ (Concur.sequence actionStream1)
52+ (Stream.zipWith (,)
53+ (Concur.sequence actionStream2)
54+ (Concur.sequence actionStream3))
55+ & Stream.fold Fold.one
56+ & fmap fromJust
57+```
58+
59+Alternatively, generate a type with concurrent zipping property to use the
60+`Applicative` instance,
3061
3162```haskell
63+{-# LANGUAGE UndecidableInstances #-}
64+{-# LANGUAGE FlexibleInstances #-}
65+{-# LANGUAGE MultiParamTypeClasses #-}
66+{-# LANGUAGE TemplateHaskell #-}
67+
68+import Control.Monad.Catch (MonadThrow(..))
69+import Control.Monad.IO.Class (MonadIO(..))
70+import Control.Monad.Reader.Class (MonadReader(..))
71+import Control.Monad.Trans.Class (MonadTrans(..))
72+import Data.Function ((&))
3273import Data.Maybe (fromJust)
74+import Streamly.Data.Stream (Stream)
75+import Streamly.Data.Stream.Concurrent (MonadAsync)
76+
77+import qualified Streamly.Data.Fold as Fold
78+import qualified Streamly.Data.Stream
79+import qualified Streamly.Data.Stream as Stream
80+import qualified Streamly.Data.Stream.Concurrent
81+import qualified Streamly.Data.Stream.Concurrent as Concur
82+import qualified Streamly.Internal.Data.Stream.TypeGen as StreamTypeGen
83+
84+applyConcurrent :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
85+applyConcurrent = Concur.zipWith ($)
86+
87+$(StreamTypeGen.mkZippingType "ZipConcurrent" "applyConcurrent" True)
88+
3389f2 x =
34− (,,)
35− <$> Stream.fromEffect (return $ show x)
36− <*> Stream.fromEffect (return $ x + 1)
37− <*> Stream.fromEffect (return $ fromIntegral x / 2)
38− & Stream.fromAsync
39− & Stream.head
40− & fmap fromJust
90+ let actionStream1 = Stream.fromEffect (return $ show x)
91+ actionStream2 = Stream.fromEffect (return $ x + 1)
92+ actionStream3 = Stream.fromEffect (return $ fromIntegral x / 2)
93+ in ((,,,)
94+ <$> actionStream1
95+ <*> actionStream2
96+ <*> actionStream3)
97+ & Stream.fold Fold.one
98+ & fmap fromJust
4199```
42100
43101Then apply the function `f` concurrently to your input stream:
44102
45−```haskell
103+```haskell ghci
46104g f xs =
47105 Stream.fromList xs
48− & Stream.mapM f
49− & Stream.fromAhead
50− & Stream.toList
106+ & Concur.mapMWith (Concur.ordered True) f -- Map the actions concurrently
107+ & Stream.fold Fold.toList
51108```
52109
53−```
110+```haskell docspec
54111>>> g f1 [1,2,3,4::Int]
55112[[2,3],[3,4],[4,5],[5,6]]
56113
@@ -60,11 +117,12 @@ g f xs =
60117
61118## Sliding Window
62119
63−```
120+```haskell docspec
121+>>> import qualified Streamly.Internal.Data.Array as Array
64122>>> :{
65123 Stream.fromList [1,2,3,4,5::Int]
66124& Stream.scan (Array.writeLastN 2)
67−& Stream.toList
125+& Stream.fold Fold.toList
68126:}
69−[[],[1],[1,2],[2,3],[3,4],[4,5]]
127+[fromList [],fromList [1],fromList [1,2],fromList [2,3],fromList [3,4],fromList [4,5]]
70128```