@@ -15,8 +15,9 @@
1515-- * Output exception on channel after endpoint is closed
1616module Network.Transport.TCP ( -- * Main API
1717 createTransport
18− , -- * TCP specific functionality (exported mostly for testing purposes_
19− EndPointId
18+ -- * Internals (exposed for unit tests)
19+ , createTransportExposeInternals
20+ , EndPointId
2021 , encodeEndPointAddress
2122 , decodeEndPointAddress
2223 , ControlHeader(..)
@@ -319,6 +320,16 @@ data ConnectionRequestResponse =
319320-- TODOs: deal with hints
320321createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
321322createTransport host port = do
323+ result <- createTransportExposeInternals host port
324+ case result of
325+ Left err ->
326+ return $ Left err
327+ Right (transport, _) ->
328+ return $ Right transport
329+
330+-- | You should probably not use this function (used for unit testing only)
331+createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, ThreadId))
332+createTransportExposeInternals host port = do
322333 state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty }
323334 let transport = TCPTransport { transportState = state
324335 , transportHost = host
@@ -329,27 +340,45 @@ createTransport host port = do
329340 -- http://tangentsoft.net/wskfaq/advanced.html#backlog
330341 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
331342 transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
332− return Transport { newEndPoint = apiNewEndPoint transport
333− , closeTransport = killThread transportThread -- This will invoke the termination handler
334− }
343+ return ( Transport { newEndPoint = apiNewEndPoint transport
344+ , closeTransport = apiCloseTransport transport (Just transportThread) [EndPointClosed]
345+ }
346+ , transportThread
347+ )
335348 where
336349 terminationHandler :: TCPTransport -> SomeException -> IO ()
337− terminationHandler transport _ = do
338− -- TODO: we currently don't make a distinction between endpoint failure and manual closure
339− mTSt <- modifyMVar (transportState transport) $ \st -> case st of
340− TransportValid vst -> return (TransportClosed, Just vst)
341− TransportClosed -> return (TransportClosed, Nothing)
342− case mTSt of
343− Nothing ->
344− -- Transport already closed
345− return ()
346− Just tSt ->
347− mapM_ (apiCloseEndPoint transport) (Map.elems $ tSt ^. localEndPoints)
350+ terminationHandler transport ex = do
351+ let ev = ErrorEvent (TransportError EventTransportFailed (show ex))
352+ apiCloseTransport transport Nothing [ev]
348353
349354--------------------------------------------------------------------------------
350355-- API functions --
351356--------------------------------------------------------------------------------
352357
358+-- | Close the transport
359+apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()
360+apiCloseTransport transport mTransportThread evs = do
361+ mTSt <- modifyMVar (transportState transport) $ \st -> case st of
362+ TransportValid vst -> return (TransportClosed, Just vst)
363+ TransportClosed -> return (TransportClosed, Nothing)
364+ case mTSt of
365+ Nothing ->
366+ -- Transport already closed
367+ return ()
368+ Just tSt -> do
369+ mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)
370+ case mTransportThread of
371+ Just transportThread ->
372+ -- This will invoke the termination handler, which in turn will call
373+ -- apiCloseTransport again, but at that point the transport is
374+ -- already in closed state so at that point we return immediately
375+ killThread transportThread
376+ Nothing ->
377+ -- This will happen is apiCloseTransport was invoked *because* the
378+ -- transport thread died (in which case we don't need to kill it)
379+ return ()
380+
381+
353382-- | Create a new endpoint
354383apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
355384apiNewEndPoint transport = try $ do
@@ -358,7 +387,7 @@ apiNewEndPoint transport = try $ do
358387 { receive = readChan (localChannel ourEndPoint)
359388 , address = localAddress ourEndPoint
360389 , connect = apiConnect ourEndPoint
361− , closeEndPoint = apiCloseEndPoint transport ourEndPoint
390+ , closeEndPoint = apiCloseEndPoint transport [EndPointClosed] ourEndPoint
362391 , newMulticastGroup = return . Left $ newMulticastGroupError
363392 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
364393 }
@@ -442,8 +471,11 @@ apiSend theirEndPoint connId connAlive payload = do
442471 sendFailed = TransportError SendFailed . show
443472
444473-- | Force-close the endpoint
445−apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
446−apiCloseEndPoint transport ourEndPoint = do
474+apiCloseEndPoint :: TCPTransport -- ^ Transport
475+ -> [Event] -- ^ Events to output on the endpoint to indicate closure
476+ -> LocalEndPoint -- ^ Local endpoint
477+ -> IO ()
478+apiCloseEndPoint transport evs ourEndPoint = do
447479 -- Remove the reference from the transport state
448480 removeLocalEndPoint transport ourEndPoint
449481 -- Close the local endpoint
@@ -460,11 +492,7 @@ apiCloseEndPoint transport ourEndPoint = do
460492 -- Close all endpoints and kill all threads
461493 forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket
462494 forM_ (vst ^. internalThreads) killThread
463− -- We send a single message to the endpoint that it is closed. Subsequent
464− -- calls will block. We could change this so that all subsequent calls to
465− -- receive return an error, but this would mean checking for some state on
466− -- every call to receive, which is an unnecessary overhead.
467− writeChan (localChannel ourEndPoint) EndPointClosed
495+ forM_ evs $ writeChan (localChannel ourEndPoint)
468496 where
469497 -- Close the remote socket and return the set of all incoming connections
470498 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()