@@ -54,6 +54,7 @@ module Control.Distributed.Process.Internal.Primitives
5454 , finally
5555 -- * Auxiliary API
5656 , expectTimeout
57+ , receiveChanTimeout
5758 , spawnAsync
5859 , linkNode
5960 , linkPort
@@ -74,6 +75,7 @@ import Data.Binary (decode)
7475import Data.Time.Clock (getCurrentTime)
7576import Data.Time.Format (formatTime)
7677import System.Locale (defaultTimeLocale)
78+import System.Timeout (timeout)
7779import Control.Monad.Reader (ask)
7880import Control.Monad.IO.Class (MonadIO, liftIO)
7981import Control.Applicative ((<$>))
@@ -188,7 +190,7 @@ newChan = do
188190 )
189191 where
190192 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
191− finalizer processState lcid = modifyMVar_ processState $
193+ finalizer st lcid = modifyMVar_ st $
192194 return . (typedChannelWithId lcid ^= Nothing)
193195
194196-- | Send a message on a typed channel
@@ -204,18 +206,29 @@ sendChan (SendPort cid) msg = do
204206-- | Wait for a message on a typed channel
205207receiveChan :: Serializable a => ReceivePort a -> Process a
206208receiveChan = liftIO . atomically . receiveSTM
207− where
208− receiveSTM :: ReceivePort a -> STM a
209− receiveSTM (ReceivePortSingle c) =
210− readTQueue c
211− receiveSTM (ReceivePortBiased ps) =
212− foldr1 orElse (map receiveSTM ps)
213− receiveSTM (ReceivePortRR psVar) = do
214− ps <- readTVar psVar
215− a <- foldr1 orElse (map receiveSTM ps)
216− writeTVar psVar (rotate ps)
217− return a
218209
210+-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
211+-- non-blocking check for a message.
212+receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
213+receiveChanTimeout 0 ch = liftIO . atomically $
214+ (Just <$> receiveSTM ch) `orElse` return Nothing
215+receiveChanTimeout n ch = liftIO . timeout n . atomically $
216+ receiveSTM ch
217+
218+-- | Receive a message from a typed channel as an STM transaction.
219+--
220+-- The transaction retries when no message is available.
221+receiveSTM :: ReceivePort a -> STM a
222+receiveSTM (ReceivePortSingle c) =
223+ readTQueue c
224+receiveSTM (ReceivePortBiased ps) =
225+ foldr1 orElse (map receiveSTM ps)
226+receiveSTM (ReceivePortRR psVar) = do
227+ ps <- readTVar psVar
228+ a <- foldr1 orElse (map receiveSTM ps)
229+ writeTVar psVar (rotate ps)
230+ return a
231+ where
219232 rotate :: [a] -> [a]
220233 rotate [] = []
221234 rotate (x:xs) = xs ++ [x]
@@ -459,7 +472,7 @@ finally a sequel = bracket_ (return ()) sequel a
459472
460473-- | Like 'expect' but with a timeout
461474expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
462−expectTimeout timeout = receiveTimeout timeout [match return]
475+expectTimeout n = receiveTimeout n [match return]
463476
464477-- | Asynchronous version of 'spawn'
465478--