diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,13 @@ +# Revision history for reflex-backend-socket + +## 0.2.0.0 -- 2019-10-29 (first hackage release) + +* Decoupled reflex-binary from this library. If you want incremental + decoding, you can feed the `ByteString`s into reflex-binary + yourself. + +* Simplified, rewrote and documented almost all of the library. + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2017, Dave Laing + +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 Dave Laing 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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/example/Client.hs b/example/Client.hs new file mode 100644 --- /dev/null +++ b/example/Client.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE FlexibleContexts #-} + +{-| +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable +-} + +module Main (main) where + +import Control.Concurrent (forkIO) +import Control.Monad (void, forever) +import Control.Monad.Fix (MonadFix) +import Control.Monad.IO.Class (MonadIO(..)) +import qualified Data.ByteString.Char8 as BC +import qualified Network.Socket as NS +import Reflex +import Reflex.Backend.Socket +import Reflex.Host.Basic +import Reflex.Workflow + +type ClientCxt t m = + ( Reflex t + , MonadFix m + , PerformEvent t m + , PostBuild t m + , TriggerEvent t m + , MonadIO m + , MonadIO (Performable m) + ) + +unconnected :: ClientCxt t m => Workflow t m (Event t ()) +unconnected = Workflow $ do + (eError, eConnected) <- fanEither <$> connect (Just "127.0.0.1") "9000" + eReportedError <- + performEvent $ + (\e -> liftIO $ putStrLn "Error:\n" *> print e) <$> + eError + pure (() <$ eReportedError, connected <$> eConnected) + +connected :: ClientCxt t m => NS.Socket -> Workflow t m (Event t ()) +connected s = Workflow $ mdo + (eLine, onLine) <- newTriggerEvent + void . liftIO . forkIO . forever $ onLine . BC.pack =<< getLine + + so <- socket $ SocketConfig s 2048 eLine eQuit + performEvent_ $ liftIO . BC.putStrLn <$> _sReceive so + + let eQuit = _sClose so + pure (eQuit, unconnected <$ eQuit) + +main :: IO () +main = basicHostWithQuit $ do + deQuit <- workflow unconnected + pure (switchDyn deQuit) diff --git a/example/Others.hs b/example/Others.hs new file mode 100644 --- /dev/null +++ b/example/Others.hs @@ -0,0 +1,97 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} + +{-| +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable +-} + +module Main (main) where + +import Control.Monad.IO.Class (liftIO) +import qualified Data.ByteString.Char8 as BC +import Data.Functor ((<&>), void) +import Data.Maybe (isNothing) +import Data.Witherable (catMaybes) +import qualified Network.Socket as NS +import Reflex +import Reflex.Backend.Socket +import Reflex.Host.Basic (basicHostForever, basicHostWithQuit) +import Reflex.Network (networkHold) +import System.Environment (getArgs) + +-- | Connect to a remote host, and quit as soon as something happens. +connect1 :: IO () +connect1 = basicHostWithQuit $ do + eQuit <- connect (Just "127.0.0.1") "9000" + let (eError, eConnect) = fanEither eQuit + + performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eConnect + performEvent_ $ liftIO . print <$> eError + + pure (void eQuit) + +-- | Listen for connections, and log them as they come in. Does +-- nothing with arriving connections, so will leak FDs. +accept1 :: IO () +accept1 = basicHostForever $ do + (eListenError, eAccept) <- fanEither <$> accept + (AcceptConfig (Just "127.0.0.1") (Just "9000") 1 [(NS.ReuseAddr, 1)] never) + + eNewClient <- switchHold never $ _aAcceptSocket <$> eAccept + eListenClosed <- switchHold never $ _aClose <$> eAccept + eAcceptError <- switchHold never $ _aError <$> eAccept + + performEvent_ $ + (liftIO . putStrLn $ "Error starting listen socket") <$ eListenError + performEvent_ $ (liftIO . putStrLn $ "Listen closed") <$ eListenClosed + performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eNewClient + performEvent_ $ liftIO . print <$> eAcceptError + + pure () + +-- | Connect to a remote host. When the connection succeeds, put the +-- @'Socket' t@ into @dSocket@, send a message, then close. +connect2 :: IO () +connect2 = basicHostWithQuit $ mdo + (eConnError, eConnect) <- fanEither <$> connect (Just "127.0.0.1") "9000" + + performEvent_ $ (liftIO . putStrLn $ "Connected") <$ eConnect + performEvent_ $ liftIO . print <$> eConnError + + dSocket :: Dynamic t (Maybe (Socket t)) <- networkHold (pure Nothing) $ leftmost + [ eConnect <&> \s -> Just <$> socket (SocketConfig s 2048 eTx eOpen) + , pure Nothing <$ eClosed + ] + + eOpen <- switchHold never . fmap _sOpen . catMaybes $ updated dSocket + let eTx = "Hi" <$ eOpen + + eClosed <- switchHold never . fmap _sClose . catMaybes $ updated dSocket + eRx <- switchHold never . fmap _sReceive . catMaybes $ updated dSocket + eSockError <- switchHold never . fmap _sError . catMaybes $ updated dSocket + + performEvent_ $ liftIO . BC.putStrLn <$> eRx + performEvent_ $ liftIO . print <$> eSockError + performEvent_ $ liftIO (putStrLn "Closed") <$ eClosed + + let eQuit = leftmost + [ void eConnError + , void . ffilter isNothing $ updated dSocket + ] + + pure eQuit + +main :: IO () +main = getArgs >>= \case + ["accept1"] -> accept1 + ["connect1"] -> connect1 + ["connect2"] -> connect2 + _ -> pure () diff --git a/example/Server.hs b/example/Server.hs new file mode 100644 --- /dev/null +++ b/example/Server.hs @@ -0,0 +1,124 @@ +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE ScopedTypeVariables #-} + +{-| +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable +-} + +module Main (main) where + +import Control.Lens (_1, _2, _3, view) +import Control.Monad (void) +import Control.Monad.Fix (MonadFix) +import Control.Monad.IO.Class (MonadIO(..)) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as BC +import Data.Char (isSpace) +import Data.Function ((&)) +import Data.Functor ((<&>)) +import Data.Map (Map) +import qualified Data.Map as Map +import qualified Network.Socket as NS +import Reflex +import Reflex.Backend.Socket +import Reflex.Host.Basic +import System.IO.Error + +type ServerCxt t m = + ( Reflex t + , MonadFix m + , MonadSample t m + , PerformEvent t m + , PostBuild t m + , TriggerEvent t m + , MonadIO m + , MonadIO (Performable m) + ) + +perConnection + :: ServerCxt t m + => Dynamic t NS.Socket + -> Event t ByteString + -> m (Event t ByteString, Event t (), Event t ()) +perConnection dSocket eTx = mdo + s <- sample . current $ dSocket + so <- socket $ SocketConfig s 4096 eTx eQuit + + let + eRx = _sReceive so + eRxLine = fst . BC.spanEnd isSpace <$> eRx + eQuit = void $ ffilter (== "quit") eRxLine + eShutdown = void $ ffilter (== "shutdown") eRxLine + eOut = ffilter (`notElem` ["quit", "shutdown"]) eRxLine + + pure (eOut, eShutdown, _sClose so) + +go :: forall t m. BasicGuestConstraints t m => BasicGuest t m (Event t ()) +go = mdo + (eListenError, eAccept) <- fanEither <$> accept + (AcceptConfig (Just "127.0.0.1") (Just "9000") 1 [(NS.ReuseAddr, 1)] eQuit) + + eNewClient <- switchHold never $ fmap fst . _aAcceptSocket <$> eAccept + eListenClose <- switchHold never $ _aClose <$> eAccept + eAcceptError <- switchHold never $ _aError <$> eAccept + + performEvent_ $ eListenError + <&> liftIO . putStrLn . ("Listen error: " <>) . show + performEvent_ $ eAcceptError + <&> liftIO . putStrLn . ("Error accepting: " <>) . show . ioeGetErrorType + + eAdds :: Event t (Integer, NS.Socket) <- numberOccurrences eNewClient + + -- Maintain a map of all open sockets, indexed by their occurrence number. + dSocketMap <- foldDyn ($) Map.empty . mergeWith (.) $ + [ uncurry Map.insert <$> eAdds + , flip (foldr Map.delete) <$> eRemoves + ] + + dmeSocketEvents + :: Dynamic t (Map Integer (Event t ByteString, Event t (), Event t ())) + <- list dSocketMap $ \dSocket -> perConnection dSocket eChats + + let + eChats :: Event t ByteString + eChats = dmeSocketEvents + -- Pick out the "data received" events, and go from "dynamic map of + -- events" to "dynamic of (event of map)" + & fmap (mergeMap . fmap (view _1)) + + -- listen to the latest Event + & switchDyn + + -- Collect all the received data into one big ByteString + & fmap (mconcat . Map.elems) + + eShutdown = dmeSocketEvents + & fmap (mergeMap . fmap (view _2)) + & switchDyn + & void + + eRemoves = dmeSocketEvents + & fmap (mergeMap . fmap (view _3)) + & switchDyn + & fmap Map.keys + + eQuitMap = void . ffilter id . updated $ Map.null <$> dSocketMap + eQuit = leftmost + [ void eListenError -- Couldn't open listen socket + , eQuitMap -- Nobody is connected + , eListenClose -- Our listen socket was closed + , eShutdown -- Client asked us to shutdown everything + ] + pure eQuit + +main :: IO () +main = basicHostWithQuit go diff --git a/reflex-backend-socket.cabal b/reflex-backend-socket.cabal new file mode 100644 --- /dev/null +++ b/reflex-backend-socket.cabal @@ -0,0 +1,95 @@ +-- Initial reflex-backend-socket.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +name: reflex-backend-socket +version: 0.2.0.0 +synopsis: Reflex bindings for TCP sockets +description: + <> + . + reflex-backend-socket provides functions to handle sockets using + Reflex @Event@s. Sending\/receiving\/waiting\/accepting are all + performed on background threads. + . + The most important function in this library is + @Reflex.Backend.Socket.socket@, which wraps a @Socket@ to process + @Event t ByteString@s. + . + That @Socket@ can come from: + . + 1. @Reflex.Backend.Socket.Accept.accept@, if you're making a server; + 2. @Reflex.Backend.Socket.Connect.connect@, if you're making a client; or + 3. Your favourite networking library. +license: BSD3 +license-file: LICENSE +author: Dave Laing +maintainer: dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +copyright: (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +category: Network +build-type: Simple +extra-source-files: ChangeLog.md +cabal-version: >=1.10 +tested-with: GHC == 8.6.5 + +source-repository head + type: git + location: git@github.com/qfpl/reflex-backend-socket.git + +library + exposed-modules: Reflex.Backend.Socket + , Reflex.Backend.Socket.Accept + , Reflex.Backend.Socket.Connect + , Reflex.Backend.Socket.Error + build-depends: base >= 4.12 && < 4.13 + , bytestring >= 0.10 && < 0.11 + , lens >= 4.15.4 && < 4.19 + , mtl >= 2.2 && < 2.3 + , network >= 2.6 && < 3.2 + , reflex >= 0.5 && < 0.7 + , semialign >= 1 && < 1.2 + , semigroupoids >= 5.2.2 && < 5.4 + , stm >= 2.4 && < 2.6 + , these >= 1 && < 1.1 + hs-source-dirs: src + ghc-options: -Wall + default-language: Haskell2010 + +executable example-server + main-is: Server.hs + build-depends: base >= 4.12 && < 4.13 + , bytestring >= 0.10 && < 0.11 + , containers >= 0.5 && < 0.7 + , lens >= 4.15.4 && < 4.19 + , network >= 2.6 && < 3.2 + , reflex >= 0.5 && < 0.7 + , reflex-backend-socket + , reflex-basic-host >= 0.2 && < 0.3 + hs-source-dirs: example + ghc-options: -Wall -threaded + default-language: Haskell2010 + +executable example-client + main-is: Client.hs + build-depends: base >= 4.12 && < 4.13 + , bytestring >= 0.10 && < 0.11 + , network >= 2.6 && < 3.2 + , reflex >= 0.5 && < 0.7 + , reflex-backend-socket + , reflex-basic-host >= 0.2 && < 0.3 + hs-source-dirs: example + ghc-options: -Wall -threaded + default-language: Haskell2010 + + +executable example-others + main-is: Others.hs + build-depends: base >= 4.12 && < 4.13 + , bytestring >= 0.10 && < 0.11 + , network >= 2.6 && < 3.2 + , reflex >= 0.5 && < 0.7 + , reflex-backend-socket + , reflex-basic-host >= 0.2 && < 0.3 + , witherable >= 0.2 && < 0.4 + hs-source-dirs: example + ghc-options: -Wall -threaded + default-language: Haskell2010 diff --git a/src/Reflex/Backend/Socket.hs b/src/Reflex/Backend/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Reflex/Backend/Socket.hs @@ -0,0 +1,210 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeApplications #-} + +{-| +Module : Reflex.Backend.Socket +Description : Wrap a TCP socket to communicate through @Event t ByteString@ +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable + +Use 'socket' to wrap a network 'Socket' so that it sends out the +firings of an @'Event' t 'ByteString'@, and fires any data that it +receives on another @'Event' t 'ByteString'@. +-} + +module Reflex.Backend.Socket + ( socket + + -- * Socket configuration + , SocketConfig(..) + + -- * Socket output events + , Socket(..) + + -- * Lenses + -- ** 'SocketConfig' + , scInitSocket + , scMaxRx + , scSend + , scClose + + -- ** 'Socket' + , sReceive + , sOpen + , sClose + , sError + + -- * Convenience re-exports + , module Reflex.Backend.Socket.Accept + , module Reflex.Backend.Socket.Connect + , module Reflex.Backend.Socket.Error + ) where + +import Control.Concurrent (forkIO) +import qualified Control.Concurrent.STM as STM +import Control.Exception (IOException, try) +import Control.Lens.TH (makeLenses) +import Control.Monad.IO.Class (MonadIO(..)) +import Control.Monad.STM (atomically) +import Data.Align (align) +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import Data.Functor (($>), (<&>), void) +import Data.These +import qualified Network.Socket as NS +import Network.Socket.ByteString (sendAll, recv) +import Reflex +import Reflex.Backend.Socket.Accept +import Reflex.Backend.Socket.Connect +import Reflex.Backend.Socket.Error + +-- | Holds the socket to wire into the FRP network, and events that +-- drive it. +data SocketConfig t = SocketConfig + { _scInitSocket :: NS.Socket + -- ^ Socket to wrap. + , _scMaxRx :: Int + -- ^ Maximum number of bytes to read at a time. + , _scSend :: Event t ByteString + -- ^ Data to send out on this socket. + , _scClose :: Event t () + -- ^ Ask to close the socket. The socket will stop trying to + -- receive data (and the '_sReceive' event will stop firing), and + -- the socket will be "drained": future events on '_scSend' will + -- be ignored, and it will close after writing all pending data. + -- If '_scSend' and '_scClose' fire in the same frame, the data + -- will nevertheless be queued for sending. + } + +$(makeLenses ''SocketConfig) + +-- | Events produced by an active socket. +data Socket t = Socket + { _sReceive :: Event t ByteString + -- ^ Data has arrived. + , _sOpen :: Event t () + -- ^ The socket has opened, and its receive/send loops are running. + , _sClose :: Event t () + -- ^ The socket has closed. This will fire exactly once when the + -- socket closes for any reason, including if your '_scClose' + -- event fires, the other end disconnects, or if the socket closes + -- in response to a caught exception. + , _sError :: Event t IOException + -- ^ An exception occurred. Treat the socket as closed after you + -- see this. If the socket was open, you will see the '_sClose' + -- event fire as well, but not necessarily in the same frame. + } + +$(makeLenses ''Socket) + +data SocketState + = Open + -- ^ Data flows in both directions + | Draining + -- ^ We've been asked to close, but will transmit all pending data + -- first (and not accept any more) + | Closed + -- ^ Hard close. Don't transmit pending data. + +-- | Wire a socket into the FRP network. You will likely use this to +-- attach events to a socket that you just connected (from +-- 'Reflex.Backend.Socket.Connect.connect'), or a socket that you just +-- accepted (from the 'Reflex.Backend.Socket.Accept._aAcceptSocket' +-- event you got when you called +-- 'Reflex.Backend.Socket.Accept.accept'). +socket + :: forall t m. + ( Reflex t + , PerformEvent t m + , PostBuild t m + , TriggerEvent t m + , MonadIO (Performable m) + , MonadIO m + ) + => SocketConfig t + -> m (Socket t) +socket (SocketConfig sock maxRx eTx eClose) = do + (eRx, onRx) <- newTriggerEvent + (eOpen, onOpen) <- newTriggerEvent + (eClosed, onClosed) <- newTriggerEvent + (eError, onError) <- newTriggerEvent + + payloadQueue <- liftIO STM.newTQueueIO + state <- liftIO $ STM.newTVarIO Open + + let + start = liftIO $ do + void $ forkIO txLoop + void $ forkIO rxLoop + void $ forkIO closeSentinel + onOpen () + + where + txLoop = + let + loop = do + mBytes <- atomically $ + STM.readTVar state >>= \case + Closed -> pure Nothing + Draining -> STM.tryReadTQueue payloadQueue + Open -> STM.tryReadTQueue payloadQueue + >>= maybe STM.retry (pure . Just) + + case mBytes of + Nothing -> shutdown + Just bs -> + try (sendAll sock bs) >>= \case + Left exc -> onError exc *> shutdown + Right () -> loop + in loop + + rxLoop = + let + loop = atomically (STM.readTVar state) >>= \case + Open -> try (recv sock maxRx) >>= \case + Left exc -> onError exc *> shutdown + Right bs + | B.null bs -> shutdown + | otherwise -> onRx bs *> loop + _ -> pure () + in loop + + closeSentinel = do + atomically $ STM.readTVar state >>= \case + Closed -> pure () + _ -> STM.retry + + void . try @IOException $ NS.close sock + onClosed () + + shutdown = void . atomically $ STM.writeTVar state Closed + + ePostBuild <- getPostBuild + performEvent_ $ ePostBuild $> start + + -- If we see a tx and a close event in the same frame, we want to + -- process the tx before the close, so it doesn't get lost. + let + eTxOrClose :: Event t (These ByteString ()) + eTxOrClose = align eTx eClose + + queueSend bs = STM.readTVar state >>= \case + Open -> STM.writeTQueue payloadQueue bs + _ -> pure () + + queueClose = STM.modifyTVar state $ \case + Open -> Draining + s -> s + + performEvent_ $ eTxOrClose <&> liftIO . atomically . \case + This bs -> queueSend bs + That () -> queueClose + These bs () -> queueSend bs *> queueClose + + pure $ Socket eRx eOpen eClosed eError diff --git a/src/Reflex/Backend/Socket/Accept.hs b/src/Reflex/Backend/Socket/Accept.hs new file mode 100644 --- /dev/null +++ b/src/Reflex/Backend/Socket/Accept.hs @@ -0,0 +1,174 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} + +{-| +Module : Reflex.Backend.Socket.Accept +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable + +Use 'accept' to create listen sockets, and get an @'Event' t 'Socket'@ +of new connections. +-} + +module Reflex.Backend.Socket.Accept + ( accept + + -- * Listen socket configuration + , AcceptConfig(..) + + -- * Results of @accept@ + , Accept(..) + + -- * Lenses + -- ** 'AcceptConfig' + , acHostname + , acService + , acListenQueue + , acSocketOptions + , acClose + + -- ** 'Accept' + , aAcceptSocket + , aClose + , aError + ) where + +import Control.Concurrent (forkIO) +import qualified Control.Concurrent.STM as STM +import Control.Exception (IOException, onException, try) +import Control.Lens.TH (makeLenses) +import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT) +import Control.Monad.STM (atomically) +import Control.Monad.Trans (MonadIO(..)) +import Data.Foldable (traverse_) +import Data.Functor (($>), void) +import Data.List.NonEmpty (NonEmpty, fromList) +import Data.Semigroup.Foldable (asum1) +import Network.Socket (AddrInfo(..), AddrInfoFlag(..), Socket) +import qualified Network.Socket as NS +import Reflex +import Reflex.Backend.Socket.Error (SetupError(..)) + +-- | Configuration of a listen socket. +data AcceptConfig t = AcceptConfig + { _acHostname :: Maybe NS.HostName + -- ^ The hostname to bind to. This will often be 'Nothing', to + -- bind all interfaces. + , _acService :: Maybe NS.ServiceName + -- ^ The port number or service name to listen on. See the + -- . + , _acListenQueue :: Int + -- ^ The length of the "pending connections" queue. See the + -- . + , _acSocketOptions :: [(NS.SocketOption, Int)] + -- ^ List of socket options, passed one at a time to + -- 'NS.setSocketOption'. Many people will want + -- @[('NS.ReuseAddr', 1)]@ here. + , _acClose :: Event t () + -- ^ Close the listen socket when this event fires. If you plan to + -- run forever, use 'never'. + } + +$(makeLenses ''AcceptConfig) + +-- | Events produced by a running listen socket. +data Accept t = Accept + { _aAcceptSocket :: Event t (Socket, NS.SockAddr) + -- ^ A new connection was accepted, including its remote address. + , _aClose :: Event t () + -- ^ The socket has closed. This will fire exactly once when the + -- socket closes for any reason, including if your '_acClose' + -- event fires or if the socket closes in response to a caught + -- exception. + , _aError :: Event t IOException + -- ^ An exception occurred. Treat the socket as closed after you + -- see this. You will see the '_aClose' event fire as well, but + -- not necessarily in the same frame. + } + +$(makeLenses ''Accept) + +-- | Create a listening socket. Sockets are accepted in a background +-- thread. +accept + :: ( Reflex t + , PerformEvent t m + , PostBuild t m + , TriggerEvent t m + , MonadIO (Performable m) + , MonadIO m + ) + => AcceptConfig t + -> m (Event t (Either SetupError (Accept t))) + -- ^ This event will fire exactly once. +accept (AcceptConfig mHost mService listenQueue options eClose) = do + (eAccept, onAccept) <- newTriggerEvent + (eClosed, onClosed) <- newTriggerEvent + (eError, onError) <- newTriggerEvent + + isOpen <- liftIO STM.newEmptyTMVarIO + + let + start = liftIO $ makeListenSocket >>= \case + Left exc -> pure $ Left exc + Right sock -> do + atomically $ STM.putTMVar isOpen sock + void $ forkIO acceptLoop + pure . Right $ Accept eAccept eClosed eError + + where + makeListenSocket = + let + getAddrs :: ExceptT SetupError IO (NonEmpty AddrInfo) + getAddrs = withExceptT GetAddrInfoError . ExceptT . try $ + -- fromList is OK here, as getaddrinfo(3) is required to + -- return a nonempty list of addrinfos. + -- + -- See: http://pubs.opengroup.org/onlinepubs/9699919799/functions/getaddrinfo.html + -- And: https://github.com/haskell/network/issues/407 + fromList <$> NS.getAddrInfo (Just passiveHints) mHost mService + where passiveHints = NS.defaultHints { addrFlags = [AI_PASSIVE] } + + tryListen + :: AddrInfo + -> ExceptT (NonEmpty (AddrInfo, IOException)) IO Socket + tryListen info = withExceptT (pure . (info,)) . ExceptT . try $ do + sock <- NS.socket (addrFamily info) NS.Stream NS.defaultProtocol + (`onException` NS.close sock) $ do + traverse_ (uncurry $ NS.setSocketOption sock) options + NS.bind sock (addrAddress info) + NS.listen sock listenQueue + pure sock + + in runExceptT $ do + addrs <- getAddrs + let attempts = tryListen <$> addrs + withExceptT UseAddrInfoError $ asum1 attempts + + acceptLoop = + let + -- If we receive an exception when trying to accept, check + -- the TMVar that's meant to hold our socket. If it's + -- empty, then 'eClose' must have fired (and the socket + -- closed under us) and we should go quietly. Otherwise, + -- close the socket ourselves and signal 'eError'. + exHandlerAccept e = atomically (STM.tryReadTMVar isOpen) + >>= maybe (pure ()) (const $ close *> onError e) + in + atomically (STM.tryReadTMVar isOpen) >>= \case + Nothing -> onClosed () + Just sock -> do + try (NS.accept sock) >>= either exHandlerAccept onAccept + acceptLoop + + close = atomically (STM.tryTakeTMVar isOpen) >>= traverse_ NS.close + + performEvent_ $ eClose $> liftIO close + + ePostBuild <- getPostBuild + performEvent $ ePostBuild $> start diff --git a/src/Reflex/Backend/Socket/Connect.hs b/src/Reflex/Backend/Socket/Connect.hs new file mode 100644 --- /dev/null +++ b/src/Reflex/Backend/Socket/Connect.hs @@ -0,0 +1,74 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TupleSections #-} + +{-| +Module : Reflex.Backend.Socket.Connect +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable + +Use 'connect' to attempt a connection to a remote endpoint, and get an +@'Event' t ('Either' 'SetupError' 'Socket')@ that tells you whether or +not it worked. +-} + +module Reflex.Backend.Socket.Connect (connect) where + +import Control.Concurrent (forkIO) +import Control.Exception (IOException, onException, try) +import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT) +import Control.Monad.Trans (MonadIO(..)) +import Data.Functor (($>), void) +import Data.List.NonEmpty (NonEmpty, fromList) +import Data.Semigroup.Foldable (asum1) +import Network.Socket (Socket, AddrInfo(..), defaultProtocol) +import qualified Network.Socket as NS +import Reflex +import Reflex.Backend.Socket.Error (SetupError(..)) + +-- | Connect to a remote endpoint. The connection happens in a +-- background thread. +connect + :: ( Reflex t + , PerformEvent t m + , TriggerEvent t m + , PostBuild t m + , MonadIO (Performable m) + , MonadIO m + ) + => Maybe NS.HostName + -- ^ Host to connect to. If 'Nothing', connect via loopback. + -> NS.ServiceName + -- ^ Service (port number or service name). See the + -- . + -> m (Event t (Either SetupError Socket)) + -- ^ This event will fire exactly once. +connect mHost service = do + ePostBuild <- getPostBuild + performEventAsync $ ePostBuild $> \onRes -> void . liftIO . forkIO $ + let + getAddrs :: ExceptT SetupError IO (NonEmpty AddrInfo) + getAddrs = withExceptT GetAddrInfoError . ExceptT . try $ + -- fromList is OK here, as getaddrinfo(3) is required to + -- return a nonempty list of addrinfos. + -- + -- See: http://pubs.opengroup.org/onlinepubs/9699919799/functions/getaddrinfo.html + -- And: https://github.com/haskell/network/issues/407 + fromList <$> NS.getAddrInfo Nothing mHost (Just service) + + tryConnect + :: AddrInfo + -> ExceptT (NonEmpty (AddrInfo, IOException)) IO Socket + tryConnect info = withExceptT (pure . (info,)) . ExceptT . try $ do + sock <- NS.socket (addrFamily info) NS.Stream defaultProtocol + NS.connect sock (addrAddress info) `onException` NS.close sock + pure sock + + in do + res <- runExceptT $ do + addrs <- getAddrs + let attempts = tryConnect <$> addrs + withExceptT UseAddrInfoError $ asum1 attempts + onRes res diff --git a/src/Reflex/Backend/Socket/Error.hs b/src/Reflex/Backend/Socket/Error.hs new file mode 100644 --- /dev/null +++ b/src/Reflex/Backend/Socket/Error.hs @@ -0,0 +1,45 @@ +{-| +Module : Reflex.Backend.Socket.Error +Copyright : (c) 2018-2019, Commonwealth Scientific and Industrial Research Organisation +License : BSD3 +Maintainer : dave.laing.80@gmail.com, jack.kelly@data61.csiro.au +Stability : experimental +Portability : non-portable +-} + +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE TemplateHaskell #-} + +module Reflex.Backend.Socket.Error + ( SetupError(..) + + -- * Prisms + -- ** 'SetupError' + , _GetAddrInfoError + , _UseAddrInfoError + ) where + +import Control.Exception (IOException) +import Data.List.NonEmpty (NonEmpty) +import GHC.Generics (Generic) +import Network.Socket (AddrInfo) +import Control.Lens.TH (makePrisms) + +-- | If a "socket setup" fails ('Reflex.Backend.Socket.Accept.accept' +-- or 'Reflex.Backend.Socket.Connect.connect'), you'll inspect one of +-- these to find out why. +data SetupError + = GetAddrInfoError IOException + -- ^ Call to 'Network.Socket.getAddrInfo' failed. + | UseAddrInfoError (NonEmpty (AddrInfo, IOException)) + -- ^ We failed to set up a socket with any 'AddrInfo' we were + -- given, and here are the corresponding exceptions each time we + -- tried. For 'Reflex.Backend.Socket.Accept.accept', this means + -- either 'Network.Socket.bind' or 'Network.Socket.listen' failed, + -- and for 'Reflex.Backend.Socket.Connect.connect', this means + -- 'Network.Socket.Connect' failed. + deriving (Eq, Generic, Show) + +$(makePrisms ''SetupError)