@@ -1,22 +1,22 @@
1−-- | TCP implementation of the transport layer.
1+-- | TCP implementation of the transport layer.
3-- The TCP implementation guarantees that only a single TCP connection (socket)
3-- The TCP implementation guarantees that only a single TCP connection (socket)
4-- will be used between endpoints, provided that the addresses specified are
4-- will be used between endpoints, provided that the addresses specified are
5-- canonical. If /A/ connects to /B/ and reports its address as
5-- canonical. If /A/ connects to /B/ and reports its address as
6-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
6-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
7-- @client1.local:http-alt@ then the transport layer will not realize that the
7-- @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.
10-- Applications that use the TCP transport should use
10-- Applications that use the TCP transport should use
11-- 'Network.Socket.withSocketsDo' in their main function for Windows
11-- 'Network.Socket.withSocketsDo' in their main function for Windows
12-- compatibility (see "Network.Socket").
12-- compatibility (see "Network.Socket").
13−module Network.Transport.TCP
13+module Network.Transport.TCP
14 ( -- * Main API
14 ( -- * Main API
15 createTransport
15 createTransport
16 , TCPParameters(..)
16 , TCPParameters(..)
17 , defaultTCPParameters
17 , defaultTCPParameters
18− -- * Internals (exposed for unit tests)
18+ -- * Internals (exposed for unit tests)
19− , createTransportExposeInternals
19+ , createTransportExposeInternals
20 , TransportInternals(..)
20 , TransportInternals(..)
21 , EndPointId
21 , EndPointId
22 , encodeEndPointAddress
22 , encodeEndPointAddress
@@ -25,27 +25,27 @@ module Network.Transport.TCP
25 , ConnectionRequestResponse(..)
25 , ConnectionRequestResponse(..)
26 , firstNonReservedLightweightConnectionId
26 , firstNonReservedLightweightConnectionId
27 , firstNonReservedHeavyweightConnectionId
27 , firstNonReservedHeavyweightConnectionId
28− , socketToEndPoint
28+ , socketToEndPoint
29 , LightweightConnectionId
29 , LightweightConnectionId
30 -- * Design notes
30 -- * Design notes
31 -- $design
31 -- $design
34−import Prelude hiding
34+import Prelude hiding
36#if ! MIN_VERSION_base(4,6,0)
36#if ! MIN_VERSION_base(4,6,0)
41import Network.Transport
41import Network.Transport
42−import Network.Transport.TCP.Internal
42+import Network.Transport.TCP.Internal
43 ( forkServer
43 ( forkServer
44 , recvWithLength
44 , recvWithLength
45 , recvInt32
45 , recvInt32
46 , tryCloseSocket
46 , tryCloseSocket
48−import Network.Transport.Internal
48+import Network.Transport.Internal
49 ( encodeInt32
49 ( encodeInt32
50 , prependLength
50 , prependLength
51 , mapIOException
51 , mapIOException
@@ -59,7 +59,7 @@ import Network.Transport.Internal
59#ifdef USE_MOCK_NETWORK
59#ifdef USE_MOCK_NETWORK
60import qualified Network.Transport.TCP.Mock.Socket as N
60import qualified Network.Transport.TCP.Mock.Socket as N
62−import qualified Network.Socket as N
62+import qualified Network.Socket as N
64 ( HostName
64 ( HostName
65 , ServiceName
65 , ServiceName
@@ -71,7 +71,7 @@ import qualified Network.Socket as N
71 , SocketType(Stream)
71 , SocketType(Stream)
72 , defaultProtocol
72 , defaultProtocol
73 , setSocketOption
73 , setSocketOption
74− , SocketOption(ReuseAddr)
74+ , SocketOption(ReuseAddr)
76 , sOMAXCONN
76 , sOMAXCONN
77 , AddrInfo
77 , AddrInfo
@@ -85,7 +85,7 @@ import Network.Socket.ByteString (sendMany)
86import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
86import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
87import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
87import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
88−import Control.Concurrent.MVar
88+import Control.Concurrent.MVar
91 , modifyMVar
91 , modifyMVar
@@ -98,7 +98,7 @@ import Control.Concurrent.MVar
98import Control.Category ((>>>))
98import Control.Category ((>>>))
99import Control.Applicative ((<$>))
99import Control.Applicative ((<$>))
100import Control.Monad (when, unless, join)
100import Control.Monad (when, unless, join)
101−import Control.Exception
101+import Control.Exception
102 ( IOException
102 ( IOException
103 , SomeException
103 , SomeException
104 , AsyncException
104 , AsyncException
@@ -117,7 +117,7 @@ import qualified Data.ByteString.Char8 as BSC (pack, unpack)
117import Data.Bits (shiftL, (.|.))
117import Data.Bits (shiftL, (.|.))
118import Data.Word (Word32)
118import Data.Word (Word32)
119import Data.Set (Set)
119import Data.Set (Set)
120−import qualified Data.Set as Set
120+import qualified Data.Set as Set
@@ -128,11 +128,11 @@ import qualified Data.Set as Set
129import Data.Map (Map)
129import Data.Map (Map)
130import qualified Data.Map as Map (empty)
130import qualified Data.Map as Map (empty)
131−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
131+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
132import qualified Data.Accessor.Container as DAC (mapMaybe)
132import qualified Data.Accessor.Container as DAC (mapMaybe)
133import Data.Foldable (forM_, mapM_)
133import Data.Foldable (forM_, mapM_)
135−-- $design
135+-- $design
137-- [Goals]
137-- [Goals]
@@ -146,7 +146,7 @@ import Data.Foldable (forM_, mapM_)
146-- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |
146-- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |
147-- > | |==========================| |
147-- > | |==========================| |
148-- > +-------+ +-------+
148-- > +-------+ +-------+
150-- Ignoring the complications detailed below, the TCP connection is set up is
150-- Ignoring the complications detailed below, the TCP connection is set up is
151-- when the first lightweight connection is created (in either direction), and
151-- when the first lightweight connection is created (in either direction), and
152-- torn down when the last lightweight connection (in either direction) is
152-- torn down when the last lightweight connection (in either direction) is
@@ -177,7 +177,7 @@ import Data.Foldable (forM_, mapM_)
177-- with /B/'s connection request completes.
177-- with /B/'s connection request completes.
179-- [Disconnecting]
179-- [Disconnecting]
181-- The TCP connection is created as soon as the first logical connection from
181-- The TCP connection is created as soon as the first logical connection from
182-- /A/ to /B/ (or /B/ to /A/) is established. At this point a thread (@#@) is
182-- /A/ to /B/ (or /B/ to /A/) is established. At this point a thread (@#@) is
183-- spawned that listens for incoming connections from /B/:
183-- spawned that listens for incoming connections from /B/:
@@ -193,21 +193,21 @@ import Data.Foldable (forM_, mapM_)
193-- The question is when the TCP connection can be closed again. Conceptually,
193-- The question is when the TCP connection can be closed again. Conceptually,
194-- we want to do reference counting: when there are no logical connections left
194-- we want to do reference counting: when there are no logical connections left
195-- between /A/ and /B/ we want to close the socket (possibly after some
195-- between /A/ and /B/ we want to close the socket (possibly after some
196−-- timeout).
196+-- timeout).
198-- However, /A/ and /B/ need to agree that the refcount has reached zero. It
198-- However, /A/ and /B/ need to agree that the refcount has reached zero. It
199-- might happen that /B/ sends a connection request over the existing socket at
199-- might happen that /B/ sends a connection request over the existing socket at
200-- the same time that /A/ closes its logical connection to /B/ and closes the
200-- the same time that /A/ closes its logical connection to /B/ and closes the
201-- socket. This will cause a failure in /B/ (which will have to retry) which is
201-- socket. This will cause a failure in /B/ (which will have to retry) which is
202-- not caused by a network failure, which is unfortunate. (Note that the
202-- 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.)
205-- Instead, when /A/ is ready to close the socket it sends a 'CloseSocket'
205-- Instead, when /A/ is ready to close the socket it sends a 'CloseSocket'
206-- request to /B/ and records that its connection to /B/ is closing. If /A/
206-- request to /B/ and records that its connection to /B/ is closing. If /A/
207-- receives a new connection request from /B/ after having sent the
207-- receives a new connection request from /B/ after having sent the
208-- 'CloseSocket' request it simply forgets that it sent a 'CloseSocket' request
208-- 'CloseSocket' request it simply forgets that it sent a 'CloseSocket' request
209-- and increments the reference count of the connection again.
209-- and increments the reference count of the connection again.
211-- When /B/ receives a 'CloseSocket' message and it too is ready to close the
211-- When /B/ receives a 'CloseSocket' message and it too is ready to close the
212-- connection, it will respond with a reciprocal 'CloseSocket' request to /A/
212-- connection, it will respond with a reciprocal 'CloseSocket' request to /A/
213-- and then actually close the socket. /A/ meanwhile will not send any more
213-- and then actually close the socket. /A/ meanwhile will not send any more
@@ -216,11 +216,11 @@ import Data.Foldable (forM_, mapM_)
216-- from /B/. (Since /A/ recorded that its connection to /B/ is in closing state
216-- from /B/. (Since /A/ recorded that its connection to /B/ is in closing state
217-- after sending a 'CloseSocket' request to /B/, it knows not to reciprocate /B/
217-- after sending a 'CloseSocket' request to /B/, it knows not to reciprocate /B/
218-- reciprocal 'CloseSocket' message.)
218-- reciprocal 'CloseSocket' message.)
220-- If there is a concurrent thread in /A/ waiting to connect to /B/ after /A/
220-- If there is a concurrent thread in /A/ waiting to connect to /B/ after /A/
221-- has sent a 'CloseSocket' request then this thread will block until /A/ knows
221-- has sent a 'CloseSocket' request then this thread will block until /A/ knows
222-- whether to reuse the old socket (if /B/ sends a new connection request
222-- 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.
225--------------------------------------------------------------------------------
225--------------------------------------------------------------------------------
226-- Internal datatypes --
226-- Internal datatypes --
@@ -229,30 +229,30 @@ import Data.Foldable (forM_, mapM_)
229-- We use underscores for fields that we might update (using accessors)
229-- We use underscores for fields that we might update (using accessors)
231-- All data types follow the same structure:
231-- All data types follow the same structure:
233-- * A top-level data type describing static properties (TCPTransport,
233-- * A top-level data type describing static properties (TCPTransport,
234-- LocalEndPoint, RemoteEndPoint)
234-- LocalEndPoint, RemoteEndPoint)
235-- * The 'static' properties include an MVar containing a data structure for
235-- * The 'static' properties include an MVar containing a data structure for
236-- the dynamic properties (TransportState, LocalEndPointState,
236-- 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.
238-- * For the case of "valid" we use third data structure to give more details
238-- * For the case of "valid" we use third data structure to give more details
239-- about the state (ValidTransportState, ValidLocalEndPointState,
239-- about the state (ValidTransportState, ValidLocalEndPointState,
240-- ValidRemoteEndPointState).
240-- ValidRemoteEndPointState).
242−data TCPTransport = TCPTransport
242+data TCPTransport = TCPTransport
243 { transportHost :: !N.HostName
243 { transportHost :: !N.HostName
244 , transportPort :: !N.ServiceName
244 , transportPort :: !N.ServiceName
245 , transportState :: !(MVar TransportState)
245 , transportState :: !(MVar TransportState)
246 , transportParams :: !TCPParameters
246 , transportParams :: !TCPParameters
249−data TransportState =
249+data TransportState =
250 TransportValid !ValidTransportState
250 TransportValid !ValidTransportState
251 | TransportClosed
251 | TransportClosed
253−data ValidTransportState = ValidTransportState
253+data ValidTransportState = ValidTransportState
254 { _localEndPoints :: !(Map EndPointAddress LocalEndPoint)
254 { _localEndPoints :: !(Map EndPointAddress LocalEndPoint)
255− , _nextEndPointId :: !EndPointId
255+ , _nextEndPointId :: !EndPointId
258data LocalEndPoint = LocalEndPoint
258data LocalEndPoint = LocalEndPoint
@@ -261,12 +261,12 @@ data LocalEndPoint = LocalEndPoint
261 , localState :: !(MVar LocalEndPointState)
261 , localState :: !(MVar LocalEndPointState)
264−data LocalEndPointState =
264+data LocalEndPointState =
265 LocalEndPointValid !ValidLocalEndPointState
265 LocalEndPointValid !ValidLocalEndPointState
266 | LocalEndPointClosed
266 | LocalEndPointClosed
268−data ValidLocalEndPointState = ValidLocalEndPointState
268+data ValidLocalEndPointState = ValidLocalEndPointState
269− { -- Next available ID for an outgoing lightweight self-connection
269+ { -- Next available ID for an outgoing lightweight self-connection
270 -- (see also remoteNextConnOutId)
270 -- (see also remoteNextConnOutId)
271 _localNextConnOutId :: !LightweightConnectionId
271 _localNextConnOutId :: !LightweightConnectionId
272 -- Next available ID for an incoming heavyweight connection
272 -- Next available ID for an incoming heavyweight connection
@@ -275,7 +275,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState
275 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
275 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
278−-- REMOTE ENDPOINTS
278+-- REMOTE ENDPOINTS
280-- Remote endpoints (basically, TCP connections) have the following lifecycle:
280-- Remote endpoints (basically, TCP connections) have the following lifecycle:
@@ -302,7 +302,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState
302-- createConnectionTo when we fail to connect.
302-- createConnectionTo when we fail to connect.
304-- Valid: This is the "normal" state for a working remote endpoint.
304-- Valid: This is the "normal" state for a working remote endpoint.
306-- Closing: When we detect that a remote endpoint is no longer used, we send a
306-- Closing: When we detect that a remote endpoint is no longer used, we send a
307-- CloseSocket request across the connection and put the remote endpoint in
307-- CloseSocket request across the connection and put the remote endpoint in
308-- closing state. As with Init, 'Closing' carries an MVar () 'resolved' which
308-- closing state. As with Init, 'Closing' carries an MVar () 'resolved' which
@@ -335,16 +335,16 @@ data ValidLocalEndPointState = ValidLocalEndPointState
335-- localConnections it is guaranteed to either find a different remote
335-- localConnections it is guaranteed to either find a different remote
336-- endpoint, or else none at all (if we don't insist in this order some
336-- endpoint, or else none at all (if we don't insist in this order some
337-- threads might start spinning).
337-- threads might start spinning).
339-- INV-RESOLVE: We should only signal on 'resolved' while the remote endpoint is
339-- INV-RESOLVE: We should only signal on 'resolved' while the remote endpoint is
340-- locked, and the remote endpoint must be in Valid or Closed state once
340-- locked, and the remote endpoint must be in Valid or Closed state once
341-- unlocked. This guarantees that there will not be two threads attempting to
341-- unlocked. This guarantees that there will not be two threads attempting to
342−-- both signal on 'resolved'.
342+-- both signal on 'resolved'.
344-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we
344-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we
345-- first put the remote endpoint in Closed state, and then send a
345-- first put the remote endpoint in Closed state, and then send a
346-- EventConnectionLost event. This guarantees that we only send this event
346-- EventConnectionLost event. This guarantees that we only send this event
347−-- once.
347+-- once.
349-- INV-CLOSING: An endpoint in closing state is for all intents and purposes
349-- INV-CLOSING: An endpoint in closing state is for all intents and purposes
350-- closed; that is, we shouldn't do any 'send's on it (although 'recv' is
350-- closed; that is, we shouldn't do any 'send's on it (although 'recv' is
@@ -357,23 +357,23 @@ data ValidLocalEndPointState = ValidLocalEndPointState
357-- particular, it is okay to call removeRemoteEndPoint inside
357-- particular, it is okay to call removeRemoteEndPoint inside
358-- modifyRemoteState.
358-- modifyRemoteState.
360−data RemoteEndPoint = RemoteEndPoint
360+data RemoteEndPoint = RemoteEndPoint
361 { remoteAddress :: !EndPointAddress
361 { remoteAddress :: !EndPointAddress
362 , remoteState :: !(MVar RemoteState)
362 , remoteState :: !(MVar RemoteState)
363 , remoteId :: !HeavyweightConnectionId
363 , remoteId :: !HeavyweightConnectionId
364 , remoteScheduled :: !(Chan (IO ()))
364 , remoteScheduled :: !(Chan (IO ()))
367−data RequestedBy = RequestedByUs | RequestedByThem
367+data RequestedBy = RequestedByUs | RequestedByThem
368 deriving (Eq, Show)
368 deriving (Eq, Show)
370data RemoteState =
370data RemoteState =
371 -- | Invalid remote endpoint (for example, invalid address)
371 -- | Invalid remote endpoint (for example, invalid address)
372 RemoteEndPointInvalid !(TransportError ConnectErrorCode)
372 RemoteEndPointInvalid !(TransportError ConnectErrorCode)
373 -- | The remote endpoint is being initialized
373 -- | The remote endpoint is being initialized
374− | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy
374+ | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy
375 -- | "Normal" working endpoint
375 -- | "Normal" working endpoint
376− | RemoteEndPointValid !ValidRemoteEndPointState
376+ | RemoteEndPointValid !ValidRemoteEndPointState
377 -- | The remote endpoint is being closed (garbage collected)
377 -- | The remote endpoint is being closed (garbage collected)
378 | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState
378 | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState
379 -- | The remote endpoint has been closed (garbage collected)
379 -- | The remote endpoint has been closed (garbage collected)
@@ -382,12 +382,12 @@ data RemoteState =
382 -- using a closeTransport or closeEndPoint API call
382 -- using a closeTransport or closeEndPoint API call
383 | RemoteEndPointFailed !IOException
383 | RemoteEndPointFailed !IOException
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
386-- containers
386-- containers
388-- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if
388-- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if
389−-- we introduce a new event 'AllConnectionsClosed'
389+-- we introduce a new event 'AllConnectionsClosed'
390−data ValidRemoteEndPointState = ValidRemoteEndPointState
390+data ValidRemoteEndPointState = ValidRemoteEndPointState
391 { _remoteOutgoing :: !Int
391 { _remoteOutgoing :: !Int
392 , _remoteIncoming :: !(Set LightweightConnectionId)
392 , _remoteIncoming :: !(Set LightweightConnectionId)
393 , _remoteMaxIncoming :: !LightweightConnectionId
393 , _remoteMaxIncoming :: !LightweightConnectionId
@@ -414,24 +414,24 @@ type LightweightConnectionId = Word32
414-- 'LightweightConnectionId'.
414-- 'LightweightConnectionId'.
415type HeavyweightConnectionId = Word32
415type HeavyweightConnectionId = Word32
417−-- | Control headers
417+-- | Control headers
418−data ControlHeader =
418+data ControlHeader =
419 -- | Tell the remote endpoint that we created a new connection
419 -- | Tell the remote endpoint that we created a new connection
420− CreatedNewConnection
420+ CreatedNewConnection
421 -- | Tell the remote endpoint we will no longer be using a connection
421 -- | Tell the remote endpoint we will no longer be using a connection
422− | CloseConnection
422+ | CloseConnection
423 -- | Request to close the connection (see module description)
423 -- | Request to close the connection (see module description)
424− | CloseSocket
424+ | CloseSocket
425 deriving (Enum, Bounded, Show)
425 deriving (Enum, Bounded, Show)
427-- | Response sent by /B/ to /A/ when /A/ tries to connect
427-- | Response sent by /B/ to /A/ when /A/ tries to connect
428data ConnectionRequestResponse =
428data ConnectionRequestResponse =
429 -- | /B/ accepts the connection
429 -- | /B/ accepts the connection
430− ConnectionRequestAccepted
430+ ConnectionRequestAccepted
431 -- | /A/ requested an invalid endpoint
431 -- | /A/ requested an invalid endpoint
432− | ConnectionRequestInvalid
432+ | ConnectionRequestInvalid
433 -- | /A/s request crossed with a request from /B/ (see protocols)
433 -- | /A/s request crossed with a request from /B/ (see protocols)
434− | ConnectionRequestCrossed
434+ | ConnectionRequestCrossed
435 deriving (Enum, Bounded, Show)
435 deriving (Enum, Bounded, Show)
437-- | Parameters for setting up the TCP transport
437-- | Parameters for setting up the TCP transport
@@ -439,7 +439,7 @@ data TCPParameters = TCPParameters {
439 -- | Backlog for 'listen'.
439 -- | Backlog for 'listen'.
440 -- Defaults to SOMAXCONN.
440 -- Defaults to SOMAXCONN.
441 tcpBacklog :: Int
441 tcpBacklog :: Int
442− -- | Should we set SO_REUSEADDR on the server socket?
442+ -- | Should we set SO_REUSEADDR on the server socket?
443 -- Defaults to True.
443 -- Defaults to True.
444 , tcpReuseServerAddr :: Bool
444 , tcpReuseServerAddr :: Bool
445 -- | Should we set SO_REUSEADDR on client sockets?
445 -- | Should we set SO_REUSEADDR on client sockets?
@@ -448,13 +448,13 @@ data TCPParameters = TCPParameters {
450-- | Internal functionality we expose for unit testing
450-- | Internal functionality we expose for unit testing
451−data TransportInternals = TransportInternals
451+data TransportInternals = TransportInternals
452 { -- | The ID of the thread that listens for new incoming connections
452 { -- | The ID of the thread that listens for new incoming connections
453 transportThread :: ThreadId
453 transportThread :: ThreadId
454 -- | Find the socket between a local and a remote endpoint
454 -- | Find the socket between a local and a remote endpoint
455− , socketBetween :: EndPointAddress
455+ , socketBetween :: EndPointAddress
456− -> EndPointAddress
456+ -> EndPointAddress
457− -> IO N.Socket
457+ -> IO N.Socket
460--------------------------------------------------------------------------------
460--------------------------------------------------------------------------------
@@ -462,62 +462,62 @@ data TransportInternals = TransportInternals
462--------------------------------------------------------------------------------
462--------------------------------------------------------------------------------
464-- | Create a TCP transport
464-- | Create a TCP transport
465−createTransport :: N.HostName
465+createTransport :: N.HostName
466− -> N.ServiceName
466+ -> N.ServiceName
467 -> TCPParameters
467 -> TCPParameters
468 -> IO (Either IOException Transport)
468 -> IO (Either IOException Transport)
469−createTransport host port params =
469+createTransport host port params =
470 either Left (Right . fst) <$> createTransportExposeInternals host port params
470 either Left (Right . fst) <$> createTransportExposeInternals host port params
472-- | You should probably not use this function (used for unit testing only)
472-- | You should probably not use this function (used for unit testing only)
473−createTransportExposeInternals
473+createTransportExposeInternals
474− :: N.HostName
474+ :: N.HostName
475− -> N.ServiceName
475+ -> N.ServiceName
476 -> TCPParameters
476 -> TCPParameters
477− -> IO (Either IOException (Transport, TransportInternals))
477+ -> IO (Either IOException (Transport, TransportInternals))
478−createTransportExposeInternals host port params = do
478+createTransportExposeInternals host port params = do
479− state <- newMVar . TransportValid $ ValidTransportState
479+ state <- newMVar . TransportValid $ ValidTransportState
480− { _localEndPoints = Map.empty
480+ { _localEndPoints = Map.empty
481− , _nextEndPointId = 0
481+ , _nextEndPointId = 0
483 let transport = TCPTransport { transportState = state
483 let transport = TCPTransport { transportState = state
484 , transportHost = host
484 , transportHost = host
485 , transportPort = port
485 , transportPort = port
486 , transportParams = params
486 , transportParams = params
488− tryIO $ bracketOnError (forkServer
488+ tryIO $ bracketOnError (forkServer
491− (tcpBacklog params)
491+ (tcpBacklog params)
492 (tcpReuseServerAddr params)
492 (tcpReuseServerAddr params)
493− (terminationHandler transport)
493+ (terminationHandler transport)
494 (handleConnectionRequest transport))
494 (handleConnectionRequest transport))
495 killThread
495 killThread
496 (mkTransport transport)
496 (mkTransport transport)
498− mkTransport :: TCPTransport
498+ mkTransport :: TCPTransport
499− -> ThreadId
499+ -> ThreadId
500 -> IO (Transport, TransportInternals)
500 -> IO (Transport, TransportInternals)
501 mkTransport transport tid = return
501 mkTransport transport tid = return
502− ( Transport
502+ ( Transport
503− { newEndPoint = apiNewEndPoint transport
503+ { newEndPoint = apiNewEndPoint transport
504 , closeTransport = let evs = [ EndPointClosed
504 , closeTransport = let evs = [ EndPointClosed
505 , throw $ userError "Transport closed"
505 , throw $ userError "Transport closed"
507− apiCloseTransport transport (Just tid) evs
507+ apiCloseTransport transport (Just tid) evs
509− , TransportInternals
509+ , TransportInternals
510 { transportThread = tid
510 { transportThread = tid
511 , socketBetween = internalSocketBetween transport
511 , socketBetween = internalSocketBetween transport
515 terminationHandler :: TCPTransport -> SomeException -> IO ()
515 terminationHandler :: TCPTransport -> SomeException -> IO ()
516− terminationHandler transport ex = do
516+ terminationHandler transport ex = do
517 let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))
517 let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))
518− , throw $ userError "Transport closed"
518+ , throw $ userError "Transport closed"
520− apiCloseTransport transport Nothing evs
520+ apiCloseTransport transport Nothing evs
522-- | Default TCP parameters
522-- | Default TCP parameters
523defaultTCPParameters :: TCPParameters
523defaultTCPParameters :: TCPParameters
@@ -533,39 +533,39 @@ defaultTCPParameters = TCPParameters {
534-- | Close the transport
534-- | Close the transport
535apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()
535apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()
536−apiCloseTransport transport mTransportThread evs =
536+apiCloseTransport transport mTransportThread evs =
537 asyncWhenCancelled return $ do
537 asyncWhenCancelled return $ do
538 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
538 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
539 TransportValid vst -> return (TransportClosed, Just vst)
539 TransportValid vst -> return (TransportClosed, Just vst)
540 TransportClosed -> return (TransportClosed, Nothing)
540 TransportClosed -> return (TransportClosed, Nothing)
541− forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
541+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
542 -- This will invoke the termination handler, which in turn will call
542 -- This will invoke the termination handler, which in turn will call
543 -- apiCloseTransport again, but then the transport will already be closed
543 -- apiCloseTransport again, but then the transport will already be closed
544 -- and we won't be passed a transport thread, so we terminate immmediate
544 -- and we won't be passed a transport thread, so we terminate immmediate
545− forM_ mTransportThread killThread
545+ forM_ mTransportThread killThread
547−-- | Create a new endpoint
547+-- | Create a new endpoint
548−apiNewEndPoint :: TCPTransport
548+apiNewEndPoint :: TCPTransport
549 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
549 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
550−apiNewEndPoint transport =
550+apiNewEndPoint transport =
551 try . asyncWhenCancelled closeEndPoint $ do
551 try . asyncWhenCancelled closeEndPoint $ do
552 ourEndPoint <- createLocalEndPoint transport
552 ourEndPoint <- createLocalEndPoint transport
553− return EndPoint
553+ return EndPoint
554 { receive = readChan (localChannel ourEndPoint)
554 { receive = readChan (localChannel ourEndPoint)
555 , address = localAddress ourEndPoint
555 , address = localAddress ourEndPoint
556− , connect = apiConnect (transportParams transport) ourEndPoint
556+ , connect = apiConnect (transportParams transport) ourEndPoint
557 , closeEndPoint = let evs = [ EndPointClosed
557 , closeEndPoint = let evs = [ EndPointClosed
558− , throw $ userError "Endpoint closed"
558+ , throw $ userError "Endpoint closed"
560 apiCloseEndPoint transport evs ourEndPoint
560 apiCloseEndPoint transport evs ourEndPoint
561− , newMulticastGroup = return . Left $ newMulticastGroupError
561+ , newMulticastGroup = return . Left $ newMulticastGroupError
562 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
562 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
565− newMulticastGroupError =
565+ newMulticastGroupError =
566− TransportError NewMulticastGroupUnsupported "Multicast not supported"
566+ TransportError NewMulticastGroupUnsupported "Multicast not supported"
567− resolveMulticastGroupError =
567+ resolveMulticastGroupError =
568− TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
568+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
570-- | Connnect to an endpoint
570-- | Connnect to an endpoint
571apiConnect :: TCPParameters -- ^ Parameters
571apiConnect :: TCPParameters -- ^ Parameters
@@ -575,35 +575,35 @@ apiConnect :: TCPParameters -- ^ Parameters
575 -> ConnectHints -- ^ Hints
575 -> ConnectHints -- ^ Hints
576 -> IO (Either (TransportError ConnectErrorCode) Connection)
576 -> IO (Either (TransportError ConnectErrorCode) Connection)
577apiConnect params ourEndPoint theirAddress _reliability hints =
577apiConnect params ourEndPoint theirAddress _reliability hints =
578− try . asyncWhenCancelled close $
578+ try . asyncWhenCancelled close $
579− if localAddress ourEndPoint == theirAddress
579+ if localAddress ourEndPoint == theirAddress
580− then connectToSelf ourEndPoint
580+ then connectToSelf ourEndPoint
582 resetIfBroken ourEndPoint theirAddress
582 resetIfBroken ourEndPoint theirAddress
583− (theirEndPoint, connId) <-
583+ (theirEndPoint, connId) <-
584 createConnectionTo params ourEndPoint theirAddress hints
584 createConnectionTo params ourEndPoint theirAddress hints
585 -- connAlive can be an IORef rather than an MVar because it is protected
585 -- connAlive can be an IORef rather than an MVar because it is protected
586 -- by the remoteState MVar. We don't need the overhead of locking twice.
586 -- by the remoteState MVar. We don't need the overhead of locking twice.
587 connAlive <- newIORef True
587 connAlive <- newIORef True
588− return Connection
588+ return Connection
589− { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
589+ { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
590− , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
590+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
593-- | Close a connection
593-- | Close a connection
594apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()
594apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()
595−apiClose (ourEndPoint, theirEndPoint) connId connAlive =
595+apiClose (ourEndPoint, theirEndPoint) connId connAlive =
596− void . tryIO . asyncWhenCancelled return $ do
596+ void . tryIO . asyncWhenCancelled return $ do
597 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
597 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
598 RemoteEndPointValid vst -> do
598 RemoteEndPointValid vst -> do
599 alive <- readIORef connAlive
599 alive <- readIORef connAlive
600− if alive
600+ if alive
602 writeIORef connAlive False
602 writeIORef connAlive False
603− act <- schedule theirEndPoint $
603+ act <- schedule theirEndPoint $
604− sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
604+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
605− return ( RemoteEndPointValid
605+ return ( RemoteEndPointValid
606− . (remoteOutgoing ^: (\x -> x - 1))
606+ . (remoteOutgoing ^: (\x -> x - 1))
608 , Just act
608 , Just act
@@ -616,12 +616,12 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive =
618-- | Send data across a connection
618-- | Send data across a connection
619−apiSend :: EndPointPair -- ^ Local and remote endpoint
619+apiSend :: EndPointPair -- ^ Local and remote endpoint
620 -> LightweightConnectionId -- ^ Connection ID
620 -> LightweightConnectionId -- ^ Connection ID
621 -> IORef Bool -- ^ Is the connection still alive?
621 -> IORef Bool -- ^ Is the connection still alive?
622 -> [ByteString] -- ^ Payload
622 -> [ByteString] -- ^ Payload
623 -> IO (Either (TransportError SendErrorCode) ())
623 -> IO (Either (TransportError SendErrorCode) ())
624−apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
624+apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
625 -- We don't need the overhead of asyncWhenCancelled here
625 -- We don't need the overhead of asyncWhenCancelled here
626 try . mapIOException sendFailed $ do
626 try . mapIOException sendFailed $ do
627 act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
627 act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
@@ -631,70 +631,70 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
631 relyViolation (ourEndPoint, theirEndPoint) "apiSend"
631 relyViolation (ourEndPoint, theirEndPoint) "apiSend"
632 RemoteEndPointValid vst -> do
632 RemoteEndPointValid vst -> do
633 alive <- readIORef connAlive
633 alive <- readIORef connAlive
634− if alive
634+ if alive
635− then schedule theirEndPoint $
635+ then schedule theirEndPoint $
636 sendOn vst (encodeInt32 connId : prependLength payload)
636 sendOn vst (encodeInt32 connId : prependLength payload)
637 else throwIO $ TransportError SendClosed "Connection closed"
637 else throwIO $ TransportError SendClosed "Connection closed"
638− RemoteEndPointClosing _ _ -> do
638+ RemoteEndPointClosing _ _ -> do
639 alive <- readIORef connAlive
639 alive <- readIORef connAlive
640− if alive
640+ if alive
641− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
641+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
642 else throwIO $ TransportError SendClosed "Connection closed"
642 else throwIO $ TransportError SendClosed "Connection closed"
643 RemoteEndPointClosed -> do
643 RemoteEndPointClosed -> do
644 alive <- readIORef connAlive
644 alive <- readIORef connAlive
645− if alive
645+ if alive
646− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
646+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
647 else throwIO $ TransportError SendClosed "Connection closed"
647 else throwIO $ TransportError SendClosed "Connection closed"
648 RemoteEndPointFailed err -> do
648 RemoteEndPointFailed err -> do
649 alive <- readIORef connAlive
649 alive <- readIORef connAlive
650− if alive
650+ if alive
651− then throwIO $ TransportError SendFailed (show err)
651+ then throwIO $ TransportError SendFailed (show err)
652 else throwIO $ TransportError SendClosed "Connection closed"
652 else throwIO $ TransportError SendClosed "Connection closed"
653 runScheduledAction (ourEndPoint, theirEndPoint) act
653 runScheduledAction (ourEndPoint, theirEndPoint) act
655 sendFailed = TransportError SendFailed . show
655 sendFailed = TransportError SendFailed . show
657-- | Force-close the endpoint
657-- | Force-close the endpoint
658−apiCloseEndPoint :: TCPTransport -- ^ Transport
658+apiCloseEndPoint :: TCPTransport -- ^ Transport
659− -> [Event] -- ^ Events used to report closure
659+ -> [Event] -- ^ Events used to report closure
660 -> LocalEndPoint -- ^ Local endpoint
660 -> LocalEndPoint -- ^ Local endpoint
662apiCloseEndPoint transport evs ourEndPoint =
662apiCloseEndPoint transport evs ourEndPoint =
663 asyncWhenCancelled return $ do
663 asyncWhenCancelled return $ do
664 -- Remove the reference from the transport state
664 -- Remove the reference from the transport state
665 removeLocalEndPoint transport ourEndPoint
665 removeLocalEndPoint transport ourEndPoint
666− -- Close the local endpoint
666+ -- Close the local endpoint
667 mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
667 mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
668 case st of
668 case st of
669− LocalEndPointValid vst ->
669+ LocalEndPointValid vst ->
670 return (LocalEndPointClosed, Just vst)
670 return (LocalEndPointClosed, Just vst)
671 LocalEndPointClosed ->
671 LocalEndPointClosed ->
672 return (LocalEndPointClosed, Nothing)
672 return (LocalEndPointClosed, Nothing)
673 forM_ mOurState $ \vst -> do
673 forM_ mOurState $ \vst -> do
674 forM_ (vst ^. localConnections) tryCloseRemoteSocket
674 forM_ (vst ^. localConnections) tryCloseRemoteSocket
675− forM_ evs $ writeChan (localChannel ourEndPoint)
675+ forM_ evs $ writeChan (localChannel ourEndPoint)
677 -- Close the remote socket and return the set of all incoming connections
677 -- Close the remote socket and return the set of all incoming connections
678− tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
678+ tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
679 tryCloseRemoteSocket theirEndPoint = do
679 tryCloseRemoteSocket theirEndPoint = do
680− -- We make an attempt to close the connection nicely
680+ -- We make an attempt to close the connection nicely
681 -- (by sending a CloseSocket first)
681 -- (by sending a CloseSocket first)
682 let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"
682 let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"
683 mAct <- modifyMVar (remoteState theirEndPoint) $ \st ->
683 mAct <- modifyMVar (remoteState theirEndPoint) $ \st ->
684 case st of
684 case st of
685− RemoteEndPointInvalid _ ->
685+ RemoteEndPointInvalid _ ->
686 return (st, Nothing)
686 return (st, Nothing)
687 RemoteEndPointInit resolved _ _ -> do
687 RemoteEndPointInit resolved _ _ -> do
688 putMVar resolved ()
688 putMVar resolved ()
689 return (closed, Nothing)
689 return (closed, Nothing)
690− RemoteEndPointValid vst -> do
690+ RemoteEndPointValid vst -> do
691 act <- schedule theirEndPoint $ do
691 act <- schedule theirEndPoint $ do
692 tryIO $ sendOn vst [ encodeInt32 CloseSocket
692 tryIO $ sendOn vst [ encodeInt32 CloseSocket
693 , encodeInt32 (vst ^. remoteMaxIncoming)
693 , encodeInt32 (vst ^. remoteMaxIncoming)
695 tryCloseSocket (remoteSocket vst)
695 tryCloseSocket (remoteSocket vst)
696 return (closed, Just act)
696 return (closed, Just act)
697− RemoteEndPointClosing resolved vst -> do
697+ RemoteEndPointClosing resolved vst -> do
698 putMVar resolved ()
698 putMVar resolved ()
699 act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)
699 act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)
700 return (closed, Just act)
700 return (closed, Just act)
@@ -703,7 +703,7 @@ apiCloseEndPoint transport evs ourEndPoint =
703 RemoteEndPointFailed err ->
703 RemoteEndPointFailed err ->
704 return (RemoteEndPointFailed err, Nothing)
704 return (RemoteEndPointFailed err, Nothing)
705 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
705 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
708--------------------------------------------------------------------------------
708--------------------------------------------------------------------------------
709-- Incoming requests --
709-- Incoming requests --
@@ -720,11 +720,11 @@ apiCloseEndPoint transport evs ourEndPoint =
720-- the transport down. We must be careful to close the socket when a (possibly
720-- the transport down. We must be careful to close the socket when a (possibly
721-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
721-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
722-- handleConnectionRequest the transport will be shut down.)
722-- handleConnectionRequest the transport will be shut down.)
723−handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
723+handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
724−handleConnectionRequest transport sock = handle handleException $ do
724+handleConnectionRequest transport sock = handle handleException $ do
725 ourEndPointId <- recvInt32 sock
725 ourEndPointId <- recvInt32 sock
726− theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
726+ theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
727− let ourAddress = encodeEndPointAddress (transportHost transport)
727+ let ourAddress = encodeEndPointAddress (transportHost transport)
728 (transportPort transport)
728 (transportPort transport)
729 ourEndPointId
729 ourEndPointId
730 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
730 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
@@ -735,32 +735,32 @@ handleConnectionRequest transport sock = handle handleException $ do
735 throwIO $ userError "handleConnectionRequest: Invalid endpoint"
735 throwIO $ userError "handleConnectionRequest: Invalid endpoint"
736 Just ourEndPoint ->
736 Just ourEndPoint ->
737 return ourEndPoint
737 return ourEndPoint
738− TransportClosed ->
738+ TransportClosed ->
739 throwIO $ userError "Transport closed"
739 throwIO $ userError "Transport closed"
740 void . forkIO $ go ourEndPoint theirAddress
740 void . forkIO $ go ourEndPoint theirAddress
742 go :: LocalEndPoint -> EndPointAddress -> IO ()
742 go :: LocalEndPoint -> EndPointAddress -> IO ()
743− go ourEndPoint theirAddress = do
743+ go ourEndPoint theirAddress = do
744 -- This runs in a thread that will never be killed
744 -- This runs in a thread that will never be killed
745− mEndPoint <- handle ((>> return Nothing) . handleException) $ do
745+ mEndPoint <- handle ((>> return Nothing) . handleException) $ do
746 resetIfBroken ourEndPoint theirAddress
746 resetIfBroken ourEndPoint theirAddress
747− (theirEndPoint, isNew) <-
747+ (theirEndPoint, isNew) <-
748 findRemoteEndPoint ourEndPoint theirAddress RequestedByThem
748 findRemoteEndPoint ourEndPoint theirAddress RequestedByThem
750− if not isNew
750+ if not isNew
752 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
752 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
753 tryCloseSocket sock
753 tryCloseSocket sock
754− return Nothing
754+ return Nothing
756 sendLock <- newMVar ()
756 sendLock <- newMVar ()
757− let vst = ValidRemoteEndPointState
757+ let vst = ValidRemoteEndPointState
758 { remoteSocket = sock
758 { remoteSocket = sock
759 , remoteSendLock = sendLock
759 , remoteSendLock = sendLock
760 , _remoteOutgoing = 0
760 , _remoteOutgoing = 0
761 , _remoteIncoming = Set.empty
761 , _remoteIncoming = Set.empty
762 , _remoteMaxIncoming = 0
762 , _remoteMaxIncoming = 0
763− , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
763+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
765 sendMany sock [encodeInt32 ConnectionRequestAccepted]
765 sendMany sock [encodeInt32 ConnectionRequestAccepted]
766 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
766 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
@@ -769,30 +769,30 @@ handleConnectionRequest transport sock = handle handleException $ do
769 -- Nothing then the socket is already closed; otherwise, the socket has
769 -- Nothing then the socket is already closed; otherwise, the socket has
770 -- been recorded as part of the remote endpoint. Either way, we no longer
770 -- been recorded as part of the remote endpoint. Either way, we no longer
771 -- have to worry about closing the socket on receiving an asynchronous
771 -- have to worry about closing the socket on receiving an asynchronous
772− -- exception from this point forward.
772+ -- exception from this point forward.
773 forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
773 forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
775 handleException :: SomeException -> IO ()
775 handleException :: SomeException -> IO ()
776− handleException ex = do
776+ handleException ex = do
777− tryCloseSocket sock
777+ tryCloseSocket sock
778 rethrowIfAsync (fromException ex)
778 rethrowIfAsync (fromException ex)
780− rethrowIfAsync :: Maybe AsyncException -> IO ()
780+ rethrowIfAsync :: Maybe AsyncException -> IO ()
781− rethrowIfAsync = mapM_ throwIO
781+ rethrowIfAsync = mapM_ throwIO
783-- | Handle requests from a remote endpoint.
783-- | Handle requests from a remote endpoint.
785-- Returns only if the remote party closes the socket or if an error occurs.
785-- Returns only if the remote party closes the socket or if an error occurs.
786-- This runs in a thread that will never be killed.
786-- This runs in a thread that will never be killed.
787−handleIncomingMessages :: EndPointPair -> IO ()
787+handleIncomingMessages :: EndPointPair -> IO ()
788handleIncomingMessages (ourEndPoint, theirEndPoint) = do
788handleIncomingMessages (ourEndPoint, theirEndPoint) = do
789 mSock <- withMVar theirState $ \st ->
789 mSock <- withMVar theirState $ \st ->
790 case st of
790 case st of
791 RemoteEndPointInvalid _ ->
791 RemoteEndPointInvalid _ ->
792− relyViolation (ourEndPoint, theirEndPoint)
792+ relyViolation (ourEndPoint, theirEndPoint)
793− "handleIncomingMessages (invalid)"
793+ "handleIncomingMessages (invalid)"
794 RemoteEndPointInit _ _ _ ->
794 RemoteEndPointInit _ _ _ ->
795− relyViolation (ourEndPoint, theirEndPoint)
795+ relyViolation (ourEndPoint, theirEndPoint)
796 "handleIncomingMessages (init)"
796 "handleIncomingMessages (init)"
797 RemoteEndPointValid ep ->
797 RemoteEndPointValid ep ->
798 return . Just $ remoteSocket ep
798 return . Just $ remoteSocket ep
@@ -800,13 +800,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
800 return . Just $ remoteSocket ep
800 return . Just $ remoteSocket ep
801 RemoteEndPointClosed ->
801 RemoteEndPointClosed ->
802 return Nothing
802 return Nothing
803− RemoteEndPointFailed _ ->
803+ RemoteEndPointFailed _ ->
804 return Nothing
804 return Nothing
806− forM_ mSock $ \sock ->
806+ forM_ mSock $ \sock ->
807 tryIO (go sock) >>= either (prematureExit sock) return
807 tryIO (go sock) >>= either (prematureExit sock) return
809− -- Dispatch
809+ -- Dispatch
811 -- If a recv throws an exception this will be caught top-level and
811 -- If a recv throws an exception this will be caught top-level and
812 -- 'prematureExit' will be invoked. The same will happen if the remote
812 -- 'prematureExit' will be invoked. The same will happen if the remote
@@ -817,38 +817,38 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
817 go :: N.Socket -> IO ()
817 go :: N.Socket -> IO ()
818 go sock = do
818 go sock = do
819 lcid <- recvInt32 sock :: IO LightweightConnectionId
819 lcid <- recvInt32 sock :: IO LightweightConnectionId
820− if lcid >= firstNonReservedLightweightConnectionId
820+ if lcid >= firstNonReservedLightweightConnectionId
822− readMessage sock lcid
822+ readMessage sock lcid
825 case tryToEnum (fromIntegral lcid) of
825 case tryToEnum (fromIntegral lcid) of
826 Just CreatedNewConnection -> do
826 Just CreatedNewConnection -> do
827− recvInt32 sock >>= createdNewConnection
827+ recvInt32 sock >>= createdNewConnection
828− go sock
828+ go sock
829 Just CloseConnection -> do
829 Just CloseConnection -> do
830− recvInt32 sock >>= closeConnection
830+ recvInt32 sock >>= closeConnection
832− Just CloseSocket -> do
832+ Just CloseSocket -> do
833− didClose <- recvInt32 sock >>= closeSocket sock
833+ didClose <- recvInt32 sock >>= closeSocket sock
834 unless didClose $ go sock
834 unless didClose $ go sock
835 Nothing ->
835 Nothing ->
836 throwIO $ userError "Invalid control request"
836 throwIO $ userError "Invalid control request"
838 -- Create a new connection
838 -- Create a new connection
839− createdNewConnection :: LightweightConnectionId -> IO ()
839+ createdNewConnection :: LightweightConnectionId -> IO ()
840 createdNewConnection lcid = do
840 createdNewConnection lcid = do
841 modifyMVar_ theirState $ \st -> do
841 modifyMVar_ theirState $ \st -> do
842 vst <- case st of
842 vst <- case st of
843 RemoteEndPointInvalid _ ->
843 RemoteEndPointInvalid _ ->
844− relyViolation (ourEndPoint, theirEndPoint)
844+ relyViolation (ourEndPoint, theirEndPoint)
845 "handleIncomingMessages:createNewConnection (invalid)"
845 "handleIncomingMessages:createNewConnection (invalid)"
846 RemoteEndPointInit _ _ _ ->
846 RemoteEndPointInit _ _ _ ->
847− relyViolation (ourEndPoint, theirEndPoint)
847+ relyViolation (ourEndPoint, theirEndPoint)
848 "handleIncomingMessages:createNewConnection (init)"
848 "handleIncomingMessages:createNewConnection (init)"
849 RemoteEndPointValid vst ->
849 RemoteEndPointValid vst ->
850 return ( (remoteIncoming ^: Set.insert lcid)
850 return ( (remoteIncoming ^: Set.insert lcid)
851− $ (remoteMaxIncoming ^= lcid)
851+ $ (remoteMaxIncoming ^= lcid)
854 RemoteEndPointClosing resolved vst -> do
854 RemoteEndPointClosing resolved vst -> do
@@ -863,36 +863,36 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
863 . (remoteMaxIncoming ^= lcid)
863 . (remoteMaxIncoming ^= lcid)
866− RemoteEndPointFailed err ->
866+ RemoteEndPointFailed err ->
867 throwIO err
867 throwIO err
868 RemoteEndPointClosed ->
868 RemoteEndPointClosed ->
869− relyViolation (ourEndPoint, theirEndPoint)
869+ relyViolation (ourEndPoint, theirEndPoint)
870 "createNewConnection (closed)"
870 "createNewConnection (closed)"
871 return (RemoteEndPointValid vst)
871 return (RemoteEndPointValid vst)
872− writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)
872+ writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)
874− -- Close a connection
874+ -- Close a connection
875 -- It is important that we verify that the connection is in fact open,
875 -- It is important that we verify that the connection is in fact open,
876 -- because otherwise we should not decrement the reference count
876 -- because otherwise we should not decrement the reference count
877− closeConnection :: LightweightConnectionId -> IO ()
877+ closeConnection :: LightweightConnectionId -> IO ()
878 closeConnection lcid = do
878 closeConnection lcid = do
879 modifyMVar_ theirState $ \st -> case st of
879 modifyMVar_ theirState $ \st -> case st of
880 RemoteEndPointInvalid _ ->
880 RemoteEndPointInvalid _ ->
881 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"
881 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"
882 RemoteEndPointInit _ _ _ ->
882 RemoteEndPointInit _ _ _ ->
883 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"
883 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"
884− RemoteEndPointValid vst -> do
884+ RemoteEndPointValid vst -> do
885− unless (Set.member lcid (vst ^. remoteIncoming)) $
885+ unless (Set.member lcid (vst ^. remoteIncoming)) $
886 throwIO $ userError "Invalid CloseConnection"
886 throwIO $ userError "Invalid CloseConnection"
887− return ( RemoteEndPointValid
887+ return ( RemoteEndPointValid
888− . (remoteIncoming ^: Set.delete lcid)
888+ . (remoteIncoming ^: Set.delete lcid)
891 RemoteEndPointClosing _ _ ->
891 RemoteEndPointClosing _ _ ->
892 -- If the remote endpoint is in Closing state, that means that are as
892 -- If the remote endpoint is in Closing state, that means that are as
893 -- far as we are concerned there are no incoming connections. This
893 -- far as we are concerned there are no incoming connections. This
894 -- means that a CloseConnection request at this point is invalid.
894 -- means that a CloseConnection request at this point is invalid.
895− throwIO $ userError "Invalid CloseConnection request"
895+ throwIO $ userError "Invalid CloseConnection request"
896 RemoteEndPointFailed err ->
896 RemoteEndPointFailed err ->
897 throwIO err
897 throwIO err
898 RemoteEndPointClosed ->
898 RemoteEndPointClosed ->
@@ -900,7 +900,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
900 writeChan ourChannel (ConnectionClosed (connId lcid))
900 writeChan ourChannel (ConnectionClosed (connId lcid))
902 -- Close the socket (if we don't have any outgoing connections)
902 -- 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
904 closeSocket sock lastReceivedId = do
904 closeSocket sock lastReceivedId = do
905 mAct <- modifyMVar theirState $ \st -> do
905 mAct <- modifyMVar theirState $ \st -> do
906 case st of
906 case st of
@@ -908,38 +908,38 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
908 relyViolation (ourEndPoint, theirEndPoint)
908 relyViolation (ourEndPoint, theirEndPoint)
909 "handleIncomingMessages:closeSocket (invalid)"
909 "handleIncomingMessages:closeSocket (invalid)"
910 RemoteEndPointInit _ _ _ ->
910 RemoteEndPointInit _ _ _ ->
911− relyViolation (ourEndPoint, theirEndPoint)
911+ relyViolation (ourEndPoint, theirEndPoint)
912 "handleIncomingMessages:closeSocket (init)"
912 "handleIncomingMessages:closeSocket (init)"
913 RemoteEndPointValid vst -> do
913 RemoteEndPointValid vst -> do
914 -- We regard a CloseSocket message as an (optimized) way for the
914 -- We regard a CloseSocket message as an (optimized) way for the
915 -- remote endpoint to indicate that all its connections to us are
915 -- remote endpoint to indicate that all its connections to us are
916 -- now properly closed
916 -- now properly closed
917− forM_ (Set.elems $ vst ^. remoteIncoming) $
917+ forM_ (Set.elems $ vst ^. remoteIncoming) $
918 writeChan ourChannel . ConnectionClosed . connId
918 writeChan ourChannel . ConnectionClosed . connId
919− let vst' = remoteIncoming ^= Set.empty $ vst
919+ let vst' = remoteIncoming ^= Set.empty $ vst
920 -- If we still have outgoing connections then we ignore the
920 -- If we still have outgoing connections then we ignore the
921 -- CloseSocket request (we sent a ConnectionCreated message to the
921 -- CloseSocket request (we sent a ConnectionCreated message to the
922 -- remote endpoint, but it did not receive it before sending the
922 -- remote endpoint, but it did not receive it before sending the
923 -- CloseSocket request). Similarly, if lastReceivedId < lastSentId
923 -- CloseSocket request). Similarly, if lastReceivedId < lastSentId
924 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
924 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
925 -- message to the remote endpoint, *both of which* it did not yet
925 -- message to the remote endpoint, *both of which* it did not yet
926− -- receive before sending the CloseSocket request.
926+ -- receive before sending the CloseSocket request.
927− if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
927+ if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
929 return (RemoteEndPointValid vst', Nothing)
929 return (RemoteEndPointValid vst', Nothing)
930− else do
930+ else do
931 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
931 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
932 -- Attempt to reply (but don't insist)
932 -- Attempt to reply (but don't insist)
933 act <- schedule theirEndPoint $ do
933 act <- schedule theirEndPoint $ do
934 tryIO $ sendOn vst' [ encodeInt32 CloseSocket
934 tryIO $ sendOn vst' [ encodeInt32 CloseSocket
935 , encodeInt32 (vst ^. remoteMaxIncoming)
935 , encodeInt32 (vst ^. remoteMaxIncoming)
937− tryCloseSocket sock
937+ tryCloseSocket sock
938 return (RemoteEndPointClosed, Just act)
938 return (RemoteEndPointClosed, Just act)
939 RemoteEndPointClosing resolved vst -> do
939 RemoteEndPointClosing resolved vst -> do
940 -- Like above, we need to check if there is a ConnectionCreated
940 -- Like above, we need to check if there is a ConnectionCreated
941− -- message that we sent but that the remote endpoint has not yet
941+ -- message that we sent but that the remote endpoint has not yet
942− -- received. However, since we are in 'closing' state, the only
942+ -- received. However, since we are in 'closing' state, the only
943 -- way this may happen is when we sent a ConnectionCreated,
943 -- way this may happen is when we sent a ConnectionCreated,
944 -- ConnectionClosed, and CloseSocket message, none of which have
944 -- ConnectionClosed, and CloseSocket message, none of which have
945 -- yet been received. We leave the endpoint in closing state in
945 -- yet been received. We leave the endpoint in closing state in
@@ -949,29 +949,29 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
949 return (RemoteEndPointClosing resolved vst, Nothing)
949 return (RemoteEndPointClosing resolved vst, Nothing)
951 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
951 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
952− act <- schedule theirEndPoint $ tryCloseSocket sock
952+ act <- schedule theirEndPoint $ tryCloseSocket sock
953 putMVar resolved ()
953 putMVar resolved ()
954 return (RemoteEndPointClosed, Just act)
954 return (RemoteEndPointClosed, Just act)
955 RemoteEndPointFailed err ->
955 RemoteEndPointFailed err ->
956 throwIO err
956 throwIO err
957 RemoteEndPointClosed ->
957 RemoteEndPointClosed ->
958− relyViolation (ourEndPoint, theirEndPoint)
958+ relyViolation (ourEndPoint, theirEndPoint)
959 "handleIncomingMessages:closeSocket (closed)"
959 "handleIncomingMessages:closeSocket (closed)"
960 case mAct of
960 case mAct of
961 Nothing -> return False
961 Nothing -> return False
962 Just act -> do
962 Just act -> do
963 runScheduledAction (ourEndPoint, theirEndPoint) act
963 runScheduledAction (ourEndPoint, theirEndPoint) act
964 return True
964 return True
966 -- Read a message and output it on the endPoint's channel. By rights we
966 -- Read a message and output it on the endPoint's channel. By rights we
967 -- should verify that the connection ID is valid, but this is unnecessary
967 -- should verify that the connection ID is valid, but this is unnecessary
968 -- overhead
968 -- overhead
969− readMessage :: N.Socket -> LightweightConnectionId -> IO ()
969+ readMessage :: N.Socket -> LightweightConnectionId -> IO ()
970− readMessage sock lcid =
970+ readMessage sock lcid =
971− recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)
971+ recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)
973 -- Arguments
973 -- Arguments
974− ourChannel = localChannel ourEndPoint
974+ ourChannel = localChannel ourEndPoint
975 theirState = remoteState theirEndPoint
975 theirState = remoteState theirEndPoint
976 theirAddr = remoteAddress theirEndPoint
976 theirAddr = remoteAddress theirEndPoint
@@ -982,20 +982,20 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
982 modifyMVar_ theirState $ \st ->
982 modifyMVar_ theirState $ \st ->
983 case st of
983 case st of
984 RemoteEndPointInvalid _ ->
984 RemoteEndPointInvalid _ ->
985− relyViolation (ourEndPoint, theirEndPoint)
985+ relyViolation (ourEndPoint, theirEndPoint)
986 "handleIncomingMessages:prematureExit"
986 "handleIncomingMessages:prematureExit"
987 RemoteEndPointInit _ _ _ ->
987 RemoteEndPointInit _ _ _ ->
988− relyViolation (ourEndPoint, theirEndPoint)
988+ relyViolation (ourEndPoint, theirEndPoint)
989 "handleIncomingMessages:prematureExit"
989 "handleIncomingMessages:prematureExit"
990 RemoteEndPointValid _ -> do
990 RemoteEndPointValid _ -> do
991− let code = EventConnectionLost (remoteAddress theirEndPoint)
991+ let code = EventConnectionLost (remoteAddress theirEndPoint)
992 writeChan ourChannel . ErrorEvent $ TransportError code (show err)
992 writeChan ourChannel . ErrorEvent $ TransportError code (show err)
993 return (RemoteEndPointFailed err)
993 return (RemoteEndPointFailed err)
994 RemoteEndPointClosing resolved _ -> do
994 RemoteEndPointClosing resolved _ -> do
995 putMVar resolved ()
995 putMVar resolved ()
996 return (RemoteEndPointFailed err)
996 return (RemoteEndPointFailed err)
997 RemoteEndPointClosed ->
997 RemoteEndPointClosed ->
998− relyViolation (ourEndPoint, theirEndPoint)
998+ relyViolation (ourEndPoint, theirEndPoint)
999 "handleIncomingMessages:prematureExit"
999 "handleIncomingMessages:prematureExit"
1000 RemoteEndPointFailed err' ->
1000 RemoteEndPointFailed err' ->
1001 return (RemoteEndPointFailed err')
1001 return (RemoteEndPointFailed err')
@@ -1006,7 +1006,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1007 -- The ID of the last connection _we_ created (or 0 for none)
1007 -- The ID of the last connection _we_ created (or 0 for none)
1008 lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId
1008 lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId
1009− lastSentId vst =
1009+ lastSentId vst =
1010 if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId
1010 if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId
1012 else (vst ^. remoteNextConnOutId) - 1
1012 else (vst ^. remoteNextConnOutId) - 1
@@ -1021,12 +1021,12 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1021-- | Create a connection to a remote endpoint
1021-- | Create a connection to a remote endpoint
1023-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will
1023-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will
1024−-- block until that is resolved.
1024+-- block until that is resolved.
1026-- May throw a TransportError ConnectErrorCode exception.
1026-- May throw a TransportError ConnectErrorCode exception.
1027−createConnectionTo :: TCPParameters
1027+createConnectionTo :: TCPParameters
1028− -> LocalEndPoint
1028+ -> LocalEndPoint
1029− -> EndPointAddress
1029+ -> EndPointAddress
1030 -> ConnectHints
1030 -> ConnectHints
1031 -> IO (RemoteEndPoint, LightweightConnectionId)
1031 -> IO (RemoteEndPoint, LightweightConnectionId)
1032createConnectionTo params ourEndPoint theirAddress hints = go
1032createConnectionTo params ourEndPoint theirAddress hints = go
@@ -1035,27 +1035,27 @@ createConnectionTo params ourEndPoint theirAddress hints = go
1035 (theirEndPoint, isNew) <- mapIOException connectFailed $
1035 (theirEndPoint, isNew) <- mapIOException connectFailed $
1036 findRemoteEndPoint ourEndPoint theirAddress RequestedByUs
1036 findRemoteEndPoint ourEndPoint theirAddress RequestedByUs
1038− if isNew
1038+ if isNew
1040− forkIO . handle absorbAllExceptions $
1040+ forkIO . handle absorbAllExceptions $
1041− setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
1041+ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
1044 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
1044 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
1045 mapIOException connectFailed $ do
1045 mapIOException connectFailed $ do
1046 act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1046 act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1047− RemoteEndPointValid vst -> do
1047+ RemoteEndPointValid vst -> do
1048− let connId = vst ^. remoteNextConnOutId
1048+ let connId = vst ^. remoteNextConnOutId
1049 act <- schedule theirEndPoint $ do
1049 act <- schedule theirEndPoint $ do
1050 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
1050 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
1051 return connId
1051 return connId
1052− return ( RemoteEndPointValid
1052+ return ( RemoteEndPointValid
1053− $ remoteNextConnOutId ^= connId + 1
1053+ $ remoteNextConnOutId ^= connId + 1
1057 -- Error cases
1057 -- Error cases
1058− RemoteEndPointInvalid err ->
1058+ RemoteEndPointInvalid err ->
1059 throwIO err
1059 throwIO err
1060 RemoteEndPointFailed err ->
1060 RemoteEndPointFailed err ->
1061 throwIO err
1061 throwIO err
@@ -1065,32 +1065,32 @@ createConnectionTo params ourEndPoint theirAddress hints = go
1065 -- TODO: deal with exception case?
1065 -- TODO: deal with exception case?
1066 connId <- runScheduledAction (ourEndPoint, theirEndPoint) act
1066 connId <- runScheduledAction (ourEndPoint, theirEndPoint) act
1067 return (theirEndPoint, connId)
1067 return (theirEndPoint, connId)
1070 connectFailed :: IOException -> TransportError ConnectErrorCode
1070 connectFailed :: IOException -> TransportError ConnectErrorCode
1071 connectFailed = TransportError ConnectFailed . show
1071 connectFailed = TransportError ConnectFailed . show
1073 absorbAllExceptions :: SomeException -> IO ()
1073 absorbAllExceptions :: SomeException -> IO ()
1074− absorbAllExceptions _ex =
1074+ absorbAllExceptions _ex =
1075 return ()
1075 return ()
1077-- | Set up a remote endpoint
1077-- | Set up a remote endpoint
1078−setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
1078+setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
1079setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1079setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1080− result <- socketToEndPoint ourAddress
1080+ result <- socketToEndPoint ourAddress
1081− theirAddress
1081+ theirAddress
1082 (tcpReuseClientAddr params)
1082 (tcpReuseClientAddr params)
1083 (connectTimeout hints)
1083 (connectTimeout hints)
1084 didAccept <- case result of
1084 didAccept <- case result of
1085− Right (sock, ConnectionRequestAccepted) -> do
1085+ Right (sock, ConnectionRequestAccepted) -> do
1086− sendLock <- newMVar ()
1086+ sendLock <- newMVar ()
1087− let vst = ValidRemoteEndPointState
1087+ let vst = ValidRemoteEndPointState
1088 { remoteSocket = sock
1088 { remoteSocket = sock
1089 , remoteSendLock = sendLock
1089 , remoteSendLock = sendLock
1090− , _remoteOutgoing = 0
1090+ , _remoteOutgoing = 0
1091 , _remoteIncoming = Set.empty
1091 , _remoteIncoming = Set.empty
1092 , _remoteMaxIncoming = 0
1092 , _remoteMaxIncoming = 0
1093− , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
1093+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
1095 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1095 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1096 return True
1096 return True
@@ -1101,7 +1101,7 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1101 return False
1101 return False
1102 Right (sock, ConnectionRequestCrossed) -> do
1102 Right (sock, ConnectionRequestCrossed) -> do
1103 withMVar (remoteState theirEndPoint) $ \st -> case st of
1103 withMVar (remoteState theirEndPoint) $ \st -> case st of
1104− RemoteEndPointInit _ crossed _ ->
1104+ RemoteEndPointInit _ crossed _ ->
1105 putMVar crossed ()
1105 putMVar crossed ()
1106 RemoteEndPointFailed ex ->
1106 RemoteEndPointFailed ex ->
1107 throwIO ex
1107 throwIO ex
@@ -1109,11 +1109,11 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1109 relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"
1109 relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"
1110 tryCloseSocket sock
1110 tryCloseSocket sock
1111 return False
1111 return False
1112− Left err -> do
1112+ Left err -> do
1113 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1113 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1114 return False
1114 return False
1116− when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
1116+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
1118 ourAddress = localAddress ourEndPoint
1118 ourAddress = localAddress ourEndPoint
1119 theirAddress = remoteAddress theirEndPoint
1119 theirAddress = remoteAddress theirEndPoint
@@ -1121,23 +1121,23 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1122-- | Send a CloseSocket request if the remote endpoint is unused
1122-- | Send a CloseSocket request if the remote endpoint is unused
1123closeIfUnused :: EndPointPair -> IO ()
1123closeIfUnused :: EndPointPair -> IO ()
1124−closeIfUnused (ourEndPoint, theirEndPoint) = do
1124+closeIfUnused (ourEndPoint, theirEndPoint) = do
1125 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1125 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1126 RemoteEndPointValid vst ->
1126 RemoteEndPointValid vst ->
1127− if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)
1127+ if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)
1128− then do
1128+ then do
1129 resolved <- newEmptyMVar
1129 resolved <- newEmptyMVar
1130− act <- schedule theirEndPoint $
1130+ act <- schedule theirEndPoint $
1131 sendOn vst [ encodeInt32 CloseSocket
1131 sendOn vst [ encodeInt32 CloseSocket
1132 , encodeInt32 (vst ^. remoteMaxIncoming)
1132 , encodeInt32 (vst ^. remoteMaxIncoming)
1134 return (RemoteEndPointClosing resolved vst, Just act)
1134 return (RemoteEndPointClosing resolved vst, Just act)
1136 return (RemoteEndPointValid vst, Nothing)
1136 return (RemoteEndPointValid vst, Nothing)
1138 return (st, Nothing)
1138 return (st, Nothing)
1139− forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
1139+ forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
1141-- | Reset a remote endpoint if it is in Invalid mode
1141-- | Reset a remote endpoint if it is in Invalid mode
1143-- If the remote endpoint is currently in broken state, and
1143-- If the remote endpoint is currently in broken state, and
@@ -1156,7 +1156,7 @@ resetIfBroken ourEndPoint theirAddress = do
1156 return (vst ^. localConnectionTo theirAddress)
1156 return (vst ^. localConnectionTo theirAddress)
1157 LocalEndPointClosed ->
1157 LocalEndPointClosed ->
1158 throwIO $ TransportError ConnectFailed "Endpoint closed"
1158 throwIO $ TransportError ConnectFailed "Endpoint closed"
1159− forM_ mTheirEndPoint $ \theirEndPoint ->
1159+ forM_ mTheirEndPoint $ \theirEndPoint ->
1160 withMVar (remoteState theirEndPoint) $ \st -> case st of
1160 withMVar (remoteState theirEndPoint) $ \st -> case st of
1161 RemoteEndPointInvalid _ ->
1161 RemoteEndPointInvalid _ ->
1162 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1162 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
@@ -1166,26 +1166,26 @@ resetIfBroken ourEndPoint theirAddress = do
1166 return ()
1166 return ()
1168-- | Special case of 'apiConnect': connect an endpoint to itself
1168-- | Special case of 'apiConnect': connect an endpoint to itself
1170-- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)
1170-- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)
1171−connectToSelf :: LocalEndPoint
1171+connectToSelf :: LocalEndPoint
1172 -> IO Connection
1172 -> IO Connection
1173−connectToSelf ourEndPoint = do
1173+connectToSelf ourEndPoint = do
1174 connAlive <- newIORef True -- Protected by the local endpoint lock
1174 connAlive <- newIORef True -- Protected by the local endpoint lock
1175− lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
1175+ lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
1176 let connId = createConnectionId heavyweightSelfConnectionId lconnId
1176 let connId = createConnectionId heavyweightSelfConnectionId lconnId
1177 writeChan ourChan $
1177 writeChan ourChan $
1178 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
1178 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
1179− return Connection
1179+ return Connection
1180− { send = selfSend connAlive connId
1180+ { send = selfSend connAlive connId
1181 , close = selfClose connAlive connId
1181 , close = selfClose connAlive connId
1184− selfSend :: IORef Bool
1184+ selfSend :: IORef Bool
1185− -> ConnectionId
1185+ -> ConnectionId
1186− -> [ByteString]
1186+ -> [ByteString]
1187 -> IO (Either (TransportError SendErrorCode) ())
1187 -> IO (Either (TransportError SendErrorCode) ())
1188− selfSend connAlive connId msg =
1188+ selfSend connAlive connId msg =
1189 try . withMVar ourState $ \st -> case st of
1189 try . withMVar ourState $ \st -> case st of
1190 LocalEndPointValid _ -> do
1190 LocalEndPointValid _ -> do
1191 alive <- readIORef connAlive
1191 alive <- readIORef connAlive
@@ -1196,19 +1196,19 @@ connectToSelf ourEndPoint = do
1196 throwIO $ TransportError SendFailed "Endpoint closed"
1196 throwIO $ TransportError SendFailed "Endpoint closed"
1198 selfClose :: IORef Bool -> ConnectionId -> IO ()
1198 selfClose :: IORef Bool -> ConnectionId -> IO ()
1199− selfClose connAlive connId =
1199+ selfClose connAlive connId =
1200 withMVar ourState $ \st -> case st of
1200 withMVar ourState $ \st -> case st of
1201 LocalEndPointValid _ -> do
1201 LocalEndPointValid _ -> do
1202 alive <- readIORef connAlive
1202 alive <- readIORef connAlive
1203 when alive $ do
1203 when alive $ do
1204− writeChan ourChan (ConnectionClosed connId)
1204+ writeChan ourChan (ConnectionClosed connId)
1205 writeIORef connAlive False
1205 writeIORef connAlive False
1206 LocalEndPointClosed ->
1206 LocalEndPointClosed ->
1207− return ()
1207+ return ()
1209 ourChan = localChannel ourEndPoint
1209 ourChan = localChannel ourEndPoint
1210 ourState = localState ourEndPoint
1210 ourState = localState ourEndPoint
1211− connectFailed = TransportError ConnectFailed . show
1211+ connectFailed = TransportError ConnectFailed . show
1213-- | Resolve an endpoint currently in 'Init' state
1213-- | Resolve an endpoint currently in 'Init' state
1214resolveInit :: EndPointPair -> RemoteState -> IO ()
1214resolveInit :: EndPointPair -> RemoteState -> IO ()
@@ -1216,57 +1216,57 @@ resolveInit (ourEndPoint, theirEndPoint) newState =
1216 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
1216 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
1217 RemoteEndPointInit resolved _ _ -> do
1217 RemoteEndPointInit resolved _ _ -> do
1218 putMVar resolved ()
1218 putMVar resolved ()
1219− case newState of
1219+ case newState of
1220− RemoteEndPointClosed ->
1220+ RemoteEndPointClosed ->
1221 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1221 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1223 return ()
1223 return ()
1224 return newState
1224 return newState
1225− RemoteEndPointFailed ex ->
1225+ RemoteEndPointFailed ex ->
1226 throwIO ex
1226 throwIO ex
1228 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
1228 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
1230-- | Get the next outgoing self-connection ID
1230-- | Get the next outgoing self-connection ID
1232-- Throws an IO exception when the endpoint is closed.
1232-- Throws an IO exception when the endpoint is closed.
1233getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1233getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1234−getLocalNextConnOutId ourEndpoint =
1234+getLocalNextConnOutId ourEndpoint =
1235 modifyMVar (localState ourEndpoint) $ \st -> case st of
1235 modifyMVar (localState ourEndpoint) $ \st -> case st of
1236 LocalEndPointValid vst -> do
1236 LocalEndPointValid vst -> do
1237− let connId = vst ^. localNextConnOutId
1237+ let connId = vst ^. localNextConnOutId
1238− return ( LocalEndPointValid
1238+ return ( LocalEndPointValid
1239− . (localNextConnOutId ^= connId + 1)
1239+ . (localNextConnOutId ^= connId + 1)
1241 , connId)
1241 , connId)
1242 LocalEndPointClosed ->
1242 LocalEndPointClosed ->
1243 throwIO $ userError "Local endpoint closed"
1243 throwIO $ userError "Local endpoint closed"
1245-- | Create a new local endpoint
1245-- | Create a new local endpoint
1247-- May throw a TransportError NewEndPointErrorCode exception if the transport
1247-- May throw a TransportError NewEndPointErrorCode exception if the transport
1248-- is closed.
1248-- is closed.
1249createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
1249createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
1250−createLocalEndPoint transport = do
1250+createLocalEndPoint transport = do
1251 chan <- newChan
1251 chan <- newChan
1252− state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1252+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1253 { _localNextConnOutId = firstNonReservedLightweightConnectionId
1253 { _localNextConnOutId = firstNonReservedLightweightConnectionId
1254 , _localConnections = Map.empty
1254 , _localConnections = Map.empty
1255− , _nextConnInId = firstNonReservedHeavyweightConnectionId
1255+ , _nextConnInId = firstNonReservedHeavyweightConnectionId
1257 modifyMVar (transportState transport) $ \st -> case st of
1257 modifyMVar (transportState transport) $ \st -> case st of
1258 TransportValid vst -> do
1258 TransportValid vst -> do
1259 let ix = vst ^. nextEndPointId
1259 let ix = vst ^. nextEndPointId
1260− let addr = encodeEndPointAddress (transportHost transport)
1260+ let addr = encodeEndPointAddress (transportHost transport)
1261 (transportPort transport)
1261 (transportPort transport)
1263 let localEndPoint = LocalEndPoint { localAddress = addr
1263 let localEndPoint = LocalEndPoint { localAddress = addr
1264 , localChannel = chan
1264 , localChannel = chan
1265 , localState = state
1265 , localState = state
1267− return ( TransportValid
1267+ return ( TransportValid
1268− . (localEndPointAt addr ^= Just localEndPoint)
1268+ . (localEndPointAt addr ^= Just localEndPoint)
1269− . (nextEndPointId ^= ix + 1)
1269+ . (nextEndPointId ^= ix + 1)
1271 , localEndPoint
1271 , localEndPoint
@@ -1282,13 +1282,13 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
1282 modifyMVar_ ourState $ \st -> case st of
1282 modifyMVar_ ourState $ \st -> case st of
1283 LocalEndPointValid vst ->
1283 LocalEndPointValid vst ->
1284 case vst ^. localConnectionTo theirAddress of
1284 case vst ^. localConnectionTo theirAddress of
1285− Nothing ->
1285+ Nothing ->
1286 return st
1286 return st
1287 Just remoteEndPoint' ->
1287 Just remoteEndPoint' ->
1288− if remoteId remoteEndPoint' == remoteId theirEndPoint
1288+ if remoteId remoteEndPoint' == remoteId theirEndPoint
1289− then return
1289+ then return
1290− ( LocalEndPointValid
1290+ ( LocalEndPointValid
1291− . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
1291+ . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
1294 else return st
1294 else return st
@@ -1302,23 +1302,23 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
1303-- Does nothing if the transport is closed
1303-- Does nothing if the transport is closed
1304removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
1304removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
1305−removeLocalEndPoint transport ourEndPoint =
1305+removeLocalEndPoint transport ourEndPoint =
1306− modifyMVar_ (transportState transport) $ \st -> case st of
1306+ modifyMVar_ (transportState transport) $ \st -> case st of
1307 TransportValid vst ->
1307 TransportValid vst ->
1308− return ( TransportValid
1308+ return ( TransportValid
1309− . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
1309+ . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
1312 TransportClosed ->
1312 TransportClosed ->
1313 return TransportClosed
1313 return TransportClosed
1315-- | Find a remote endpoint. If the remote endpoint does not yet exist we
1315-- | 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.
1316+-- create it in Init state. Returns if the endpoint was new.
1317−findRemoteEndPoint
1317+findRemoteEndPoint
1318 :: LocalEndPoint
1318 :: LocalEndPoint
1319 -> EndPointAddress
1319 -> EndPointAddress
1320− -> RequestedBy
1320+ -> RequestedBy
1321− -> IO (RemoteEndPoint, Bool)
1321+ -> IO (RemoteEndPoint, Bool)
1322findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1322findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
@@ -1335,55 +1335,55 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1335 { remoteAddress = theirAddress
1335 { remoteAddress = theirAddress
1336 , remoteState = theirState
1336 , remoteState = theirState
1337 , remoteId = vst ^. nextConnInId
1337 , remoteId = vst ^. nextConnInId
1338− , remoteScheduled = scheduled
1338+ , remoteScheduled = scheduled
1340− return ( LocalEndPointValid
1340+ return ( LocalEndPointValid
1341− . (localConnectionTo theirAddress ^= Just theirEndPoint)
1341+ . (localConnectionTo theirAddress ^= Just theirEndPoint)
1342− . (nextConnInId ^: (+ 1))
1342+ . (nextConnInId ^: (+ 1))
1344− , (theirEndPoint, True)
1344+ , (theirEndPoint, True)
1346 LocalEndPointClosed ->
1346 LocalEndPointClosed ->
1347 throwIO $ userError "Local endpoint closed"
1347 throwIO $ userError "Local endpoint closed"
1349− if isNew
1349+ if isNew
1351 return (theirEndPoint, True)
1351 return (theirEndPoint, True)
1353 let theirState = remoteState theirEndPoint
1353 let theirState = remoteState theirEndPoint
1354 snapshot <- modifyMVar theirState $ \st -> case st of
1354 snapshot <- modifyMVar theirState $ \st -> case st of
1355− RemoteEndPointValid vst ->
1355+ RemoteEndPointValid vst ->
1356 case findOrigin of
1356 case findOrigin of
1357 RequestedByUs -> do
1357 RequestedByUs -> do
1358− let st' = RemoteEndPointValid
1358+ let st' = RemoteEndPointValid
1359− . (remoteOutgoing ^: (+ 1))
1359+ . (remoteOutgoing ^: (+ 1))
1361 return (st', st')
1361 return (st', st')
1362 RequestedByThem ->
1362 RequestedByThem ->
1363− return (st, st)
1363+ return (st, st)
1365 return (st, st)
1365 return (st, st)
1366 -- The snapshot may no longer be up to date at this point, but if we
1366 -- 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
1367+ -- increased the refcount then it can only either be Valid or Failed
1368− -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
1368+ -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
1369 case snapshot of
1369 case snapshot of
1370 RemoteEndPointInvalid err ->
1370 RemoteEndPointInvalid err ->
1371 throwIO err
1371 throwIO err
1372 RemoteEndPointInit resolved crossed initOrigin ->
1372 RemoteEndPointInit resolved crossed initOrigin ->
1373 case (findOrigin, initOrigin) of
1373 case (findOrigin, initOrigin) of
1374− (RequestedByUs, RequestedByUs) ->
1374+ (RequestedByUs, RequestedByUs) ->
1375− readMVar resolved >> go
1376− (RequestedByUs, RequestedByThem) ->
1377 readMVar resolved >> go
1375 readMVar resolved >> go
1376+ (RequestedByUs, RequestedByThem) ->
1377+ readMVar resolved >> go
1378− (RequestedByThem, RequestedByUs) ->
1378+ (RequestedByThem, RequestedByUs) ->
1379− if ourAddress > theirAddress
1379+ if ourAddress > theirAddress
1381 -- Wait for the Crossed message
1381 -- Wait for the Crossed message
1382− readMVar crossed
1382+ readMVar crossed
1383 return (theirEndPoint, True)
1383 return (theirEndPoint, True)
1385 return (theirEndPoint, False)
1385 return (theirEndPoint, False)
1386− (RequestedByThem, RequestedByThem) ->
1386+ (RequestedByThem, RequestedByThem) ->
1387 throwIO $ userError "Already connected"
1387 throwIO $ userError "Already connected"
1388 RemoteEndPointValid _ ->
1388 RemoteEndPointValid _ ->
1389 -- We assume that the request crossed if we find the endpoint in
1389 -- We assume that the request crossed if we find the endpoint in
@@ -1395,15 +1395,15 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1395 readMVar resolved >> go
1395 readMVar resolved >> go
1396 RemoteEndPointClosed ->
1396 RemoteEndPointClosed ->
1398− RemoteEndPointFailed err ->
1398+ RemoteEndPointFailed err ->
1399 throwIO err
1399 throwIO err
1401− ourState = localState ourEndPoint
1401+ ourState = localState ourEndPoint
1402 ourAddress = localAddress ourEndPoint
1402 ourAddress = localAddress ourEndPoint
1404-- | Send a payload over a heavyweight connection (thread safe)
1404-- | Send a payload over a heavyweight connection (thread safe)
1405−sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()
1405+sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()
1406−sendOn vst bs = withMVar (remoteSendLock vst) $ \() ->
1406+sendOn vst bs = withMVar (remoteSendLock vst) $ \() ->
1407 sendMany (remoteSocket vst) bs
1407 sendMany (remoteSocket vst) bs
1409--------------------------------------------------------------------------------
1409--------------------------------------------------------------------------------
@@ -1417,7 +1417,7 @@ type Action a = MVar (Either SomeException a)
1417schedule :: RemoteEndPoint -> IO a -> IO (Action a)
1417schedule :: RemoteEndPoint -> IO a -> IO (Action a)
1418schedule theirEndPoint act = do
1418schedule theirEndPoint act = do
1419 mvar <- newEmptyMVar
1419 mvar <- newEmptyMVar
1420− writeChan (remoteScheduled theirEndPoint) $
1420+ writeChan (remoteScheduled theirEndPoint) $
1421 catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)
1421 catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)
1422 return mvar
1422 return mvar
@@ -1429,7 +1429,7 @@ schedule theirEndPoint act = do
1429-- > runScheduledAction
1429-- > runScheduledAction
1431-- 'runScheduledAction' will run @p@ (it might run some other scheduled action).
1431-- '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
1433-- 'runScheduledAction' or by another).
1433-- 'runScheduledAction' or by another).
1434runScheduledAction :: EndPointPair -> Action a -> IO a
1434runScheduledAction :: EndPointPair -> Action a -> IO a
1435runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
1435runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
@@ -1438,21 +1438,21 @@ runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
1438 case ma of
1438 case ma of
1439 Right a -> return a
1439 Right a -> return a
1440 Left e -> do
1440 Left e -> do
1441− forM_ (fromException e) $ \ioe ->
1441+ forM_ (fromException e) $ \ioe ->
1442 modifyMVar_ (remoteState theirEndPoint) $ \st ->
1442 modifyMVar_ (remoteState theirEndPoint) $ \st ->
1443− case st of
1443+ case st of
1444 RemoteEndPointValid vst -> handleIOException ioe vst
1444 RemoteEndPointValid vst -> handleIOException ioe vst
1445 _ -> return (RemoteEndPointFailed ioe)
1445 _ -> return (RemoteEndPointFailed ioe)
1446 throwIO e
1446 throwIO e
1448− handleIOException :: IOException
1448+ handleIOException :: IOException
1449− -> ValidRemoteEndPointState
1449+ -> ValidRemoteEndPointState
1450− -> IO RemoteState
1450+ -> IO RemoteState
1451 handleIOException ex vst = do
1451 handleIOException ex vst = do
1452 tryCloseSocket (remoteSocket vst)
1452 tryCloseSocket (remoteSocket vst)
1453− let code = EventConnectionLost (remoteAddress theirEndPoint)
1453+ let code = EventConnectionLost (remoteAddress theirEndPoint)
1454 err = TransportError code (show ex)
1454 err = TransportError code (show ex)
1455− writeChan (localChannel ourEndPoint) $ ErrorEvent err
1455+ writeChan (localChannel ourEndPoint) $ ErrorEvent err
1456 return (RemoteEndPointFailed ex)
1456 return (RemoteEndPointFailed ex)
1458--------------------------------------------------------------------------------
1458--------------------------------------------------------------------------------
@@ -1462,25 +1462,25 @@ runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
1462-- | Establish a connection to a remote endpoint
1462-- | Establish a connection to a remote endpoint
1464-- Maybe throw a TransportError
1464-- Maybe throw a TransportError
1465−socketToEndPoint :: EndPointAddress -- ^ Our address
1465+socketToEndPoint :: EndPointAddress -- ^ Our address
1466 -> EndPointAddress -- ^ Their address
1466 -> EndPointAddress -- ^ Their address
1467 -> Bool -- ^ Use SO_REUSEADDR?
1467 -> Bool -- ^ Use SO_REUSEADDR?
1468− -> Maybe Int -- ^ Timeout for connect
1468+ -> Maybe Int -- ^ Timeout for connect
1469− -> IO (Either (TransportError ConnectErrorCode)
1469+ -> IO (Either (TransportError ConnectErrorCode)
1470− (N.Socket, ConnectionRequestResponse))
1470+ (N.Socket, ConnectionRequestResponse))
1471−socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1471+socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1472− try $ do
1472+ try $ do
1473− (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
1473+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
1474 Nothing -> throwIO (failed . userError $ "Could not parse")
1474 Nothing -> throwIO (failed . userError $ "Could not parse")
1475 Just dec -> return dec
1475 Just dec -> return dec
1476− addr:_ <- mapIOException invalidAddress $
1476+ addr:_ <- mapIOException invalidAddress $
1477 N.getAddrInfo Nothing (Just host) (Just port)
1477 N.getAddrInfo Nothing (Just host) (Just port)
1478 bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
1478 bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
1479− when reuseAddr $
1479+ when reuseAddr $
1480 mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
1480 mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
1481− mapIOException invalidAddress $
1481+ mapIOException invalidAddress $
1482− timeoutMaybe timeout timeoutError $
1482+ timeoutMaybe timeout timeoutError $
1483− N.connect sock (N.addrAddress addr)
1483+ N.connect sock (N.addrAddress addr)
1484 response <- mapIOException failed $ do
1484 response <- mapIOException failed $ do
1485 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
1485 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
1486 recvInt32 sock
1486 recvInt32 sock
@@ -1489,50 +1489,50 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1489 Just r -> return (sock, r)
1489 Just r -> return (sock, r)
1491 createSocket :: N.AddrInfo -> IO N.Socket
1491 createSocket :: N.AddrInfo -> IO N.Socket
1492− createSocket addr = mapIOException insufficientResources $
1492+ createSocket addr = mapIOException insufficientResources $
1493 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
1493 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
1495− invalidAddress = TransportError ConnectNotFound . show
1495+ invalidAddress = TransportError ConnectNotFound . show
1496− insufficientResources = TransportError ConnectInsufficientResources . show
1496+ insufficientResources = TransportError ConnectInsufficientResources . show
1497 failed = TransportError ConnectFailed . show
1497 failed = TransportError ConnectFailed . show
1498 timeoutError = TransportError ConnectTimeout "Timed out"
1498 timeoutError = TransportError ConnectTimeout "Timed out"
1500-- | Encode end point address
1500-- | Encode end point address
1501−encodeEndPointAddress :: N.HostName
1501+encodeEndPointAddress :: N.HostName
1502− -> N.ServiceName
1502+ -> N.ServiceName
1503− -> EndPointId
1503+ -> EndPointId
1504 -> EndPointAddress
1504 -> EndPointAddress
1505encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
1505encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
1506− host ++ ":" ++ port ++ ":" ++ show ix
1506+ host ++ ":" ++ port ++ ":" ++ show ix
1508-- | Decode end point address
1508-- | Decode end point address
1509−decodeEndPointAddress :: EndPointAddress
1509+decodeEndPointAddress :: EndPointAddress
1510 -> Maybe (N.HostName, N.ServiceName, EndPointId)
1510 -> Maybe (N.HostName, N.ServiceName, EndPointId)
1511−decodeEndPointAddress (EndPointAddress bs) =
1511+decodeEndPointAddress (EndPointAddress bs) =
1512 case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of
1512 case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of
1513− [host, port, endPointIdStr] ->
1513+ [host, port, endPointIdStr] ->
1514− case reads endPointIdStr of
1514+ case reads endPointIdStr of
1515 [(endPointId, "")] -> Just (host, port, endPointId)
1515 [(endPointId, "")] -> Just (host, port, endPointId)
1516 _ -> Nothing
1516 _ -> Nothing
1520-- | Construct a ConnectionId
1520-- | Construct a ConnectionId
1521−createConnectionId :: HeavyweightConnectionId
1521+createConnectionId :: HeavyweightConnectionId
1522− -> LightweightConnectionId
1522+ -> LightweightConnectionId
1523 -> ConnectionId
1523 -> ConnectionId
1524−createConnectionId hcid lcid =
1524+createConnectionId hcid lcid =
1525 (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid
1525 (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid
1527-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,
1527-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,
1528-- returning at most @p@ segments -- counting from the /end/
1528-- returning at most @p@ segments -- counting from the /end/
1530-- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"]
1530-- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"]
1531splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]]
1531splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]]
1532−splitMaxFromEnd p = \n -> go [[]] n . reverse
1532+splitMaxFromEnd p = \n -> go [[]] n . reverse
1534 -- go :: [[a]] -> Int -> [a] -> [[a]]
1534 -- go :: [[a]] -> Int -> [a] -> [[a]]
1535− go accs _ [] = accs
1535+ go accs _ [] = accs
1536 go ([] : accs) 0 xs = reverse xs : accs
1536 go ([] : accs) 0 xs = reverse xs : accs
1537 go (acc : accs) n (x:xs) =
1537 go (acc : accs) n (x:xs) =
1538 if p x then go ([] : acc : accs) (n - 1) xs
1538 if p x then go ([] : acc : accs) (n - 1) xs
@@ -1544,40 +1544,40 @@ splitMaxFromEnd p = \n -> go [[]] n . reverse
1544--------------------------------------------------------------------------------
1544--------------------------------------------------------------------------------
1546-- Find a socket between two endpoints
1546-- Find a socket between two endpoints
1548-- Throws an IO exception if the socket could not be found.
1548-- Throws an IO exception if the socket could not be found.
1549−internalSocketBetween :: TCPTransport -- ^ Transport
1549+internalSocketBetween :: TCPTransport -- ^ Transport
1550 -> EndPointAddress -- ^ Local endpoint
1550 -> EndPointAddress -- ^ Local endpoint
1551 -> EndPointAddress -- ^ Remote endpoint
1551 -> EndPointAddress -- ^ Remote endpoint
1552− -> IO N.Socket
1552+ -> IO N.Socket
1553internalSocketBetween transport ourAddress theirAddress = do
1553internalSocketBetween transport ourAddress theirAddress = do
1554 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1554 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1555− TransportClosed ->
1555+ TransportClosed ->
1556− throwIO $ userError "Transport closed"
1556+ throwIO $ userError "Transport closed"
1557− TransportValid vst ->
1557+ TransportValid vst ->
1558 case vst ^. localEndPointAt ourAddress of
1558 case vst ^. localEndPointAt ourAddress of
1559 Nothing -> throwIO $ userError "Local endpoint not found"
1559 Nothing -> throwIO $ userError "Local endpoint not found"
1560 Just ep -> return ep
1560 Just ep -> return ep
1561 theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
1561 theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
1562 LocalEndPointClosed ->
1562 LocalEndPointClosed ->
1563 throwIO $ userError "Local endpoint closed"
1563 throwIO $ userError "Local endpoint closed"
1564− LocalEndPointValid vst ->
1564+ LocalEndPointValid vst ->
1565 case vst ^. localConnectionTo theirAddress of
1565 case vst ^. localConnectionTo theirAddress of
1566 Nothing -> throwIO $ userError "Remote endpoint not found"
1566 Nothing -> throwIO $ userError "Remote endpoint not found"
1567 Just ep -> return ep
1567 Just ep -> return ep
1568 withMVar (remoteState theirEndPoint) $ \st -> case st of
1568 withMVar (remoteState theirEndPoint) $ \st -> case st of
1569 RemoteEndPointInit _ _ _ ->
1569 RemoteEndPointInit _ _ _ ->
1570 throwIO $ userError "Remote endpoint not yet initialized"
1570 throwIO $ userError "Remote endpoint not yet initialized"
1571− RemoteEndPointValid vst ->
1571+ RemoteEndPointValid vst ->
1572 return $ remoteSocket vst
1572 return $ remoteSocket vst
1573 RemoteEndPointClosing _ vst ->
1573 RemoteEndPointClosing _ vst ->
1574− return $ remoteSocket vst
1574+ return $ remoteSocket vst
1575 RemoteEndPointClosed ->
1575 RemoteEndPointClosed ->
1576 throwIO $ userError "Remote endpoint closed"
1576 throwIO $ userError "Remote endpoint closed"
1577 RemoteEndPointInvalid err ->
1577 RemoteEndPointInvalid err ->
1578− throwIO err
1578+ throwIO err
1579 RemoteEndPointFailed err ->
1579 RemoteEndPointFailed err ->
1580− throwIO err
1580+ throwIO err
1582--------------------------------------------------------------------------------
1582--------------------------------------------------------------------------------
1583-- Constants --
1583-- Constants --
@@ -1611,7 +1611,7 @@ localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextCon
1611localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
1611localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
1612localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
1612localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
1614−nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId
1614+nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId
1615nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })
1615nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })
1617remoteOutgoing :: Accessor ValidRemoteEndPointState Int
1617remoteOutgoing :: Accessor ValidRemoteEndPointState Int
@@ -1627,10 +1627,10 @@ remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId
1627remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })
1627remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })
1629localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1629localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1630−localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1630+localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1632localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)
1632localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)
1633−localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1633+localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1635-------------------------------------------------------------------------------
1635-------------------------------------------------------------------------------
1636-- Debugging --
1636-- Debugging --
@@ -1638,14 +1638,14 @@ localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1639relyViolation :: EndPointPair -> String -> IO a
1639relyViolation :: EndPointPair -> String -> IO a
1640relyViolation (ourEndPoint, theirEndPoint) str = do
1640relyViolation (ourEndPoint, theirEndPoint) str = do
1641− elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation")
1641+ elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation")
1642 fail (str ++ " RELY violation")
1642 fail (str ++ " RELY violation")
1644elog :: EndPointPair -> String -> IO ()
1644elog :: EndPointPair -> String -> IO ()
1645elog (ourEndPoint, theirEndPoint) msg = do
1645elog (ourEndPoint, theirEndPoint) msg = do
1646 tid <- myThreadId
1646 tid <- myThreadId
1647− putStrLn $ show (localAddress ourEndPoint)
1647+ putStrLn $ show (localAddress ourEndPoint)
1648− ++ "/" ++ show (remoteAddress theirEndPoint)
1648+ ++ "/" ++ show (remoteAddress theirEndPoint)
1649 ++ "(" ++ show (remoteId theirEndPoint) ++ ")"
1649 ++ "(" ++ show (remoteId theirEndPoint) ++ ")"
1650− ++ "/" ++ show tid
1650+ ++ "/" ++ show tid
1651 ++ ": " ++ msg
1651 ++ ": " ++ msg