@@ -1,22 +1,22 @@
1−-- | TCP implementation of the transport layer.
2−--
1+-- | TCP implementation of the transport layer.
2+--
33-- The TCP implementation guarantees that only a single TCP connection (socket)
44-- will be used between endpoints, provided that the addresses specified are
55-- canonical. If /A/ connects to /B/ and reports its address as
66-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
77-- @client1.local:http-alt@ then the transport layer will not realize that the
8−-- TCP connection can be reused.
8+-- TCP connection can be reused.
99--
1010-- Applications that use the TCP transport should use
1111-- 'Network.Socket.withSocketsDo' in their main function for Windows
1212-- compatibility (see "Network.Socket").
13−module Network.Transport.TCP
13+module Network.Transport.TCP
1414 ( -- * Main API
1515 createTransport
1616 , TCPParameters(..)
1717 , defaultTCPParameters
18− -- * Internals (exposed for unit tests)
19− , createTransportExposeInternals
18+ -- * Internals (exposed for unit tests)
19+ , createTransportExposeInternals
2020 , TransportInternals(..)
2121 , EndPointId
2222 , encodeEndPointAddress
@@ -25,27 +25,27 @@ module Network.Transport.TCP
2525 , ConnectionRequestResponse(..)
2626 , firstNonReservedLightweightConnectionId
2727 , firstNonReservedHeavyweightConnectionId
28− , socketToEndPoint
28+ , socketToEndPoint
2929 , LightweightConnectionId
3030 -- * Design notes
3131 -- $design
3232 ) where
3333
34−import Prelude hiding
34+import Prelude hiding
3535 ( mapM_
3636#if ! MIN_VERSION_base(4,6,0)
3737 , catch
3838#endif
3939 )
40−
40+
4141import Network.Transport
42−import Network.Transport.TCP.Internal
42+import Network.Transport.TCP.Internal
4343 ( forkServer
4444 , recvWithLength
4545 , recvInt32
4646 , tryCloseSocket
4747 )
48−import Network.Transport.Internal
48+import Network.Transport.Internal
4949 ( encodeInt32
5050 , prependLength
5151 , mapIOException
@@ -59,7 +59,7 @@ import Network.Transport.Internal
5959#ifdef USE_MOCK_NETWORK
6060import qualified Network.Transport.TCP.Mock.Socket as N
6161#else
62−import qualified Network.Socket as N
62+import qualified Network.Socket as N
6363#endif
6464 ( HostName
6565 , ServiceName
@@ -71,7 +71,7 @@ import qualified Network.Socket as N
7171 , SocketType(Stream)
7272 , defaultProtocol
7373 , setSocketOption
74− , SocketOption(ReuseAddr)
74+ , SocketOption(ReuseAddr)
7575 , connect
7676 , sOMAXCONN
7777 , AddrInfo
@@ -85,7 +85,7 @@ import Network.Socket.ByteString (sendMany)
8585
8686import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
8787import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
88−import Control.Concurrent.MVar
88+import Control.Concurrent.MVar
8989 ( MVar
9090 , newMVar
9191 , modifyMVar
@@ -98,7 +98,7 @@ import Control.Concurrent.MVar
9898import Control.Category ((>>>))
9999import Control.Applicative ((<$>))
100100import Control.Monad (when, unless, join)
101−import Control.Exception
101+import Control.Exception
102102 ( IOException
103103 , SomeException
104104 , AsyncException
@@ -117,7 +117,7 @@ import qualified Data.ByteString.Char8 as BSC (pack, unpack)
117117import Data.Bits (shiftL, (.|.))
118118import Data.Word (Word32)
119119import Data.Set (Set)
120−import qualified Data.Set as Set
120+import qualified Data.Set as Set
121121 ( empty
122122 , insert
123123 , elems
@@ -128,11 +128,11 @@ import qualified Data.Set as Set
128128 )
129129import Data.Map (Map)
130130import qualified Data.Map as Map (empty)
131−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
131+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
132132import qualified Data.Accessor.Container as DAC (mapMaybe)
133133import Data.Foldable (forM_, mapM_)
134134
135−-- $design
135+-- $design
136136--
137137-- [Goals]
138138--
@@ -146,7 +146,7 @@ import Data.Foldable (forM_, mapM_)
146146-- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |
147147-- > | |==========================| |
148148-- > +-------+ +-------+
149−--
149+--
150150-- Ignoring the complications detailed below, the TCP connection is set up is
151151-- when the first lightweight connection is created (in either direction), and
152152-- torn down when the last lightweight connection (in either direction) is
@@ -177,7 +177,7 @@ import Data.Foldable (forM_, mapM_)
177177-- with /B/'s connection request completes.
178178--
179179-- [Disconnecting]
180−--
180+--
181181-- The TCP connection is created as soon as the first logical connection from
182182-- /A/ to /B/ (or /B/ to /A/) is established. At this point a thread (@#@) is
183183-- spawned that listens for incoming connections from /B/:
@@ -193,21 +193,21 @@ import Data.Foldable (forM_, mapM_)
193193-- The question is when the TCP connection can be closed again. Conceptually,
194194-- we want to do reference counting: when there are no logical connections left
195195-- between /A/ and /B/ we want to close the socket (possibly after some
196−-- timeout).
196+-- timeout).
197197--
198198-- However, /A/ and /B/ need to agree that the refcount has reached zero. It
199199-- might happen that /B/ sends a connection request over the existing socket at
200200-- the same time that /A/ closes its logical connection to /B/ and closes the
201201-- socket. This will cause a failure in /B/ (which will have to retry) which is
202202-- not caused by a network failure, which is unfortunate. (Note that the
203−-- connection request from /B/ might succeed even if /A/ closes the socket.)
203+-- connection request from /B/ might succeed even if /A/ closes the socket.)
204204--
205205-- Instead, when /A/ is ready to close the socket it sends a 'CloseSocket'
206206-- request to /B/ and records that its connection to /B/ is closing. If /A/
207207-- receives a new connection request from /B/ after having sent the
208208-- 'CloseSocket' request it simply forgets that it sent a 'CloseSocket' request
209209-- and increments the reference count of the connection again.
210−--
210+--
211211-- When /B/ receives a 'CloseSocket' message and it too is ready to close the
212212-- connection, it will respond with a reciprocal 'CloseSocket' request to /A/
213213-- and then actually close the socket. /A/ meanwhile will not send any more
@@ -216,11 +216,11 @@ import Data.Foldable (forM_, mapM_)
216216-- from /B/. (Since /A/ recorded that its connection to /B/ is in closing state
217217-- after sending a 'CloseSocket' request to /B/, it knows not to reciprocate /B/
218218-- reciprocal 'CloseSocket' message.)
219−--
219+--
220220-- If there is a concurrent thread in /A/ waiting to connect to /B/ after /A/
221221-- has sent a 'CloseSocket' request then this thread will block until /A/ knows
222222-- whether to reuse the old socket (if /B/ sends a new connection request
223−-- instead of acknowledging the 'CloseSocket') or to set up a new socket.
223+-- instead of acknowledging the 'CloseSocket') or to set up a new socket.
224224
225225--------------------------------------------------------------------------------
226226-- Internal datatypes --
@@ -229,30 +229,30 @@ import Data.Foldable (forM_, mapM_)
229229-- We use underscores for fields that we might update (using accessors)
230230--
231231-- All data types follow the same structure:
232−--
232+--
233233-- * A top-level data type describing static properties (TCPTransport,
234234-- LocalEndPoint, RemoteEndPoint)
235235-- * The 'static' properties include an MVar containing a data structure for
236236-- the dynamic properties (TransportState, LocalEndPointState,
237−-- RemoteEndPointState). The state could be invalid/valid/closed,/etc.
237+-- RemoteEndPointState). The state could be invalid/valid/closed,/etc.
238238-- * For the case of "valid" we use third data structure to give more details
239239-- about the state (ValidTransportState, ValidLocalEndPointState,
240240-- ValidRemoteEndPointState).
241241
242−data TCPTransport = TCPTransport
242+data TCPTransport = TCPTransport
243243 { transportHost :: !N.HostName
244244 , transportPort :: !N.ServiceName
245245 , transportState :: !(MVar TransportState)
246246 , transportParams :: !TCPParameters
247247 }
248248
249−data TransportState =
249+data TransportState =
250250 TransportValid !ValidTransportState
251251 | TransportClosed
252252
253−data ValidTransportState = ValidTransportState
253+data ValidTransportState = ValidTransportState
254254 { _localEndPoints :: !(Map EndPointAddress LocalEndPoint)
255− , _nextEndPointId :: !EndPointId
255+ , _nextEndPointId :: !EndPointId
256256 }
257257
258258data LocalEndPoint = LocalEndPoint
@@ -261,12 +261,12 @@ data LocalEndPoint = LocalEndPoint
261261 , localState :: !(MVar LocalEndPointState)
262262 }
263263
264−data LocalEndPointState =
264+data LocalEndPointState =
265265 LocalEndPointValid !ValidLocalEndPointState
266266 | LocalEndPointClosed
267267
268−data ValidLocalEndPointState = ValidLocalEndPointState
269− { -- Next available ID for an outgoing lightweight self-connection
268+data ValidLocalEndPointState = ValidLocalEndPointState
269+ { -- Next available ID for an outgoing lightweight self-connection
270270 -- (see also remoteNextConnOutId)
271271 _localNextConnOutId :: !LightweightConnectionId
272272 -- Next available ID for an incoming heavyweight connection
@@ -275,7 +275,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState
275275 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
276276 }
277277
278−-- REMOTE ENDPOINTS
278+-- REMOTE ENDPOINTS
279279--
280280-- Remote endpoints (basically, TCP connections) have the following lifecycle:
281281--
@@ -302,7 +302,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState
302302-- createConnectionTo when we fail to connect.
303303--
304304-- Valid: This is the "normal" state for a working remote endpoint.
305−--
305+--
306306-- Closing: When we detect that a remote endpoint is no longer used, we send a
307307-- CloseSocket request across the connection and put the remote endpoint in
308308-- closing state. As with Init, 'Closing' carries an MVar () 'resolved' which
@@ -335,16 +335,16 @@ data ValidLocalEndPointState = ValidLocalEndPointState
335335-- localConnections it is guaranteed to either find a different remote
336336-- endpoint, or else none at all (if we don't insist in this order some
337337-- threads might start spinning).
338−--
338+--
339339-- INV-RESOLVE: We should only signal on 'resolved' while the remote endpoint is
340340-- locked, and the remote endpoint must be in Valid or Closed state once
341341-- unlocked. This guarantees that there will not be two threads attempting to
342−-- both signal on 'resolved'.
342+-- both signal on 'resolved'.
343343--
344344-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we
345345-- first put the remote endpoint in Closed state, and then send a
346346-- EventConnectionLost event. This guarantees that we only send this event
347−-- once.
347+-- once.
348348--
349349-- INV-CLOSING: An endpoint in closing state is for all intents and purposes
350350-- closed; that is, we shouldn't do any 'send's on it (although 'recv' is
@@ -357,23 +357,23 @@ data ValidLocalEndPointState = ValidLocalEndPointState
357357-- particular, it is okay to call removeRemoteEndPoint inside
358358-- modifyRemoteState.
359359
360−data RemoteEndPoint = RemoteEndPoint
360+data RemoteEndPoint = RemoteEndPoint
361361 { remoteAddress :: !EndPointAddress
362362 , remoteState :: !(MVar RemoteState)
363363 , remoteId :: !HeavyweightConnectionId
364364 , remoteScheduled :: !(Chan (IO ()))
365365 }
366366
367−data RequestedBy = RequestedByUs | RequestedByThem
367+data RequestedBy = RequestedByUs | RequestedByThem
368368 deriving (Eq, Show)
369369
370370data RemoteState =
371371 -- | Invalid remote endpoint (for example, invalid address)
372372 RemoteEndPointInvalid !(TransportError ConnectErrorCode)
373373 -- | The remote endpoint is being initialized
374− | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy
374+ | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy
375375 -- | "Normal" working endpoint
376− | RemoteEndPointValid !ValidRemoteEndPointState
376+ | RemoteEndPointValid !ValidRemoteEndPointState
377377 -- | The remote endpoint is being closed (garbage collected)
378378 | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState
379379 -- | The remote endpoint has been closed (garbage collected)
@@ -382,12 +382,12 @@ data RemoteState =
382382 -- using a closeTransport or closeEndPoint API call
383383 | RemoteEndPointFailed !IOException
384384
385−-- TODO: we might want to replace Set (here and elsewhere) by faster
385+-- TODO: we might want to replace Set (here and elsewhere) by faster
386386-- containers
387387--
388388-- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if
389−-- we introduce a new event 'AllConnectionsClosed'
390−data ValidRemoteEndPointState = ValidRemoteEndPointState
389+-- we introduce a new event 'AllConnectionsClosed'
390+data ValidRemoteEndPointState = ValidRemoteEndPointState
391391 { _remoteOutgoing :: !Int
392392 , _remoteIncoming :: !(Set LightweightConnectionId)
393393 , _remoteMaxIncoming :: !LightweightConnectionId
@@ -414,24 +414,24 @@ type LightweightConnectionId = Word32
414414-- 'LightweightConnectionId'.
415415type HeavyweightConnectionId = Word32
416416
417−-- | Control headers
418−data ControlHeader =
417+-- | Control headers
418+data ControlHeader =
419419 -- | Tell the remote endpoint that we created a new connection
420− CreatedNewConnection
420+ CreatedNewConnection
421421 -- | Tell the remote endpoint we will no longer be using a connection
422− | CloseConnection
422+ | CloseConnection
423423 -- | Request to close the connection (see module description)
424− | CloseSocket
424+ | CloseSocket
425425 deriving (Enum, Bounded, Show)
426426
427427-- | Response sent by /B/ to /A/ when /A/ tries to connect
428428data ConnectionRequestResponse =
429429 -- | /B/ accepts the connection
430− ConnectionRequestAccepted
430+ ConnectionRequestAccepted
431431 -- | /A/ requested an invalid endpoint
432− | ConnectionRequestInvalid
432+ | ConnectionRequestInvalid
433433 -- | /A/s request crossed with a request from /B/ (see protocols)
434− | ConnectionRequestCrossed
434+ | ConnectionRequestCrossed
435435 deriving (Enum, Bounded, Show)
436436
437437-- | Parameters for setting up the TCP transport
@@ -439,7 +439,7 @@ data TCPParameters = TCPParameters {
439439 -- | Backlog for 'listen'.
440440 -- Defaults to SOMAXCONN.
441441 tcpBacklog :: Int
442− -- | Should we set SO_REUSEADDR on the server socket?
442+ -- | Should we set SO_REUSEADDR on the server socket?
443443 -- Defaults to True.
444444 , tcpReuseServerAddr :: Bool
445445 -- | Should we set SO_REUSEADDR on client sockets?
@@ -448,13 +448,13 @@ data TCPParameters = TCPParameters {
448448 }
449449
450450-- | Internal functionality we expose for unit testing
451−data TransportInternals = TransportInternals
451+data TransportInternals = TransportInternals
452452 { -- | The ID of the thread that listens for new incoming connections
453453 transportThread :: ThreadId
454454 -- | Find the socket between a local and a remote endpoint
455− , socketBetween :: EndPointAddress
456− -> EndPointAddress
457− -> IO N.Socket
455+ , socketBetween :: EndPointAddress
456+ -> EndPointAddress
457+ -> IO N.Socket
458458 }
459459
460460--------------------------------------------------------------------------------
@@ -462,62 +462,62 @@ data TransportInternals = TransportInternals
462462--------------------------------------------------------------------------------
463463
464464-- | Create a TCP transport
465−createTransport :: N.HostName
466− -> N.ServiceName
465+createTransport :: N.HostName
466+ -> N.ServiceName
467467 -> TCPParameters
468468 -> IO (Either IOException Transport)
469−createTransport host port params =
469+createTransport host port params =
470470 either Left (Right . fst) <$> createTransportExposeInternals host port params
471471
472472-- | You should probably not use this function (used for unit testing only)
473−createTransportExposeInternals
474− :: N.HostName
475− -> N.ServiceName
473+createTransportExposeInternals
474+ :: N.HostName
475+ -> N.ServiceName
476476 -> TCPParameters
477− -> IO (Either IOException (Transport, TransportInternals))
478−createTransportExposeInternals host port params = do
479− state <- newMVar . TransportValid $ ValidTransportState
480− { _localEndPoints = Map.empty
481− , _nextEndPointId = 0
477+ -> IO (Either IOException (Transport, TransportInternals))
478+createTransportExposeInternals host port params = do
479+ state <- newMVar . TransportValid $ ValidTransportState
480+ { _localEndPoints = Map.empty
481+ , _nextEndPointId = 0
482482 }
483483 let transport = TCPTransport { transportState = state
484484 , transportHost = host
485485 , transportPort = port
486486 , transportParams = params
487487 }
488− tryIO $ bracketOnError (forkServer
489− host
490− port
491− (tcpBacklog params)
488+ tryIO $ bracketOnError (forkServer
489+ host
490+ port
491+ (tcpBacklog params)
492492 (tcpReuseServerAddr params)
493− (terminationHandler transport)
493+ (terminationHandler transport)
494494 (handleConnectionRequest transport))
495495 killThread
496496 (mkTransport transport)
497497 where
498− mkTransport :: TCPTransport
499− -> ThreadId
498+ mkTransport :: TCPTransport
499+ -> ThreadId
500500 -> IO (Transport, TransportInternals)
501501 mkTransport transport tid = return
502− ( Transport
503− { newEndPoint = apiNewEndPoint transport
502+ ( Transport
503+ { newEndPoint = apiNewEndPoint transport
504504 , closeTransport = let evs = [ EndPointClosed
505505 , throw $ userError "Transport closed"
506506 ] in
507− apiCloseTransport transport (Just tid) evs
508− }
509− , TransportInternals
507+ apiCloseTransport transport (Just tid) evs
508+ }
509+ , TransportInternals
510510 { transportThread = tid
511511 , socketBetween = internalSocketBetween transport
512512 }
513513 )
514514
515515 terminationHandler :: TCPTransport -> SomeException -> IO ()
516− terminationHandler transport ex = do
516+ terminationHandler transport ex = do
517517 let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))
518− , throw $ userError "Transport closed"
518+ , throw $ userError "Transport closed"
519519 ]
520− apiCloseTransport transport Nothing evs
520+ apiCloseTransport transport Nothing evs
521521
522522-- | Default TCP parameters
523523defaultTCPParameters :: TCPParameters
@@ -533,39 +533,39 @@ defaultTCPParameters = TCPParameters {
533533
534534-- | Close the transport
535535apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()
536−apiCloseTransport transport mTransportThread evs =
536+apiCloseTransport transport mTransportThread evs =
537537 asyncWhenCancelled return $ do
538538 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
539539 TransportValid vst -> return (TransportClosed, Just vst)
540540 TransportClosed -> return (TransportClosed, Nothing)
541− forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
541+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
542542 -- This will invoke the termination handler, which in turn will call
543543 -- apiCloseTransport again, but then the transport will already be closed
544544 -- and we won't be passed a transport thread, so we terminate immmediate
545− forM_ mTransportThread killThread
546−
547−-- | Create a new endpoint
548−apiNewEndPoint :: TCPTransport
545+ forM_ mTransportThread killThread
546+
547+-- | Create a new endpoint
548+apiNewEndPoint :: TCPTransport
549549 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
550−apiNewEndPoint transport =
550+apiNewEndPoint transport =
551551 try . asyncWhenCancelled closeEndPoint $ do
552552 ourEndPoint <- createLocalEndPoint transport
553− return EndPoint
553+ return EndPoint
554554 { receive = readChan (localChannel ourEndPoint)
555555 , address = localAddress ourEndPoint
556− , connect = apiConnect (transportParams transport) ourEndPoint
556+ , connect = apiConnect (transportParams transport) ourEndPoint
557557 , closeEndPoint = let evs = [ EndPointClosed
558− , throw $ userError "Endpoint closed"
558+ , throw $ userError "Endpoint closed"
559559 ] in
560560 apiCloseEndPoint transport evs ourEndPoint
561− , newMulticastGroup = return . Left $ newMulticastGroupError
561+ , newMulticastGroup = return . Left $ newMulticastGroupError
562562 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
563563 }
564564 where
565− newMulticastGroupError =
566− TransportError NewMulticastGroupUnsupported "Multicast not supported"
567− resolveMulticastGroupError =
568− TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
565+ newMulticastGroupError =
566+ TransportError NewMulticastGroupUnsupported "Multicast not supported"
567+ resolveMulticastGroupError =
568+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
569569
570570-- | Connnect to an endpoint
571571apiConnect :: TCPParameters -- ^ Parameters
@@ -575,35 +575,35 @@ apiConnect :: TCPParameters -- ^ Parameters
575575 -> ConnectHints -- ^ Hints
576576 -> IO (Either (TransportError ConnectErrorCode) Connection)
577577apiConnect params ourEndPoint theirAddress _reliability hints =
578− try . asyncWhenCancelled close $
579− if localAddress ourEndPoint == theirAddress
580− then connectToSelf ourEndPoint
578+ try . asyncWhenCancelled close $
579+ if localAddress ourEndPoint == theirAddress
580+ then connectToSelf ourEndPoint
581581 else do
582582 resetIfBroken ourEndPoint theirAddress
583− (theirEndPoint, connId) <-
583+ (theirEndPoint, connId) <-
584584 createConnectionTo params ourEndPoint theirAddress hints
585585 -- connAlive can be an IORef rather than an MVar because it is protected
586586 -- by the remoteState MVar. We don't need the overhead of locking twice.
587587 connAlive <- newIORef True
588− return Connection
589− { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
590− , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
588+ return Connection
589+ { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
590+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
591591 }
592592
593593-- | Close a connection
594594apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()
595−apiClose (ourEndPoint, theirEndPoint) connId connAlive =
596− void . tryIO . asyncWhenCancelled return $ do
595+apiClose (ourEndPoint, theirEndPoint) connId connAlive =
596+ void . tryIO . asyncWhenCancelled return $ do
597597 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
598598 RemoteEndPointValid vst -> do
599599 alive <- readIORef connAlive
600− if alive
600+ if alive
601601 then do
602602 writeIORef connAlive False
603− act <- schedule theirEndPoint $
604− sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
605− return ( RemoteEndPointValid
606− . (remoteOutgoing ^: (\x -> x - 1))
603+ act <- schedule theirEndPoint $
604+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
605+ return ( RemoteEndPointValid
606+ . (remoteOutgoing ^: (\x -> x - 1))
607607 $ vst
608608 , Just act
609609 )
@@ -616,12 +616,12 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive =
616616
617617
618618-- | Send data across a connection
619−apiSend :: EndPointPair -- ^ Local and remote endpoint
619+apiSend :: EndPointPair -- ^ Local and remote endpoint
620620 -> LightweightConnectionId -- ^ Connection ID
621621 -> IORef Bool -- ^ Is the connection still alive?
622622 -> [ByteString] -- ^ Payload
623623 -> IO (Either (TransportError SendErrorCode) ())
624−apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
624+apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
625625 -- We don't need the overhead of asyncWhenCancelled here
626626 try . mapIOException sendFailed $ do
627627 act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
@@ -631,70 +631,70 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
631631 relyViolation (ourEndPoint, theirEndPoint) "apiSend"
632632 RemoteEndPointValid vst -> do
633633 alive <- readIORef connAlive
634− if alive
635− then schedule theirEndPoint $
634+ if alive
635+ then schedule theirEndPoint $
636636 sendOn vst (encodeInt32 connId : prependLength payload)
637637 else throwIO $ TransportError SendClosed "Connection closed"
638− RemoteEndPointClosing _ _ -> do
638+ RemoteEndPointClosing _ _ -> do
639639 alive <- readIORef connAlive
640− if alive
641− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
640+ if alive
641+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
642642 else throwIO $ TransportError SendClosed "Connection closed"
643643 RemoteEndPointClosed -> do
644644 alive <- readIORef connAlive
645− if alive
646− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
645+ if alive
646+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
647647 else throwIO $ TransportError SendClosed "Connection closed"
648648 RemoteEndPointFailed err -> do
649649 alive <- readIORef connAlive
650− if alive
651− then throwIO $ TransportError SendFailed (show err)
650+ if alive
651+ then throwIO $ TransportError SendFailed (show err)
652652 else throwIO $ TransportError SendClosed "Connection closed"
653653 runScheduledAction (ourEndPoint, theirEndPoint) act
654654 where
655655 sendFailed = TransportError SendFailed . show
656656
657657-- | Force-close the endpoint
658−apiCloseEndPoint :: TCPTransport -- ^ Transport
659− -> [Event] -- ^ Events used to report closure
658+apiCloseEndPoint :: TCPTransport -- ^ Transport
659+ -> [Event] -- ^ Events used to report closure
660660 -> LocalEndPoint -- ^ Local endpoint
661661 -> IO ()
662662apiCloseEndPoint transport evs ourEndPoint =
663663 asyncWhenCancelled return $ do
664664 -- Remove the reference from the transport state
665665 removeLocalEndPoint transport ourEndPoint
666− -- Close the local endpoint
666+ -- Close the local endpoint
667667 mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
668668 case st of
669− LocalEndPointValid vst ->
669+ LocalEndPointValid vst ->
670670 return (LocalEndPointClosed, Just vst)
671671 LocalEndPointClosed ->
672672 return (LocalEndPointClosed, Nothing)
673673 forM_ mOurState $ \vst -> do
674674 forM_ (vst ^. localConnections) tryCloseRemoteSocket
675− forM_ evs $ writeChan (localChannel ourEndPoint)
675+ forM_ evs $ writeChan (localChannel ourEndPoint)
676676 where
677677 -- Close the remote socket and return the set of all incoming connections
678− tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
678+ tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
679679 tryCloseRemoteSocket theirEndPoint = do
680− -- We make an attempt to close the connection nicely
680+ -- We make an attempt to close the connection nicely
681681 -- (by sending a CloseSocket first)
682682 let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"
683683 mAct <- modifyMVar (remoteState theirEndPoint) $ \st ->
684684 case st of
685− RemoteEndPointInvalid _ ->
685+ RemoteEndPointInvalid _ ->
686686 return (st, Nothing)
687687 RemoteEndPointInit resolved _ _ -> do
688688 putMVar resolved ()
689689 return (closed, Nothing)
690− RemoteEndPointValid vst -> do
690+ RemoteEndPointValid vst -> do
691691 act <- schedule theirEndPoint $ do
692692 tryIO $ sendOn vst [ encodeInt32 CloseSocket
693693 , encodeInt32 (vst ^. remoteMaxIncoming)
694694 ]
695695 tryCloseSocket (remoteSocket vst)
696696 return (closed, Just act)
697− RemoteEndPointClosing resolved vst -> do
697+ RemoteEndPointClosing resolved vst -> do
698698 putMVar resolved ()
699699 act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)
700700 return (closed, Just act)
@@ -703,7 +703,7 @@ apiCloseEndPoint transport evs ourEndPoint =
703703 RemoteEndPointFailed err ->
704704 return (RemoteEndPointFailed err, Nothing)
705705 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
706−
706+
707707
708708--------------------------------------------------------------------------------
709709-- Incoming requests --
@@ -720,11 +720,11 @@ apiCloseEndPoint transport evs ourEndPoint =
720720-- the transport down. We must be careful to close the socket when a (possibly
721721-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
722722-- handleConnectionRequest the transport will be shut down.)
723−handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
724−handleConnectionRequest transport sock = handle handleException $ do
723+handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
724+handleConnectionRequest transport sock = handle handleException $ do
725725 ourEndPointId <- recvInt32 sock
726− theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
727− let ourAddress = encodeEndPointAddress (transportHost transport)
726+ theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
727+ let ourAddress = encodeEndPointAddress (transportHost transport)
728728 (transportPort transport)
729729 ourEndPointId
730730 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
@@ -735,32 +735,32 @@ handleConnectionRequest transport sock = handle handleException $ do
735735 throwIO $ userError "handleConnectionRequest: Invalid endpoint"
736736 Just ourEndPoint ->
737737 return ourEndPoint
738− TransportClosed ->
738+ TransportClosed ->
739739 throwIO $ userError "Transport closed"
740740 void . forkIO $ go ourEndPoint theirAddress
741741 where
742742 go :: LocalEndPoint -> EndPointAddress -> IO ()
743− go ourEndPoint theirAddress = do
743+ go ourEndPoint theirAddress = do
744744 -- This runs in a thread that will never be killed
745− mEndPoint <- handle ((>> return Nothing) . handleException) $ do
745+ mEndPoint <- handle ((>> return Nothing) . handleException) $ do
746746 resetIfBroken ourEndPoint theirAddress
747− (theirEndPoint, isNew) <-
747+ (theirEndPoint, isNew) <-
748748 findRemoteEndPoint ourEndPoint theirAddress RequestedByThem
749−
750− if not isNew
749+
750+ if not isNew
751751 then do
752752 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
753753 tryCloseSocket sock
754− return Nothing
754+ return Nothing
755755 else do
756756 sendLock <- newMVar ()
757− let vst = ValidRemoteEndPointState
757+ let vst = ValidRemoteEndPointState
758758 { remoteSocket = sock
759759 , remoteSendLock = sendLock
760760 , _remoteOutgoing = 0
761761 , _remoteIncoming = Set.empty
762762 , _remoteMaxIncoming = 0
763− , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
763+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
764764 }
765765 sendMany sock [encodeInt32 ConnectionRequestAccepted]
766766 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
@@ -769,30 +769,30 @@ handleConnectionRequest transport sock = handle handleException $ do
769769 -- Nothing then the socket is already closed; otherwise, the socket has
770770 -- been recorded as part of the remote endpoint. Either way, we no longer
771771 -- have to worry about closing the socket on receiving an asynchronous
772− -- exception from this point forward.
772+ -- exception from this point forward.
773773 forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
774774
775775 handleException :: SomeException -> IO ()
776− handleException ex = do
777− tryCloseSocket sock
776+ handleException ex = do
777+ tryCloseSocket sock
778778 rethrowIfAsync (fromException ex)
779779
780− rethrowIfAsync :: Maybe AsyncException -> IO ()
781− rethrowIfAsync = mapM_ throwIO
780+ rethrowIfAsync :: Maybe AsyncException -> IO ()
781+ rethrowIfAsync = mapM_ throwIO
782782
783783-- | Handle requests from a remote endpoint.
784−--
784+--
785785-- Returns only if the remote party closes the socket or if an error occurs.
786786-- This runs in a thread that will never be killed.
787−handleIncomingMessages :: EndPointPair -> IO ()
787+handleIncomingMessages :: EndPointPair -> IO ()
788788handleIncomingMessages (ourEndPoint, theirEndPoint) = do
789789 mSock <- withMVar theirState $ \st ->
790790 case st of
791791 RemoteEndPointInvalid _ ->
792− relyViolation (ourEndPoint, theirEndPoint)
793− "handleIncomingMessages (invalid)"
792+ relyViolation (ourEndPoint, theirEndPoint)
793+ "handleIncomingMessages (invalid)"
794794 RemoteEndPointInit _ _ _ ->
795− relyViolation (ourEndPoint, theirEndPoint)
795+ relyViolation (ourEndPoint, theirEndPoint)
796796 "handleIncomingMessages (init)"
797797 RemoteEndPointValid ep ->
798798 return . Just $ remoteSocket ep
@@ -800,13 +800,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
800800 return . Just $ remoteSocket ep
801801 RemoteEndPointClosed ->
802802 return Nothing
803− RemoteEndPointFailed _ ->
803+ RemoteEndPointFailed _ ->
804804 return Nothing
805−
806− forM_ mSock $ \sock ->
805+
806+ forM_ mSock $ \sock ->
807807 tryIO (go sock) >>= either (prematureExit sock) return
808808 where
809− -- Dispatch
809+ -- Dispatch
810810 --
811811 -- If a recv throws an exception this will be caught top-level and
812812 -- 'prematureExit' will be invoked. The same will happen if the remote
@@ -817,38 +817,38 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
817817 go :: N.Socket -> IO ()
818818 go sock = do
819819 lcid <- recvInt32 sock :: IO LightweightConnectionId
820− if lcid >= firstNonReservedLightweightConnectionId
820+ if lcid >= firstNonReservedLightweightConnectionId
821821 then do
822− readMessage sock lcid
822+ readMessage sock lcid
823823 go sock
824− else
824+ else
825825 case tryToEnum (fromIntegral lcid) of
826826 Just CreatedNewConnection -> do
827− recvInt32 sock >>= createdNewConnection
828− go sock
827+ recvInt32 sock >>= createdNewConnection
828+ go sock
829829 Just CloseConnection -> do
830− recvInt32 sock >>= closeConnection
830+ recvInt32 sock >>= closeConnection
831831 go sock
832− Just CloseSocket -> do
833− didClose <- recvInt32 sock >>= closeSocket sock
832+ Just CloseSocket -> do
833+ didClose <- recvInt32 sock >>= closeSocket sock
834834 unless didClose $ go sock
835835 Nothing ->
836836 throwIO $ userError "Invalid control request"
837−
837+
838838 -- Create a new connection
839− createdNewConnection :: LightweightConnectionId -> IO ()
839+ createdNewConnection :: LightweightConnectionId -> IO ()
840840 createdNewConnection lcid = do
841841 modifyMVar_ theirState $ \st -> do
842842 vst <- case st of
843843 RemoteEndPointInvalid _ ->
844− relyViolation (ourEndPoint, theirEndPoint)
844+ relyViolation (ourEndPoint, theirEndPoint)
845845 "handleIncomingMessages:createNewConnection (invalid)"
846846 RemoteEndPointInit _ _ _ ->
847− relyViolation (ourEndPoint, theirEndPoint)
847+ relyViolation (ourEndPoint, theirEndPoint)
848848 "handleIncomingMessages:createNewConnection (init)"
849849 RemoteEndPointValid vst ->
850850 return ( (remoteIncoming ^: Set.insert lcid)
851− $ (remoteMaxIncoming ^= lcid)
851+ $ (remoteMaxIncoming ^= lcid)
852852 vst
853853 )
854854 RemoteEndPointClosing resolved vst -> do
@@ -863,36 +863,36 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
863863 . (remoteMaxIncoming ^= lcid)
864864 $ vst
865865 )
866− RemoteEndPointFailed err ->
866+ RemoteEndPointFailed err ->
867867 throwIO err
868868 RemoteEndPointClosed ->
869− relyViolation (ourEndPoint, theirEndPoint)
869+ relyViolation (ourEndPoint, theirEndPoint)
870870 "createNewConnection (closed)"
871871 return (RemoteEndPointValid vst)
872− writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)
872+ writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)
873873
874− -- Close a connection
874+ -- Close a connection
875875 -- It is important that we verify that the connection is in fact open,
876876 -- because otherwise we should not decrement the reference count
877− closeConnection :: LightweightConnectionId -> IO ()
877+ closeConnection :: LightweightConnectionId -> IO ()
878878 closeConnection lcid = do
879879 modifyMVar_ theirState $ \st -> case st of
880880 RemoteEndPointInvalid _ ->
881881 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"
882882 RemoteEndPointInit _ _ _ ->
883883 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"
884− RemoteEndPointValid vst -> do
885− unless (Set.member lcid (vst ^. remoteIncoming)) $
884+ RemoteEndPointValid vst -> do
885+ unless (Set.member lcid (vst ^. remoteIncoming)) $
886886 throwIO $ userError "Invalid CloseConnection"
887− return ( RemoteEndPointValid
888− . (remoteIncoming ^: Set.delete lcid)
887+ return ( RemoteEndPointValid
888+ . (remoteIncoming ^: Set.delete lcid)
889889 $ vst
890890 )
891891 RemoteEndPointClosing _ _ ->
892892 -- If the remote endpoint is in Closing state, that means that are as
893893 -- far as we are concerned there are no incoming connections. This
894894 -- means that a CloseConnection request at this point is invalid.
895− throwIO $ userError "Invalid CloseConnection request"
895+ throwIO $ userError "Invalid CloseConnection request"
896896 RemoteEndPointFailed err ->
897897 throwIO err
898898 RemoteEndPointClosed ->
@@ -900,7 +900,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
900900 writeChan ourChannel (ConnectionClosed (connId lcid))
901901
902902 -- Close the socket (if we don't have any outgoing connections)
903− closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool
903+ closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool
904904 closeSocket sock lastReceivedId = do
905905 mAct <- modifyMVar theirState $ \st -> do
906906 case st of
@@ -908,38 +908,38 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
908908 relyViolation (ourEndPoint, theirEndPoint)
909909 "handleIncomingMessages:closeSocket (invalid)"
910910 RemoteEndPointInit _ _ _ ->
911− relyViolation (ourEndPoint, theirEndPoint)
911+ relyViolation (ourEndPoint, theirEndPoint)
912912 "handleIncomingMessages:closeSocket (init)"
913913 RemoteEndPointValid vst -> do
914914 -- We regard a CloseSocket message as an (optimized) way for the
915915 -- remote endpoint to indicate that all its connections to us are
916916 -- now properly closed
917− forM_ (Set.elems $ vst ^. remoteIncoming) $
917+ forM_ (Set.elems $ vst ^. remoteIncoming) $
918918 writeChan ourChannel . ConnectionClosed . connId
919− let vst' = remoteIncoming ^= Set.empty $ vst
919+ let vst' = remoteIncoming ^= Set.empty $ vst
920920 -- If we still have outgoing connections then we ignore the
921921 -- CloseSocket request (we sent a ConnectionCreated message to the
922922 -- remote endpoint, but it did not receive it before sending the
923923 -- CloseSocket request). Similarly, if lastReceivedId < lastSentId
924924 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
925925 -- message to the remote endpoint, *both of which* it did not yet
926− -- receive before sending the CloseSocket request.
927− if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
926+ -- receive before sending the CloseSocket request.
927+ if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
928928 then
929929 return (RemoteEndPointValid vst', Nothing)
930− else do
930+ else do
931931 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
932932 -- Attempt to reply (but don't insist)
933933 act <- schedule theirEndPoint $ do
934934 tryIO $ sendOn vst' [ encodeInt32 CloseSocket
935935 , encodeInt32 (vst ^. remoteMaxIncoming)
936936 ]
937− tryCloseSocket sock
937+ tryCloseSocket sock
938938 return (RemoteEndPointClosed, Just act)
939939 RemoteEndPointClosing resolved vst -> do
940940 -- Like above, we need to check if there is a ConnectionCreated
941− -- message that we sent but that the remote endpoint has not yet
942− -- received. However, since we are in 'closing' state, the only
941+ -- message that we sent but that the remote endpoint has not yet
942+ -- received. However, since we are in 'closing' state, the only
943943 -- way this may happen is when we sent a ConnectionCreated,
944944 -- ConnectionClosed, and CloseSocket message, none of which have
945945 -- yet been received. We leave the endpoint in closing state in
@@ -949,29 +949,29 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
949949 return (RemoteEndPointClosing resolved vst, Nothing)
950950 else do
951951 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
952− act <- schedule theirEndPoint $ tryCloseSocket sock
952+ act <- schedule theirEndPoint $ tryCloseSocket sock
953953 putMVar resolved ()
954954 return (RemoteEndPointClosed, Just act)
955955 RemoteEndPointFailed err ->
956956 throwIO err
957957 RemoteEndPointClosed ->
958− relyViolation (ourEndPoint, theirEndPoint)
958+ relyViolation (ourEndPoint, theirEndPoint)
959959 "handleIncomingMessages:closeSocket (closed)"
960960 case mAct of
961961 Nothing -> return False
962962 Just act -> do
963963 runScheduledAction (ourEndPoint, theirEndPoint) act
964964 return True
965−
965+
966966 -- Read a message and output it on the endPoint's channel. By rights we
967967 -- should verify that the connection ID is valid, but this is unnecessary
968968 -- overhead
969− readMessage :: N.Socket -> LightweightConnectionId -> IO ()
970− readMessage sock lcid =
971− recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)
969+ readMessage :: N.Socket -> LightweightConnectionId -> IO ()
970+ readMessage sock lcid =
971+ recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)
972972
973973 -- Arguments
974− ourChannel = localChannel ourEndPoint
974+ ourChannel = localChannel ourEndPoint
975975 theirState = remoteState theirEndPoint
976976 theirAddr = remoteAddress theirEndPoint
977977
@@ -982,20 +982,20 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
982982 modifyMVar_ theirState $ \st ->
983983 case st of
984984 RemoteEndPointInvalid _ ->
985− relyViolation (ourEndPoint, theirEndPoint)
985+ relyViolation (ourEndPoint, theirEndPoint)
986986 "handleIncomingMessages:prematureExit"
987987 RemoteEndPointInit _ _ _ ->
988− relyViolation (ourEndPoint, theirEndPoint)
988+ relyViolation (ourEndPoint, theirEndPoint)
989989 "handleIncomingMessages:prematureExit"
990990 RemoteEndPointValid _ -> do
991− let code = EventConnectionLost (remoteAddress theirEndPoint)
991+ let code = EventConnectionLost (remoteAddress theirEndPoint)
992992 writeChan ourChannel . ErrorEvent $ TransportError code (show err)
993993 return (RemoteEndPointFailed err)
994994 RemoteEndPointClosing resolved _ -> do
995995 putMVar resolved ()
996996 return (RemoteEndPointFailed err)
997997 RemoteEndPointClosed ->
998− relyViolation (ourEndPoint, theirEndPoint)
998+ relyViolation (ourEndPoint, theirEndPoint)
999999 "handleIncomingMessages:prematureExit"
10001000 RemoteEndPointFailed err' ->
10011001 return (RemoteEndPointFailed err')
@@ -1006,7 +1006,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
10061006
10071007 -- The ID of the last connection _we_ created (or 0 for none)
10081008 lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId
1009− lastSentId vst =
1009+ lastSentId vst =
10101010 if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId
10111011 then 0
10121012 else (vst ^. remoteNextConnOutId) - 1
@@ -1021,12 +1021,12 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
10211021-- | Create a connection to a remote endpoint
10221022--
10231023-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will
1024−-- block until that is resolved.
1024+-- block until that is resolved.
10251025--
10261026-- May throw a TransportError ConnectErrorCode exception.
1027−createConnectionTo :: TCPParameters
1028− -> LocalEndPoint
1029− -> EndPointAddress
1027+createConnectionTo :: TCPParameters
1028+ -> LocalEndPoint
1029+ -> EndPointAddress
10301030 -> ConnectHints
10311031 -> IO (RemoteEndPoint, LightweightConnectionId)
10321032createConnectionTo params ourEndPoint theirAddress hints = go
@@ -1035,27 +1035,27 @@ createConnectionTo params ourEndPoint theirAddress hints = go
10351035 (theirEndPoint, isNew) <- mapIOException connectFailed $
10361036 findRemoteEndPoint ourEndPoint theirAddress RequestedByUs
10371037
1038− if isNew
1038+ if isNew
10391039 then do
1040− forkIO . handle absorbAllExceptions $
1041− setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
1040+ forkIO . handle absorbAllExceptions $
1041+ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
10421042 go
10431043 else do
10441044 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
10451045 mapIOException connectFailed $ do
10461046 act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1047− RemoteEndPointValid vst -> do
1048− let connId = vst ^. remoteNextConnOutId
1047+ RemoteEndPointValid vst -> do
1048+ let connId = vst ^. remoteNextConnOutId
10491049 act <- schedule theirEndPoint $ do
10501050 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
10511051 return connId
1052− return ( RemoteEndPointValid
1053− $ remoteNextConnOutId ^= connId + 1
1052+ return ( RemoteEndPointValid
1053+ $ remoteNextConnOutId ^= connId + 1
10541054 $ vst
1055− , act
1055+ , act
10561056 )
10571057 -- Error cases
1058− RemoteEndPointInvalid err ->
1058+ RemoteEndPointInvalid err ->
10591059 throwIO err
10601060 RemoteEndPointFailed err ->
10611061 throwIO err
@@ -1065,32 +1065,32 @@ createConnectionTo params ourEndPoint theirAddress hints = go
10651065 -- TODO: deal with exception case?
10661066 connId <- runScheduledAction (ourEndPoint, theirEndPoint) act
10671067 return (theirEndPoint, connId)
1068−
1068+
10691069
10701070 connectFailed :: IOException -> TransportError ConnectErrorCode
10711071 connectFailed = TransportError ConnectFailed . show
10721072
10731073 absorbAllExceptions :: SomeException -> IO ()
1074− absorbAllExceptions _ex =
1074+ absorbAllExceptions _ex =
10751075 return ()
10761076
10771077-- | Set up a remote endpoint
1078−setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
1078+setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
10791079setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1080− result <- socketToEndPoint ourAddress
1081− theirAddress
1080+ result <- socketToEndPoint ourAddress
1081+ theirAddress
10821082 (tcpReuseClientAddr params)
10831083 (connectTimeout hints)
10841084 didAccept <- case result of
1085− Right (sock, ConnectionRequestAccepted) -> do
1086− sendLock <- newMVar ()
1087− let vst = ValidRemoteEndPointState
1085+ Right (sock, ConnectionRequestAccepted) -> do
1086+ sendLock <- newMVar ()
1087+ let vst = ValidRemoteEndPointState
10881088 { remoteSocket = sock
10891089 , remoteSendLock = sendLock
1090− , _remoteOutgoing = 0
1090+ , _remoteOutgoing = 0
10911091 , _remoteIncoming = Set.empty
10921092 , _remoteMaxIncoming = 0
1093− , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
1093+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
10941094 }
10951095 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
10961096 return True
@@ -1101,7 +1101,7 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
11011101 return False
11021102 Right (sock, ConnectionRequestCrossed) -> do
11031103 withMVar (remoteState theirEndPoint) $ \st -> case st of
1104− RemoteEndPointInit _ crossed _ ->
1104+ RemoteEndPointInit _ crossed _ ->
11051105 putMVar crossed ()
11061106 RemoteEndPointFailed ex ->
11071107 throwIO ex
@@ -1109,11 +1109,11 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
11091109 relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"
11101110 tryCloseSocket sock
11111111 return False
1112− Left err -> do
1112+ Left err -> do
11131113 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
11141114 return False
11151115
1116− when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
1116+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
11171117 where
11181118 ourAddress = localAddress ourEndPoint
11191119 theirAddress = remoteAddress theirEndPoint
@@ -1121,23 +1121,23 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
11211121
11221122-- | Send a CloseSocket request if the remote endpoint is unused
11231123closeIfUnused :: EndPointPair -> IO ()
1124−closeIfUnused (ourEndPoint, theirEndPoint) = do
1124+closeIfUnused (ourEndPoint, theirEndPoint) = do
11251125 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
11261126 RemoteEndPointValid vst ->
1127− if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)
1128− then do
1127+ if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)
1128+ then do
11291129 resolved <- newEmptyMVar
1130− act <- schedule theirEndPoint $
1130+ act <- schedule theirEndPoint $
11311131 sendOn vst [ encodeInt32 CloseSocket
11321132 , encodeInt32 (vst ^. remoteMaxIncoming)
11331133 ]
11341134 return (RemoteEndPointClosing resolved vst, Just act)
1135− else
1135+ else
11361136 return (RemoteEndPointValid vst, Nothing)
11371137 _ ->
11381138 return (st, Nothing)
1139− forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
1140−
1139+ forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
1140+
11411141-- | Reset a remote endpoint if it is in Invalid mode
11421142--
11431143-- If the remote endpoint is currently in broken state, and
@@ -1156,7 +1156,7 @@ resetIfBroken ourEndPoint theirAddress = do
11561156 return (vst ^. localConnectionTo theirAddress)
11571157 LocalEndPointClosed ->
11581158 throwIO $ TransportError ConnectFailed "Endpoint closed"
1159− forM_ mTheirEndPoint $ \theirEndPoint ->
1159+ forM_ mTheirEndPoint $ \theirEndPoint ->
11601160 withMVar (remoteState theirEndPoint) $ \st -> case st of
11611161 RemoteEndPointInvalid _ ->
11621162 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
@@ -1166,26 +1166,26 @@ resetIfBroken ourEndPoint theirAddress = do
11661166 return ()
11671167
11681168-- | Special case of 'apiConnect': connect an endpoint to itself
1169−--
1169+--
11701170-- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)
1171−connectToSelf :: LocalEndPoint
1171+connectToSelf :: LocalEndPoint
11721172 -> IO Connection
1173−connectToSelf ourEndPoint = do
1173+connectToSelf ourEndPoint = do
11741174 connAlive <- newIORef True -- Protected by the local endpoint lock
1175− lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
1175+ lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
11761176 let connId = createConnectionId heavyweightSelfConnectionId lconnId
11771177 writeChan ourChan $
11781178 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
1179− return Connection
1180− { send = selfSend connAlive connId
1179+ return Connection
1180+ { send = selfSend connAlive connId
11811181 , close = selfClose connAlive connId
11821182 }
11831183 where
1184− selfSend :: IORef Bool
1185− -> ConnectionId
1186− -> [ByteString]
1184+ selfSend :: IORef Bool
1185+ -> ConnectionId
1186+ -> [ByteString]
11871187 -> IO (Either (TransportError SendErrorCode) ())
1188− selfSend connAlive connId msg =
1188+ selfSend connAlive connId msg =
11891189 try . withMVar ourState $ \st -> case st of
11901190 LocalEndPointValid _ -> do
11911191 alive <- readIORef connAlive
@@ -1196,19 +1196,19 @@ connectToSelf ourEndPoint = do
11961196 throwIO $ TransportError SendFailed "Endpoint closed"
11971197
11981198 selfClose :: IORef Bool -> ConnectionId -> IO ()
1199− selfClose connAlive connId =
1199+ selfClose connAlive connId =
12001200 withMVar ourState $ \st -> case st of
12011201 LocalEndPointValid _ -> do
12021202 alive <- readIORef connAlive
12031203 when alive $ do
1204− writeChan ourChan (ConnectionClosed connId)
1204+ writeChan ourChan (ConnectionClosed connId)
12051205 writeIORef connAlive False
12061206 LocalEndPointClosed ->
1207− return ()
1207+ return ()
12081208
12091209 ourChan = localChannel ourEndPoint
12101210 ourState = localState ourEndPoint
1211− connectFailed = TransportError ConnectFailed . show
1211+ connectFailed = TransportError ConnectFailed . show
12121212
12131213-- | Resolve an endpoint currently in 'Init' state
12141214resolveInit :: EndPointPair -> RemoteState -> IO ()
@@ -1216,57 +1216,57 @@ resolveInit (ourEndPoint, theirEndPoint) newState =
12161216 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
12171217 RemoteEndPointInit resolved _ _ -> do
12181218 putMVar resolved ()
1219− case newState of
1220− RemoteEndPointClosed ->
1219+ case newState of
1220+ RemoteEndPointClosed ->
12211221 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
12221222 _ ->
12231223 return ()
12241224 return newState
1225− RemoteEndPointFailed ex ->
1225+ RemoteEndPointFailed ex ->
12261226 throwIO ex
12271227 _ ->
12281228 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
12291229
12301230-- | Get the next outgoing self-connection ID
1231−--
1231+--
12321232-- Throws an IO exception when the endpoint is closed.
12331233getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1234−getLocalNextConnOutId ourEndpoint =
1234+getLocalNextConnOutId ourEndpoint =
12351235 modifyMVar (localState ourEndpoint) $ \st -> case st of
12361236 LocalEndPointValid vst -> do
1237− let connId = vst ^. localNextConnOutId
1238− return ( LocalEndPointValid
1239− . (localNextConnOutId ^= connId + 1)
1237+ let connId = vst ^. localNextConnOutId
1238+ return ( LocalEndPointValid
1239+ . (localNextConnOutId ^= connId + 1)
12401240 $ vst
12411241 , connId)
12421242 LocalEndPointClosed ->
12431243 throwIO $ userError "Local endpoint closed"
12441244
12451245-- | Create a new local endpoint
1246−--
1246+--
12471247-- May throw a TransportError NewEndPointErrorCode exception if the transport
12481248-- is closed.
12491249createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
1250−createLocalEndPoint transport = do
1250+createLocalEndPoint transport = do
12511251 chan <- newChan
1252− state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1252+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
12531253 { _localNextConnOutId = firstNonReservedLightweightConnectionId
12541254 , _localConnections = Map.empty
1255− , _nextConnInId = firstNonReservedHeavyweightConnectionId
1255+ , _nextConnInId = firstNonReservedHeavyweightConnectionId
12561256 }
12571257 modifyMVar (transportState transport) $ \st -> case st of
12581258 TransportValid vst -> do
12591259 let ix = vst ^. nextEndPointId
1260− let addr = encodeEndPointAddress (transportHost transport)
1260+ let addr = encodeEndPointAddress (transportHost transport)
12611261 (transportPort transport)
1262− ix
1262+ ix
12631263 let localEndPoint = LocalEndPoint { localAddress = addr
12641264 , localChannel = chan
12651265 , localState = state
12661266 }
1267− return ( TransportValid
1268− . (localEndPointAt addr ^= Just localEndPoint)
1269− . (nextEndPointId ^= ix + 1)
1267+ return ( TransportValid
1268+ . (localEndPointAt addr ^= Just localEndPoint)
1269+ . (nextEndPointId ^= ix + 1)
12701270 $ vst
12711271 , localEndPoint
12721272 )
@@ -1282,13 +1282,13 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
12821282 modifyMVar_ ourState $ \st -> case st of
12831283 LocalEndPointValid vst ->
12841284 case vst ^. localConnectionTo theirAddress of
1285− Nothing ->
1285+ Nothing ->
12861286 return st
12871287 Just remoteEndPoint' ->
1288− if remoteId remoteEndPoint' == remoteId theirEndPoint
1289− then return
1290− ( LocalEndPointValid
1291− . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
1288+ if remoteId remoteEndPoint' == remoteId theirEndPoint
1289+ then return
1290+ ( LocalEndPointValid
1291+ . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
12921292 $ vst
12931293 )
12941294 else return st
@@ -1302,23 +1302,23 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
13021302--
13031303-- Does nothing if the transport is closed
13041304removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
1305−removeLocalEndPoint transport ourEndPoint =
1306− modifyMVar_ (transportState transport) $ \st -> case st of
1305+removeLocalEndPoint transport ourEndPoint =
1306+ modifyMVar_ (transportState transport) $ \st -> case st of
13071307 TransportValid vst ->
1308− return ( TransportValid
1309− . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
1308+ return ( TransportValid
1309+ . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
13101310 $ vst
13111311 )
13121312 TransportClosed ->
13131313 return TransportClosed
13141314
13151315-- | Find a remote endpoint. If the remote endpoint does not yet exist we
1316−-- create it in Init state. Returns if the endpoint was new.
1317−findRemoteEndPoint
1316+-- create it in Init state. Returns if the endpoint was new.
1317+findRemoteEndPoint
13181318 :: LocalEndPoint
13191319 -> EndPointAddress
1320− -> RequestedBy
1321− -> IO (RemoteEndPoint, Bool)
1320+ -> RequestedBy
1321+ -> IO (RemoteEndPoint, Bool)
13221322findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
13231323 where
13241324 go = do
@@ -1335,55 +1335,55 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
13351335 { remoteAddress = theirAddress
13361336 , remoteState = theirState
13371337 , remoteId = vst ^. nextConnInId
1338− , remoteScheduled = scheduled
1338+ , remoteScheduled = scheduled
13391339 }
1340− return ( LocalEndPointValid
1341− . (localConnectionTo theirAddress ^= Just theirEndPoint)
1342− . (nextConnInId ^: (+ 1))
1340+ return ( LocalEndPointValid
1341+ . (localConnectionTo theirAddress ^= Just theirEndPoint)
1342+ . (nextConnInId ^: (+ 1))
13431343 $ vst
1344− , (theirEndPoint, True)
1344+ , (theirEndPoint, True)
13451345 )
13461346 LocalEndPointClosed ->
13471347 throwIO $ userError "Local endpoint closed"
1348−
1349− if isNew
1348+
1349+ if isNew
13501350 then
13511351 return (theirEndPoint, True)
13521352 else do
13531353 let theirState = remoteState theirEndPoint
13541354 snapshot <- modifyMVar theirState $ \st -> case st of
1355− RemoteEndPointValid vst ->
1355+ RemoteEndPointValid vst ->
13561356 case findOrigin of
13571357 RequestedByUs -> do
1358− let st' = RemoteEndPointValid
1359− . (remoteOutgoing ^: (+ 1))
1360− $ vst
1358+ let st' = RemoteEndPointValid
1359+ . (remoteOutgoing ^: (+ 1))
1360+ $ vst
13611361 return (st', st')
13621362 RequestedByThem ->
1363− return (st, st)
1363+ return (st, st)
13641364 _ ->
13651365 return (st, st)
13661366 -- The snapshot may no longer be up to date at this point, but if we
1367− -- increased the refcount then it can only either be Valid or Failed
1368− -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
1367+ -- increased the refcount then it can only either be Valid or Failed
1368+ -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
13691369 case snapshot of
13701370 RemoteEndPointInvalid err ->
13711371 throwIO err
13721372 RemoteEndPointInit resolved crossed initOrigin ->
13731373 case (findOrigin, initOrigin) of
1374− (RequestedByUs, RequestedByUs) ->
1375− readMVar resolved >> go
1376− (RequestedByUs, RequestedByThem) ->
1374+ (RequestedByUs, RequestedByUs) ->
13771375 readMVar resolved >> go
1378− (RequestedByThem, RequestedByUs) ->
1379− if ourAddress > theirAddress
1376+ (RequestedByUs, RequestedByThem) ->
1377+ readMVar resolved >> go
1378+ (RequestedByThem, RequestedByUs) ->
1379+ if ourAddress > theirAddress
13801380 then do
13811381 -- Wait for the Crossed message
1382− readMVar crossed
1382+ readMVar crossed
13831383 return (theirEndPoint, True)
13841384 else
13851385 return (theirEndPoint, False)
1386− (RequestedByThem, RequestedByThem) ->
1386+ (RequestedByThem, RequestedByThem) ->
13871387 throwIO $ userError "Already connected"
13881388 RemoteEndPointValid _ ->
13891389 -- We assume that the request crossed if we find the endpoint in
@@ -1395,15 +1395,15 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
13951395 readMVar resolved >> go
13961396 RemoteEndPointClosed ->
13971397 go
1398− RemoteEndPointFailed err ->
1398+ RemoteEndPointFailed err ->
13991399 throwIO err
1400−
1401− ourState = localState ourEndPoint
1400+
1401+ ourState = localState ourEndPoint
14021402 ourAddress = localAddress ourEndPoint
14031403
14041404-- | Send a payload over a heavyweight connection (thread safe)
1405−sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()
1406−sendOn vst bs = withMVar (remoteSendLock vst) $ \() ->
1405+sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()
1406+sendOn vst bs = withMVar (remoteSendLock vst) $ \() ->
14071407 sendMany (remoteSocket vst) bs
14081408
14091409--------------------------------------------------------------------------------
@@ -1417,7 +1417,7 @@ type Action a = MVar (Either SomeException a)
14171417schedule :: RemoteEndPoint -> IO a -> IO (Action a)
14181418schedule theirEndPoint act = do
14191419 mvar <- newEmptyMVar
1420− writeChan (remoteScheduled theirEndPoint) $
1420+ writeChan (remoteScheduled theirEndPoint) $
14211421 catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)
14221422 return mvar
14231423
@@ -1429,7 +1429,7 @@ schedule theirEndPoint act = do
14291429-- > runScheduledAction
14301430--
14311431-- 'runScheduledAction' will run @p@ (it might run some other scheduled action).
1432−-- However, it will then wait until @p@ is executed (by this call to
1432+-- However, it will then wait until @p@ is executed (by this call to
14331433-- 'runScheduledAction' or by another).
14341434runScheduledAction :: EndPointPair -> Action a -> IO a
14351435runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
@@ -1438,21 +1438,21 @@ runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
14381438 case ma of
14391439 Right a -> return a
14401440 Left e -> do
1441− forM_ (fromException e) $ \ioe ->
1441+ forM_ (fromException e) $ \ioe ->
14421442 modifyMVar_ (remoteState theirEndPoint) $ \st ->
1443− case st of
1443+ case st of
14441444 RemoteEndPointValid vst -> handleIOException ioe vst
14451445 _ -> return (RemoteEndPointFailed ioe)
14461446 throwIO e
14471447 where
1448− handleIOException :: IOException
1449− -> ValidRemoteEndPointState
1450− -> IO RemoteState
1448+ handleIOException :: IOException
1449+ -> ValidRemoteEndPointState
1450+ -> IO RemoteState
14511451 handleIOException ex vst = do
14521452 tryCloseSocket (remoteSocket vst)
1453− let code = EventConnectionLost (remoteAddress theirEndPoint)
1453+ let code = EventConnectionLost (remoteAddress theirEndPoint)
14541454 err = TransportError code (show ex)
1455− writeChan (localChannel ourEndPoint) $ ErrorEvent err
1455+ writeChan (localChannel ourEndPoint) $ ErrorEvent err
14561456 return (RemoteEndPointFailed ex)
14571457
14581458--------------------------------------------------------------------------------
@@ -1462,25 +1462,25 @@ runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
14621462-- | Establish a connection to a remote endpoint
14631463--
14641464-- Maybe throw a TransportError
1465−socketToEndPoint :: EndPointAddress -- ^ Our address
1465+socketToEndPoint :: EndPointAddress -- ^ Our address
14661466 -> EndPointAddress -- ^ Their address
14671467 -> Bool -- ^ Use SO_REUSEADDR?
1468− -> Maybe Int -- ^ Timeout for connect
1469− -> IO (Either (TransportError ConnectErrorCode)
1470− (N.Socket, ConnectionRequestResponse))
1471−socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1472− try $ do
1473− (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
1468+ -> Maybe Int -- ^ Timeout for connect
1469+ -> IO (Either (TransportError ConnectErrorCode)
1470+ (N.Socket, ConnectionRequestResponse))
1471+socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1472+ try $ do
1473+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
14741474 Nothing -> throwIO (failed . userError $ "Could not parse")
14751475 Just dec -> return dec
1476− addr:_ <- mapIOException invalidAddress $
1476+ addr:_ <- mapIOException invalidAddress $
14771477 N.getAddrInfo Nothing (Just host) (Just port)
14781478 bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
1479− when reuseAddr $
1479+ when reuseAddr $
14801480 mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
1481− mapIOException invalidAddress $
1482− timeoutMaybe timeout timeoutError $
1483− N.connect sock (N.addrAddress addr)
1481+ mapIOException invalidAddress $
1482+ timeoutMaybe timeout timeoutError $
1483+ N.connect sock (N.addrAddress addr)
14841484 response <- mapIOException failed $ do
14851485 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
14861486 recvInt32 sock
@@ -1489,50 +1489,50 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
14891489 Just r -> return (sock, r)
14901490 where
14911491 createSocket :: N.AddrInfo -> IO N.Socket
1492− createSocket addr = mapIOException insufficientResources $
1492+ createSocket addr = mapIOException insufficientResources $
14931493 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
14941494
1495− invalidAddress = TransportError ConnectNotFound . show
1496− insufficientResources = TransportError ConnectInsufficientResources . show
1495+ invalidAddress = TransportError ConnectNotFound . show
1496+ insufficientResources = TransportError ConnectInsufficientResources . show
14971497 failed = TransportError ConnectFailed . show
14981498 timeoutError = TransportError ConnectTimeout "Timed out"
14991499
15001500-- | Encode end point address
1501−encodeEndPointAddress :: N.HostName
1502− -> N.ServiceName
1503− -> EndPointId
1501+encodeEndPointAddress :: N.HostName
1502+ -> N.ServiceName
1503+ -> EndPointId
15041504 -> EndPointAddress
15051505encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
1506− host ++ ":" ++ port ++ ":" ++ show ix
1506+ host ++ ":" ++ port ++ ":" ++ show ix
15071507
15081508-- | Decode end point address
1509−decodeEndPointAddress :: EndPointAddress
1509+decodeEndPointAddress :: EndPointAddress
15101510 -> Maybe (N.HostName, N.ServiceName, EndPointId)
1511−decodeEndPointAddress (EndPointAddress bs) =
1511+decodeEndPointAddress (EndPointAddress bs) =
15121512 case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of
1513− [host, port, endPointIdStr] ->
1514− case reads endPointIdStr of
1513+ [host, port, endPointIdStr] ->
1514+ case reads endPointIdStr of
15151515 [(endPointId, "")] -> Just (host, port, endPointId)
15161516 _ -> Nothing
15171517 _ ->
15181518 Nothing
15191519
15201520-- | Construct a ConnectionId
1521−createConnectionId :: HeavyweightConnectionId
1522− -> LightweightConnectionId
1521+createConnectionId :: HeavyweightConnectionId
1522+ -> LightweightConnectionId
15231523 -> ConnectionId
1524−createConnectionId hcid lcid =
1524+createConnectionId hcid lcid =
15251525 (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid
15261526
15271527-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,
15281528-- returning at most @p@ segments -- counting from the /end/
1529−--
1529+--
15301530-- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"]
15311531splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]]
1532−splitMaxFromEnd p = \n -> go [[]] n . reverse
1532+splitMaxFromEnd p = \n -> go [[]] n . reverse
15331533 where
15341534 -- go :: [[a]] -> Int -> [a] -> [[a]]
1535− go accs _ [] = accs
1535+ go accs _ [] = accs
15361536 go ([] : accs) 0 xs = reverse xs : accs
15371537 go (acc : accs) n (x:xs) =
15381538 if p x then go ([] : acc : accs) (n - 1) xs
@@ -1544,40 +1544,40 @@ splitMaxFromEnd p = \n -> go [[]] n . reverse
15441544--------------------------------------------------------------------------------
15451545
15461546-- Find a socket between two endpoints
1547−--
1547+--
15481548-- Throws an IO exception if the socket could not be found.
1549−internalSocketBetween :: TCPTransport -- ^ Transport
1549+internalSocketBetween :: TCPTransport -- ^ Transport
15501550 -> EndPointAddress -- ^ Local endpoint
15511551 -> EndPointAddress -- ^ Remote endpoint
1552− -> IO N.Socket
1552+ -> IO N.Socket
15531553internalSocketBetween transport ourAddress theirAddress = do
15541554 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1555− TransportClosed ->
1556− throwIO $ userError "Transport closed"
1557− TransportValid vst ->
1555+ TransportClosed ->
1556+ throwIO $ userError "Transport closed"
1557+ TransportValid vst ->
15581558 case vst ^. localEndPointAt ourAddress of
15591559 Nothing -> throwIO $ userError "Local endpoint not found"
15601560 Just ep -> return ep
15611561 theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
15621562 LocalEndPointClosed ->
15631563 throwIO $ userError "Local endpoint closed"
1564− LocalEndPointValid vst ->
1564+ LocalEndPointValid vst ->
15651565 case vst ^. localConnectionTo theirAddress of
15661566 Nothing -> throwIO $ userError "Remote endpoint not found"
15671567 Just ep -> return ep
15681568 withMVar (remoteState theirEndPoint) $ \st -> case st of
15691569 RemoteEndPointInit _ _ _ ->
15701570 throwIO $ userError "Remote endpoint not yet initialized"
1571− RemoteEndPointValid vst ->
1571+ RemoteEndPointValid vst ->
15721572 return $ remoteSocket vst
15731573 RemoteEndPointClosing _ vst ->
1574− return $ remoteSocket vst
1574+ return $ remoteSocket vst
15751575 RemoteEndPointClosed ->
15761576 throwIO $ userError "Remote endpoint closed"
15771577 RemoteEndPointInvalid err ->
1578− throwIO err
1578+ throwIO err
15791579 RemoteEndPointFailed err ->
1580− throwIO err
1580+ throwIO err
15811581
15821582--------------------------------------------------------------------------------
15831583-- Constants --
@@ -1611,7 +1611,7 @@ localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextCon
16111611localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
16121612localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
16131613
1614−nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId
1614+nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId
16151615nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })
16161616
16171617remoteOutgoing :: Accessor ValidRemoteEndPointState Int
@@ -1627,10 +1627,10 @@ remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId
16271627remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })
16281628
16291629localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1630−localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1630+localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
16311631
16321632localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)
1633−localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1633+localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
16341634
16351635-------------------------------------------------------------------------------
16361636-- Debugging --
@@ -1638,14 +1638,14 @@ localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
16381638
16391639relyViolation :: EndPointPair -> String -> IO a
16401640relyViolation (ourEndPoint, theirEndPoint) str = do
1641− elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation")
1641+ elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation")
16421642 fail (str ++ " RELY violation")
16431643
16441644elog :: EndPointPair -> String -> IO ()
16451645elog (ourEndPoint, theirEndPoint) msg = do
16461646 tid <- myThreadId
1647− putStrLn $ show (localAddress ourEndPoint)
1648− ++ "/" ++ show (remoteAddress theirEndPoint)
1647+ putStrLn $ show (localAddress ourEndPoint)
1648+ ++ "/" ++ show (remoteAddress theirEndPoint)
16491649 ++ "(" ++ show (remoteId theirEndPoint) ++ ")"
1650− ++ "/" ++ show tid
1650+ ++ "/" ++ show tid
16511651 ++ ": " ++ msg