@@ -0,0 +1,177 @@
1+-- | Combine two transports into a single new transport
2+--
3+-- Given two transports A and B, endpoints in the combined transport are
4+-- pairs of endpoints for the underlying transports. Consequently, the address
5+-- of an endpoint in the combined transport is a pair of addresses of the
6+-- two underlying endpoints. Events from both underlying endpoints are posted
7+-- to the combined endpoint.
8+--
9+-- When the address of combined endpoint C1 is sent to combined endpoint C2,
10+-- and C2 attempts to make a connection to C1, it needs to choice an underlying
11+-- endpoint to make the connection. The choice is made by a parameter
12+-- 'resolveAddress' of the composed transport.
13+--
14+-- When C2 connects to C1 then it is assumed that whatever transport C2 uses
15+-- to connect to C1, this is the same transport that C1 needs to use to connect
16+-- back to C2.
17+module Network.Transport.Composed
18+ ( ComposedTransport(..)
19+ , createTransport
20+ ) where
21+
22+import qualified Data.ByteString as BSS
23+ ( concat
24+ , length
25+ , splitAt
26+ , singleton
27+ , uncons
28+ )
29+import Control.Monad (void)
30+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)
31+import Control.Concurrent (forkIO)
32+import Network.Transport
33+ ( Transport(..)
34+ , EndPointAddress(EndPointAddress)
35+ , TransportError(TransportError)
36+ , NewEndPointErrorCode
37+ , EndPoint(..)
38+ , NewMulticastGroupErrorCode(NewMulticastGroupUnsupported)
39+ , ResolveMulticastGroupErrorCode(ResolveMulticastGroupUnsupported)
40+ , Event(EndPointClosed, ConnectionOpened, ErrorEvent)
41+ , EventErrorCode(EventConnectionLost)
42+ , Reliability
43+ , ConnectHints
44+ , ConnectErrorCode
45+ , Connection
46+ )
47+import Network.Transport.Internal (encodeInt32, decodeInt32)
48+
49+data ComposedTransport = ComposedTransport {
50+ -- | The left transport
51+ transportA :: Transport
52+ -- | The right transport
53+ , transportB :: Transport
54+ -- | Pick a suitable transport for an address
55+ , resolveAddress :: (EndPoint, EndPoint)
56+ -> (EndPointAddress, EndPointAddress)
57+ -> IO (EndPoint, EndPointAddress)
58+ }
59+
60+data ComposedAddress =
61+ LeftAddress EndPointAddress
62+ | RightAddress EndPointAddress
63+ | EitherAddress EndPointAddress EndPointAddress
64+
65+serializeComposedAddress :: ComposedAddress -> EndPointAddress
66+serializeComposedAddress (LeftAddress (EndPointAddress addr)) =
67+ EndPointAddress $ BSS.concat [BSS.singleton 0, addr]
68+serializeComposedAddress (RightAddress (EndPointAddress addr)) =
69+ EndPointAddress $ BSS.concat [BSS.singleton 1, addr]
70+serializeComposedAddress (EitherAddress (EndPointAddress addr1) (EndPointAddress addr2)) =
71+ EndPointAddress $ BSS.concat [BSS.singleton 2, encodeInt32 (BSS.length addr1), addr1, addr2]
72+
73+deserializeComposedAddress :: EndPointAddress -> ComposedAddress
74+deserializeComposedAddress (EndPointAddress addr) =
75+ let (header, remainder) = case BSS.uncons addr of
76+ Just (x, xs) -> (x, xs)
77+ Nothing -> error "deserializeComposedAddress"
78+ in case header of
79+ 0 -> LeftAddress (EndPointAddress remainder)
80+ 1 -> RightAddress (EndPointAddress remainder)
81+ 2 -> let (len, remainder') = BSS.splitAt 4 remainder
82+ (a, b) = BSS.splitAt (decodeInt32 len) remainder'
83+ in EitherAddress (EndPointAddress a) (EndPointAddress b)
84+ _ -> error "deserializeComposedAddress"
85+
86+createTransport :: ComposedTransport -> IO Transport
87+createTransport composed =
88+ return Transport {
89+ newEndPoint = apiNewEndPoint composed
90+ , closeTransport = apiCloseTransport composed
91+ }
92+
93+apiNewEndPoint :: ComposedTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
94+apiNewEndPoint composed = do
95+ mEndPoints <- createEndPoints
96+ case mEndPoints of
97+ Left err ->
98+ return $ Left err
99+ Right (endPointA, endPointB) -> do
100+ let addr = serializeComposedAddress (EitherAddress (address endPointA) (address endPointB))
101+ chan <- newChan
102+ void . forkIO $ forwardEvents endPointA chan LeftAddress
103+ void . forkIO $ forwardEvents endPointB chan RightAddress
104+ return . Right $ EndPoint {
105+ receive = readChan chan
106+ , address = addr
107+ , connect = apiConnect composed endPointA endPointB
108+ , closeEndPoint = apiCloseEndPoint endPointA endPointB
109+ , newMulticastGroup = return . Left $ newMulticastGroupError
110+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
111+ }
112+ where
113+ newMulticastGroupError =
114+ TransportError NewMulticastGroupUnsupported "Multicast not supported"
115+ resolveMulticastGroupError =
116+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
117+
118+ createEndPoints = do
119+ mEndPointA <- newEndPoint (transportA composed)
120+ case mEndPointA of
121+ Left err -> return $ Left err
122+ Right endPointA -> do
123+ mEndPointB <- newEndPoint (transportB composed)
124+ case mEndPointB of
125+ Left err -> do
126+ closeEndPoint endPointA
127+ return $ Left err
128+ Right endPointB ->
129+ return $ Right (endPointA, endPointB)
130+
131+ forwardEvents :: EndPoint -> Chan Event -> (EndPointAddress -> ComposedAddress) -> IO ()
132+ forwardEvents endPoint chan inj = go
133+ where
134+ go = do
135+ event <- receive endPoint
136+ case event of
137+ ConnectionOpened cid rel addr -> do
138+ let addr' = serializeComposedAddress (inj addr)
139+ writeChan chan (ConnectionOpened cid rel addr')
140+ go
141+ ErrorEvent (TransportError (EventConnectionLost addr) msg) -> do
142+ let addr' = serializeComposedAddress (inj addr)
143+ writeChan chan $ ErrorEvent (TransportError (EventConnectionLost addr') msg)
144+ go
145+ EndPointClosed -> do
146+ writeChan chan event
147+ return ()
148+ _ -> do
149+ writeChan chan event
150+ go
151+
152+apiCloseTransport :: ComposedTransport -> IO ()
153+apiCloseTransport composed = do
154+ closeTransport (transportA composed)
155+ closeTransport (transportB composed)
156+
157+apiConnect :: ComposedTransport
158+ -> EndPoint
159+ -> EndPoint
160+ -> EndPointAddress
161+ -> Reliability
162+ -> ConnectHints
163+ -> IO (Either (TransportError ConnectErrorCode) Connection)
164+apiConnect composed endPointA endPointB theirComposedAddr rel hints =
165+ case deserializeComposedAddress theirComposedAddr of
166+ LeftAddress addr ->
167+ connect endPointA addr rel hints
168+ RightAddress addr ->
169+ connect endPointB addr rel hints
170+ EitherAddress addrA addrB -> do
171+ (endPoint, addr) <- resolveAddress composed (endPointA, endPointB) (addrA, addrB)
172+ connect endPoint addr rel hints
173+
174+apiCloseEndPoint :: EndPoint -> EndPoint -> IO ()
175+apiCloseEndPoint endPointA endPointB = do
176+ closeEndPoint endPointA
177+ closeEndPoint endPointB