network-transport-composed (empty) → 0.2.0.0
raw patch · 4 files changed
+241/−0 lines, 4 filesdep +basedep +bytestringdep +network-transportsetup-changed
Dependencies added: base, bytestring, network-transport
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- network-transport-composed.cabal +28/−0
- src/Network/Transport/Composed.hs +181/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2012, Edsko de Vries++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Edsko de Vries nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ network-transport-composed.cabal view
@@ -0,0 +1,28 @@+name: network-transport-composed+version: 0.2.0.0+synopsis: Compose network transports+-- description: +homepage: http://haskell-distributed.github.com+license: BSD3+license-file: LICENSE+author: Edsko de Vries+maintainer: edsko@well-typed.com, watson.timothy@gmail.com+Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTCOMP+copyright: Well-Typed LLP+category: Network+build-type: Simple+cabal-version: >=1.8++Source-Repository head+ Type: git+ Location: https://github.com/haskell-distributed/network-transport-composed++library+ exposed-modules: Network.Transport.Composed+ -- other-modules: + build-depends: base >= 4.3 && < 5,+ bytestring,+ network-transport >= 0.4.0.0 && < 0.5+ hs-source-dirs: src+ ghc-options: -Wall+ extensions: BangPatterns
+ src/Network/Transport/Composed.hs view
@@ -0,0 +1,181 @@+-- | Combine two transports into a single new transport+--+-- Given two transports A and B, endpoints in the combined transport are+-- pairs of endpoints for the underlying transports. Consequently, the address+-- of an endpoint in the combined transport is a pair of addresses of the+-- two underlying endpoints. Events from both underlying endpoints are posted+-- to the combined endpoint.+--+-- When the address of combined endpoint C1 is sent to combined endpoint C2,+-- and C2 attempts to make a connection to C1, it needs to choice an underlying+-- endpoint to make the connection. The choice is made by a parameter+-- 'resolveAddress' of the composed transport.+--+-- When C2 connects to C1 then it is assumed that whatever transport C2 uses+-- to connect to C1, this is the same transport that C1 needs to use to connect+-- back to C2.+--+-- TODO: at the moment the address reported by a ConnectionOpened might differ+-- from the address reported by the remote endpoint itself. This might cause+-- difficulties?+module Network.Transport.Composed+ ( ComposedTransport(..)+ , createTransport+ ) where++import qualified Data.ByteString as BSS+ ( concat+ , length+ , splitAt+ , singleton+ , uncons+ )+import Control.Monad (void)+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)+import Control.Concurrent (forkIO)+import Network.Transport+ ( Transport(..)+ , EndPointAddress(EndPointAddress)+ , TransportError(TransportError)+ , NewEndPointErrorCode+ , EndPoint(..)+ , NewMulticastGroupErrorCode(NewMulticastGroupUnsupported)+ , ResolveMulticastGroupErrorCode(ResolveMulticastGroupUnsupported)+ , Event(EndPointClosed, ConnectionOpened, ErrorEvent)+ , EventErrorCode(EventConnectionLost)+ , Reliability+ , ConnectHints+ , ConnectErrorCode+ , Connection+ )+import Network.Transport.Internal (encodeInt32, decodeInt32)++data ComposedTransport = ComposedTransport {+ -- | The left transport+ transportA :: Transport+ -- | The right transport+ , transportB :: Transport+ -- | Pick a suitable transport for an address+ , resolveAddress :: (EndPoint, EndPoint)+ -> (EndPointAddress, EndPointAddress)+ -> IO (EndPoint, EndPointAddress)+ }++data ComposedAddress =+ LeftAddress EndPointAddress+ | RightAddress EndPointAddress+ | EitherAddress EndPointAddress EndPointAddress++serializeComposedAddress :: ComposedAddress -> EndPointAddress+serializeComposedAddress (LeftAddress (EndPointAddress addr)) =+ EndPointAddress $ BSS.concat [BSS.singleton 0, addr]+serializeComposedAddress (RightAddress (EndPointAddress addr)) =+ EndPointAddress $ BSS.concat [BSS.singleton 1, addr]+serializeComposedAddress (EitherAddress (EndPointAddress addr1) (EndPointAddress addr2)) =+ EndPointAddress $ BSS.concat [BSS.singleton 2, encodeInt32 (BSS.length addr1), addr1, addr2]++deserializeComposedAddress :: EndPointAddress -> ComposedAddress+deserializeComposedAddress (EndPointAddress addr) =+ let (header, remainder) = case BSS.uncons addr of+ Just (x, xs) -> (x, xs)+ Nothing -> error "deserializeComposedAddress"+ in case header of+ 0 -> LeftAddress (EndPointAddress remainder)+ 1 -> RightAddress (EndPointAddress remainder)+ 2 -> let (len, remainder') = BSS.splitAt 4 remainder+ (a, b) = BSS.splitAt (decodeInt32 len) remainder'+ in EitherAddress (EndPointAddress a) (EndPointAddress b)+ _ -> error "deserializeComposedAddress"++createTransport :: ComposedTransport -> IO Transport+createTransport composed =+ return Transport {+ newEndPoint = apiNewEndPoint composed+ , closeTransport = apiCloseTransport composed+ }++apiNewEndPoint :: ComposedTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)+apiNewEndPoint composed = do+ mEndPoints <- createEndPoints+ case mEndPoints of+ Left err ->+ return $ Left err+ Right (endPointA, endPointB) -> do+ let addr = serializeComposedAddress (EitherAddress (address endPointA) (address endPointB))+ chan <- newChan+ void . forkIO $ forwardEvents endPointA chan LeftAddress+ void . forkIO $ forwardEvents endPointB chan RightAddress+ return . Right $ EndPoint {+ receive = readChan chan+ , address = addr+ , connect = apiConnect composed endPointA endPointB+ , closeEndPoint = apiCloseEndPoint endPointA endPointB+ , newMulticastGroup = return . Left $ newMulticastGroupError+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError+ }+ where+ newMulticastGroupError =+ TransportError NewMulticastGroupUnsupported "Multicast not supported"+ resolveMulticastGroupError =+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"++ createEndPoints = do+ mEndPointA <- newEndPoint (transportA composed)+ case mEndPointA of+ Left err -> return $ Left err+ Right endPointA -> do+ mEndPointB <- newEndPoint (transportB composed)+ case mEndPointB of+ Left err -> do+ closeEndPoint endPointA+ return $ Left err+ Right endPointB ->+ return $ Right (endPointA, endPointB)++ forwardEvents :: EndPoint -> Chan Event -> (EndPointAddress -> ComposedAddress) -> IO ()+ forwardEvents endPoint chan inj = go+ where+ go = do+ event <- receive endPoint+ case event of+ ConnectionOpened cid rel addr -> do+ let addr' = serializeComposedAddress (inj addr)+ writeChan chan (ConnectionOpened cid rel addr')+ go+ ErrorEvent (TransportError (EventConnectionLost addr) msg) -> do+ let addr' = serializeComposedAddress (inj addr)+ writeChan chan $ ErrorEvent (TransportError (EventConnectionLost addr') msg)+ go+ EndPointClosed -> do+ writeChan chan event+ return ()+ _ -> do+ writeChan chan event+ go++apiCloseTransport :: ComposedTransport -> IO ()+apiCloseTransport composed = do+ closeTransport (transportA composed)+ closeTransport (transportB composed)++apiConnect :: ComposedTransport+ -> EndPoint+ -> EndPoint+ -> EndPointAddress+ -> Reliability+ -> ConnectHints+ -> IO (Either (TransportError ConnectErrorCode) Connection)+apiConnect composed endPointA endPointB theirComposedAddr rel hints =+ case deserializeComposedAddress theirComposedAddr of+ LeftAddress addr ->+ connect endPointA addr rel hints+ RightAddress addr ->+ connect endPointB addr rel hints+ EitherAddress addrA addrB -> do+ (endPoint, addr) <- resolveAddress composed (endPointA, endPointB) (addrA, addrB)+ connect endPoint addr rel hints++apiCloseEndPoint :: EndPoint -> EndPoint -> IO ()+apiCloseEndPoint endPointA endPointB = do+ closeEndPoint endPointA+ closeEndPoint endPointB