@@ -1,3 +1,37 @@
1+module Control.Distributed.Process ( expect ) where
2+
3+import Data.ByteString.Lazy (ByteString)
4+import Data.Typeable (Typeable(typeOf))
5+import Data.Binary (decode)
6+import Control.Monad.Reader (MonadReader(..), ReaderT, liftIO)
7+import Control.Applicative ((<$>))
8+import Control.Distributed.Process.Internal.CQueue ( CQueue
9+ , dequeueMatching
10+ )
11+import Control.Distributed.Process.Serializable ( Serializable
12+ , Fingerprint
13+ , encodeFingerprint
14+ , fingerprint
15+ )
16+
17+data Message = Message { messageFingerprint :: Fingerprint
18+ , messageEncoding :: ByteString
19+ }
20+
21+data ProcessState = ProcessState { processQueue :: CQueue Message }
22+
23+newtype Process a = Process (ReaderT ProcessState IO a)
24+ deriving (Monad)
25+
26+expect :: forall a. Serializable a => Process a
27+expect = Process $ do
28+ queue <- processQueue <$> ask
29+ let fp = fingerprint (undefined :: a)
30+ msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)
31+ return (decode . messageEncoding $ msg)
32+
33+
34+{-
135{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}
236
337module Control.Distributed.Process (
@@ -198,33 +232,5 @@ monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
198232
199233-}
200234
201−-- Concurrent queue for single reader, single writer
202235--
203−data CQueue a = CQueue (MVar [a]) -- arrived
204− (Chan a) -- incomming
205−
206−newCQueue :: IO (CQueue a)
207−newCQueue = do
208− arrived <- newMVar []
209− incomming <- newChan
210− return (CQueue arrived incomming)
211−
212−enqueue :: CQueue a -> a -> IO ()
213−enqueue (CQueue _arrived incomming) a = writeChan incomming a
214−
215−dequeueMatching :: forall a. CQueue a -> (a -> Bool) -> IO a
216−dequeueMatching (CQueue arrived incomming) matches = do
217− modifyMVar arrived (checkArrived [])
218− where
219− checkArrived :: [a] -> [a] -> IO ([a], a)
220− checkArrived xs' [] = checkIncomming xs'
221− checkArrived xs' (x:xs)
222− | matches x = return (reverse xs' ++ xs, x)
223− | otherwise = checkArrived (x:xs') xs
224−
225− checkIncomming :: [a] -> IO ([a], a)
226− checkIncomming xs' = do
227− x <- readChan incomming
228− if matches x
229− then return (reverse xs', x)
230− else checkIncomming (x:xs')
236+-}