@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
492492data IncomingTarget =
493493 Uninit
494494 | ToProc ProcessId (Weak (CQueue Message))
495− | ToChan TypedChannel
495+ | ToChan SendPortId TypedChannel
496496 | ToNode
497497
498498data ConnectionState = ConnectionState {
@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState
547547 enqueue queue msg -- 'enqueue' is strict
548548 trace node (MxReceived pid msg)
549549 go st
550− Just (_, ToChan (TypedChannel chan')) -> do
550+ Just (_, ToChan chId (TypedChannel chan')) -> do
551551 mChan <- deRefWeak chan'
552552 -- If mChan is Nothing, the process has given up the read end of
553553 -- the channel and we simply ignore the incoming message
554− forM_ mChan $ \chan -> atomically $
555− -- We make sure the message is fully decoded when it is enqueued
556− writeTQueue chan $! decode (BSL.fromChunks payload)
554+ forM_ mChan $ \chan -> do
555+ msg' <- atomically $ do
556+ msg <- return $! decode (BSL.fromChunks payload)
557+ -- We make sure the message is fully decoded when it is enqueued
558+ writeTQueue chan msg
559+ return msg
560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
557561 go st
558562 Just (_, ToNode) -> do
559563 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState
580584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
581585 case mChannel of
582586 Just channel ->
583− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
584588 Nothing ->
585589 invalidRequest cid st $
586590 "incoming attempt to connect to unknown channel of"
@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do
10501054 -- If ch is Nothing, the process has given up the read end of
10511055 -- the channel and we simply ignore the incoming message - this
10521056 ch <- deRefWeak chan'
1053− forM_ ch $ \chan -> deliverChan msg chan
1054− where deliverChan :: forall a . Message -> TQueue a -> IO ()
1055− deliverChan (UnencodedMessage _ raw) chan' =
1057+ forM_ ch $ \chan -> deliverChan node from msg chan
1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do
10561060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1057− deliverChan (EncodedMessage _ _) _ =
1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
1062+ deliverChan _ _ (EncodedMessage _ _) _ =
10581063 -- this will not happen unless someone screws with Primitives.hs
10591064 error "invalid local channel delivery"
10601065