@@ -39,6 +39,8 @@ import Control.Concurrent.Chan
3939import Control.Concurrent
4040import Data.Typeable
4141
42+import Debug.Trace
43+
4244------------------------
4345-- Cloud Haskell layer
4446--
@@ -143,8 +145,15 @@ forkProcess node (Process action) = do
143145 receiverPump :: Trans.ReceiveEnd -> CQueue Message -> IO ()
144146 receiverPump chan queue = forever $ do
145147 msgBlobs <- Trans.receive chan
146− let (typerep, body) = read (concatMap BS.unpack msgBlobs)
147− enqueue queue (Message typerep body)
148+ readBlobs (concatMap BS.unpack msgBlobs)
149+ where
150+ readBlobs :: String -> IO ()
151+ readBlobs [] = return ()
152+ readBlobs bs = do
153+ -- TODO: replace reads with something safer and faster.
154+ let [((typerep, body), bs')] = reads bs
155+ enqueue queue (Message typerep body)
156+ readBlobs bs'
148157
149158send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
150159send (ProcessId chan _ _) msg =
@@ -212,4 +221,4 @@ dequeueMatching (CQueue arrived incomming) matches = do
212221 x <- readChan incomming
213222 if matches x
214223 then return (reverse xs', x)
215− else checkIncomming (x:xs')
224+ else checkIncomming (x:xs')