-
Notifications
You must be signed in to change notification settings - Fork 0
/
Spec.hs
192 lines (176 loc) · 7.17 KB
/
Spec.hs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FunctionalDependencies #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TypeFamilies #-}
module Main where
import Control.Applicative
import Control.Concurrent
import Control.Concurrent.STM
import Control.Lens
import Control.Monad
import Control.Monad.Morph as M
import Control.Monad.State.Strict
import Control.Monad.Trans.Identity
import Data.Foldable
import Data.Maybe
import qualified Pipes as P
import qualified Pipes.Concurrent as PC
import qualified Pipes.Fluid.Impulse as PF
import qualified Pipes.Fluid.ImpulseIO as PF
import qualified Pipes.Fluid.Simultaneous as PF
import qualified Pipes.Misc.Concurrent as PM
import qualified Pipes.Misc.State.Strict as PM
import qualified Pipes.Prelude as PP
import Test.Hspec
data Model = Model
{ modelCounter1 :: Int
, modelCounter2 :: Int
} deriving (Eq, Show, Ord)
makeFields ''Model
data1 :: [Int]
data1 = [1, 2..9]
data2 :: [Int]
data2 = [10, 20..70]
sig1 :: Monad m => P.Producer Int m ()
sig1 = traverse_ P.yield data1
sig2 :: Monad m => P.Producer Int m ()
sig2 = traverse_ P.yield data2
delay :: Int -> P.Pipe a a IO ()
delay i = P.for P.cat $ \a -> do
lift $ threadDelay i
P.yield a
-- | For the Simultaneous tests, not all the input will be consumed
-- so quitEarly will close the mailbox when an of the feeders
-- have finished feeding.
testSig' :: Bool -> (P.Producer Int STM () -> P.Producer Int STM () -> IO a) -> IO a
testSig' quitEarly f = do
(o1, i1, q1) <- PC.spawn' $ PC.bounded 1
(o2, i2, q2) <- PC.spawn' $ PC.bounded 1
feederFinished1 <- newEmptyMVar
feederFinished2 <- newEmptyMVar
-- make time delayed signals
void $ forkIO $ do
P.runEffect $ sig1 P.>-> delay 30 P.>-> hoist atomically (PM.toOutputSTM o1)
when quitEarly $ atomically $ do
-- wait for input to be empty
r <- PC.recv i1 <|> pure Nothing
case r of
Nothing -> q1 *> q2
Just _ -> retry
putMVar feederFinished1 ()
void $ forkIO $ do
P.runEffect $ sig2 P.>-> delay 50 P.>-> hoist atomically (PM.toOutputSTM o2)
when quitEarly $ atomically $ do
-- wait for input to be empty
r <- PC.recv i2 <|> pure Nothing
case r of
Nothing -> q1 *> q2
Just _ -> retry
putMVar feederFinished2 ()
-- thread to destroy PC Input when both feeders have finished
void $ forkIO $ do
takeMVar feederFinished1
takeMVar feederFinished2
atomically $ q1 *> q2
-- run the test function
f (PM.fromInputSTM i1) (PM.fromInputSTM i2)
main :: IO ()
main = do
hspec $ do
describe "Simultaneous" $ do
it "only yield a value when both producers yields a value" $ do
xs <-
testSig' True $ \as bs ->
PP.toListM $
hoist atomically $
PF.simultaneously $
(\a b -> (a, b, a + b)) <$> PF.Simultaneous as <*> PF.Simultaneous bs
xs `shouldBe` (\(a, b) -> (a, b, a + b)) <$> zip data1 data2
describe "Impulse" $ do
it "Impulse STM: yield a value whenever any producer yields a value" $ do
xs <-
testSig' False $ \as bs ->
PP.toListM $
hoist atomically $
PF.impulsively $
(\a b -> (a, b, a + b)) <$> PF.Impulse as <*> PF.Impulse bs
xs `shouldSatisfy` isBigger
it "Impulse IdentityT STM: impulsively yields under 't STM'" $ do
xs <-
testSig' False $ \as bs ->
runIdentityT $
PP.toListM $
hoist (hoist atomically) $
PF.impulsively $
(\a b -> (a, b, a + b)) <$> (PF.Impulse $ hoist lift as) <*>
(PF.Impulse $ hoist lift bs)
xs `shouldSatisfy` isBigger
it "Impulse StateT STM: impulsively yields under 'StateT STM'" $ do
xs <-
testSig' False $ \as bs ->
(`evalStateT` (Model 0 0)) $
PP.toListM $
(hoist (hoist atomically) $
PF.impulsively $
(\a b -> (a, b, a + b)) <$>
(PF.Impulse $ hoist lift as P.>-> PM.store id counter1) <*>
(PF.Impulse $ hoist lift bs P.>-> PM.store id counter2)) P.>->
PM.restore id
xs `shouldSatisfy` isBigger
it "Impulse Merge: yield a Left/Right value depending on which producer yields a value" $ do
xs <-
testSig' False $ \as bs ->
PP.toListM $
hoist atomically $
PF.impulsively $
PF.Impulse as `PF.merge`PF.Impulse bs
xs `shouldSatisfy` isDifferent
describe "ImpulseIO" $ do
it "Impulse IO: impulsively yield under IO using lifted-async" $ do
xs <-
testSig' False $ \as bs ->
PP.toListM $
PF.impulsivelyIO $
(\a b -> (a, b, a + b)) <$> (PF.ImpulseIO $ hoist atomically as) <*>
(PF.ImpulseIO $ hoist atomically bs)
xs `shouldSatisfy` isBigger
it "Impulse IdentityT IO: impulsively yield under 't IO' using lifted-async" $ do
xs <-
testSig' False $ \as bs ->
runIdentityT $
PP.toListM $
PF.impulsivelyIO $
(\a b -> (a, b, a + b)) <$> (PF.ImpulseIO $ hoist (lift . atomically) as) <*>
(PF.ImpulseIO $ hoist (lift . atomically) bs)
xs `shouldSatisfy` isBigger
it "\nImpulse StateT IO is unsafe (lift-async detect this as a compile error)" $ do
pure () `shouldReturn` ()
it "ImpulseIO Merge: yield a Left/Right value depending on which producer yields a value" $ do
xs <-
testSig' False $ \as bs ->
PP.toListM $
PF.impulsivelyIO $
(PF.ImpulseIO $ hoist atomically as) `PF.merge` (PF.ImpulseIO $ hoist atomically bs)
xs `shouldSatisfy` isDifferent
isBigger :: Ord a => [a] -> Bool
isBigger = isJust . foldl' go (Just Nothing)
where
go p a =
case p of
Nothing -> Nothing
Just Nothing -> Just (Just a)
Just (Just p')
| p' == a -> Nothing
| p' < a -> Just (Just a)
| otherwise -> Nothing
isDifferent :: Eq a => [a] -> Bool
isDifferent = isJust . foldl' go (Just Nothing)
where
go p a =
case p of
Nothing -> Nothing
Just Nothing -> Just (Just a)
Just (Just p')
| p' == a -> Nothing
| otherwise -> Just (Just a)